From ecf2442420f13c3335319144286cf968ffff2641 Mon Sep 17 00:00:00 2001 From: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> Date: Tue, 30 Jun 2026 11:25:33 -0700 Subject: [PATCH 1/5] feat(discord): opt-in raw-reaction journal for durable triage state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When discord.reaction_journal (config) is set, request the non-privileged reactions intent and subscribe to on_raw_reaction_add/remove, appending one JSON line per raw transition in the reaction_state core's journal schema {channel_id, message_id, emoji, user_id, action, seq, ts}. Raw events fire regardless of message cache, so a reaction on an old/un-cached card is still captured — the durable-triage use case (greenhouse seed_triage) the REST poller approximates. Default OFF: no intent change, no extra gateway traffic, no behavior change for anyone who hasn't set the config key. Writes are best-effort (a journal error can never crash the gateway event loop); the seq counter seeds from the existing journal max so a restart never rewinds it. --- plugins/platforms/discord/adapter.py | 81 ++++++++++++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 5379c7cd643a..5ed961b4f931 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -976,6 +976,15 @@ async def connect(self, *, is_reconnect: bool = False) -> bool: ) intents.voice_states = True + # Reaction journal (opt-in): only request the (non-privileged) + # reactions intent when a journal path is configured. This lets a + # downstream consumer durably capture raw reaction transitions + # (reaction_state / seed_triage). Default off → no behavior change + # and no extra gateway traffic for anyone who hasn't opted in. + self._reaction_journal_path = self.config.extra.get("reaction_journal") or None + if self._reaction_journal_path: + intents.reactions = True + # Resolve proxy (DISCORD_PROXY > generic env vars > macOS system proxy) from gateway.platforms.base import resolve_proxy_url, proxy_kwargs_for_bot proxy_url = resolve_proxy_url(platform_env_var="DISCORD_PROXY") @@ -1181,6 +1190,20 @@ async def on_voice_state_update(member, before, after): guild_id, ) + # Reaction journal (opt-in via discord.reaction_journal): append one + # JSON line per RAW reaction transition. Raw events fire regardless of + # message cache, so a reaction on an old/un-cached card is still + # captured. Schema matches the reaction_state core's journal contract: + # {channel_id, message_id, emoji, user_id, action, seq, ts}. + if self._reaction_journal_path: + @self._client.event + async def on_raw_reaction_add(payload): + adapter_self._append_reaction_journal(payload, "add") + + @self._client.event + async def on_raw_reaction_remove(payload): + adapter_self._append_reaction_journal(payload, "remove") + # Register slash commands if self._slash_commands: self._register_slash_commands() @@ -1813,6 +1836,64 @@ def _reactions_enabled(self) -> bool: """Check if message reactions are enabled via config/env.""" return os.getenv("DISCORD_REACTIONS", "true").lower() not in {"false", "0", "no"} + def _next_reaction_seq(self) -> int: + """Monotonic per-key-safe sequence for journal events. Seeded once from + the existing journal's last seq so a gateway restart never rewinds the + counter (which would make the core reject post-restart events as stale).""" + seq = getattr(self, "_reaction_seq", None) + if seq is None: + seq = 0 + path = getattr(self, "_reaction_journal_path", None) + try: + if path and os.path.exists(path): + with open(path, "r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + s = int(json.loads(line).get("seq", 0)) + if s > seq: + seq = s + except (ValueError, TypeError): + continue + except OSError: + seq = 0 + seq += 1 + self._reaction_seq = seq + return seq + + def _append_reaction_journal(self, payload: Any, action: str) -> None: + """Append one raw reaction transition to the configured journal, in the + reaction_state core's schema. Best-effort: a journal hiccup must never + crash the gateway's event loop (this runs inside a discord.py handler).""" + path = getattr(self, "_reaction_journal_path", None) + if not path: + return + try: + emoji = getattr(payload, "emoji", None) + # Standard emoji -> unicode char; custom -> "name:id" (matches the + # canonical Discord form the core stores verbatim). + emoji_id = getattr(emoji, "id", None) + if emoji_id: + emoji_str = f"{getattr(emoji, 'name', '')}:{emoji_id}" + else: + emoji_str = str(getattr(emoji, "name", emoji) or "") + event = { + "channel_id": str(getattr(payload, "channel_id", "")), + "message_id": str(getattr(payload, "message_id", "")), + "emoji": emoji_str, + "user_id": str(getattr(payload, "user_id", "")), + "action": action, + "seq": self._next_reaction_seq(), + "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + } + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + with open(path, "a", encoding="utf-8") as fh: + fh.write(json.dumps(event, ensure_ascii=False) + "\n") + except Exception as e: # noqa: BLE001 - never let a journal error kill the loop + logger.debug("[%s] reaction-journal append soft-fail: %s", self.name, e) + async def on_processing_start(self, event: MessageEvent) -> None: """Add an in-progress reaction for normal Discord message events.""" if not self._reactions_enabled(): From 444c3e9e5c4c49255ecdc72a33a9c8d87752fb54 Mon Sep 17 00:00:00 2001 From: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> Date: Tue, 30 Jun 2026 11:27:13 -0700 Subject: [PATCH 2/5] test(discord): reaction-journal schema + restart-safe seq + core-ingest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Covers the journal contract (exact reaction_state schema), custom-emoji name:id form, add→remove ordering, monotonic seq resuming above the existing journal max (restart-safety), no-op when unconfigured, best-effort never-raises on a bad payload, and an end-to-end ingest by the reaction_state core proving byte-compatibility (skips cleanly if the core isn't in this checkout). --- .../gateway/test_discord_reaction_journal.py | 141 ++++++++++++++++++ 1 file changed, 141 insertions(+) create mode 100644 tests/gateway/test_discord_reaction_journal.py diff --git a/tests/gateway/test_discord_reaction_journal.py b/tests/gateway/test_discord_reaction_journal.py new file mode 100644 index 000000000000..052a7377712c --- /dev/null +++ b/tests/gateway/test_discord_reaction_journal.py @@ -0,0 +1,141 @@ +"""Tests for the opt-in Discord raw-reaction journal (discord.reaction_journal). + +The journal lets a downstream consumer (greenhouse seed_triage / the +reaction_state durable-state core) capture raw reaction transitions that fire +regardless of message cache. These tests exercise the two journal methods +directly against a lightweight stub adapter — no live gateway, no network — and +assert the on-disk schema is exactly what the reaction_state core ingests. +""" + +from __future__ import annotations + +import json +import types + +from plugins.platforms.discord.adapter import DiscordAdapter + + +def _stub(journal_path): + """A bare object carrying just what the journal methods touch, with the two + real DiscordAdapter methods bound to it (avoids the full adapter ctor).""" + s = types.SimpleNamespace() + s.name = "discord" + s._reaction_journal_path = str(journal_path) + s._next_reaction_seq = types.MethodType(DiscordAdapter._next_reaction_seq, s) + s._append_reaction_journal = types.MethodType( + DiscordAdapter._append_reaction_journal, s) + return s + + +class _Emoji: + def __init__(self, name, id=None): + self.name = name + self.id = id + + +class _Payload: + def __init__(self, channel_id, message_id, user_id, emoji): + self.channel_id = channel_id + self.message_id = message_id + self.user_id = user_id + self.emoji = emoji + + +def _lines(path): + return [json.loads(l) for l in path.read_text(encoding="utf-8").splitlines() if l.strip()] + + +def test_append_writes_core_schema(tmp_path): + journal = tmp_path / "reactions.jsonl" + s = _stub(journal) + s._append_reaction_journal(_Payload("C1", "M1", "U1", _Emoji("✅")), "add") + rows = _lines(journal) + assert len(rows) == 1 + r = rows[0] + # exactly the reaction_state core journal contract + assert set(r) == {"channel_id", "message_id", "emoji", "user_id", "action", "seq", "ts"} + assert r["channel_id"] == "C1" and r["message_id"] == "M1" + assert r["emoji"] == "✅" and r["user_id"] == "U1" + assert r["action"] == "add" and isinstance(r["seq"], int) + + +def test_custom_emoji_serialized_as_name_id(tmp_path): + journal = tmp_path / "reactions.jsonl" + s = _stub(journal) + s._append_reaction_journal(_Payload("C1", "M1", "U1", _Emoji("partyparrot", 12345)), "add") + assert _lines(journal)[0]["emoji"] == "partyparrot:12345" + + +def test_add_then_remove_recorded_in_order(tmp_path): + journal = tmp_path / "reactions.jsonl" + s = _stub(journal) + s._append_reaction_journal(_Payload("C1", "M1", "U1", _Emoji("✅")), "add") + s._append_reaction_journal(_Payload("C1", "M1", "U1", _Emoji("✅")), "remove") + rows = _lines(journal) + assert [r["action"] for r in rows] == ["add", "remove"] + assert rows[1]["seq"] > rows[0]["seq"] # monotonic + + +def test_seq_resumes_above_existing_journal_max(tmp_path): + # A gateway restart must NOT rewind the seq: a fresh adapter seeds its counter + # from the existing journal's max seq, so post-restart events out-rank prior + # ones (else the core would reject them as stale). + journal = tmp_path / "reactions.jsonl" + # simulate a pre-restart journal whose max seq is 7 (out-of-order on purpose) + journal.write_text("\n".join(json.dumps({"seq": s}) for s in (1, 2, 7, 3)) + "\n", + encoding="utf-8") + s = _stub(journal) + assert s._next_reaction_seq() == 8 + assert s._next_reaction_seq() == 9 + + +def test_no_path_is_noop(tmp_path): + journal = tmp_path / "reactions.jsonl" + s = types.SimpleNamespace(name="discord", _reaction_journal_path=None) + s._append_reaction_journal = types.MethodType( + DiscordAdapter._append_reaction_journal, s) + # must not raise and must not create the journal when no path is configured + s._append_reaction_journal(_Payload("C1", "M1", "U1", _Emoji("✅")), "add") + assert not journal.exists() + + +def test_append_never_raises_on_bad_payload(tmp_path): + # Best-effort: a malformed payload must be swallowed (runs inside the gateway + # event loop; a raise would crash it). No emoji attr at all. + journal = tmp_path / "reactions.jsonl" + s = _stub(journal) + broken = types.SimpleNamespace() # missing every field + s._append_reaction_journal(broken, "add") # must not raise + + +def test_journal_is_ingestible_by_reaction_state_core(tmp_path): + """The whole point: the gateway journal must be byte-compatible with the + reaction_state core's read_journal/replay_journal. Skips cleanly if the core + isn't present in this checkout (it lives in the greenhouse-tools repo).""" + import importlib.util + from importlib.machinery import SourceFileLoader + import sys + core_path = (tmp_path.home() / ".hermes" / "greenhouse" / "deployed" + / "reaction-state" / "current") + if not core_path.exists(): + import pytest + pytest.skip("reaction_state core not present in this checkout") + # the core file has no .py extension; load it explicitly via SourceFileLoader + loader = SourceFileLoader("reaction_state", str(core_path)) + spec = importlib.util.spec_from_loader("reaction_state", loader) + rs = importlib.util.module_from_spec(spec) + sys.modules["reaction_state"] = rs + loader.exec_module(rs) + + journal = tmp_path / "reactions.jsonl" + s = _stub(journal) + s._append_reaction_journal(_Payload("C1", "M1", "ACE", _Emoji("✅")), "add") + s._append_reaction_journal(_Payload("C1", "M1", "ACE", _Emoji("✅")), "remove") + s._append_reaction_journal(_Payload("C1", "M2", "ACE", _Emoji("✅")), "add") + + events = rs.read_journal(str(journal)) + conn = rs.connect(":memory:") + rs.replay_journal(conn, events) + present = rs.current_present(conn) + assert ("C1", "M2", "✅", "ACE") in present # M2 add stands + assert ("C1", "M1", "✅", "ACE") not in present # M1 add+remove → absent From 1072992f7d92997b92f93c632d61b74f42141a57 Mon Sep 17 00:00:00 2001 From: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> Date: Tue, 30 Jun 2026 11:35:40 -0700 Subject: [PATCH 3/5] fix(discord): address Greptile review on reaction journal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P1: non-dict JSON line in the journal raised AttributeError out of seed scanning, leaving _reaction_seq unset → every subsequent append retried the same bad line and the journal went permanently dark. Now skip non-dict lines (isinstance guard) and catch AttributeError. RED-proven regression test added. - P2: makedirs ran on every append; now guarded by a once-per-process flag. - P2: sync file I/O ran inside the async on_raw_reaction handlers; now offloaded via run_in_executor (_emit_reaction_journal) so an emoji-storm can't stall the gateway event loop. --- plugins/platforms/discord/adapter.py | 29 +++++++++++++++---- .../gateway/test_discord_reaction_journal.py | 19 ++++++++++++ 2 files changed, 43 insertions(+), 5 deletions(-) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 5ed961b4f931..485039e2cc43 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -1198,11 +1198,11 @@ async def on_voice_state_update(member, before, after): if self._reaction_journal_path: @self._client.event async def on_raw_reaction_add(payload): - adapter_self._append_reaction_journal(payload, "add") + await adapter_self._emit_reaction_journal(payload, "add") @self._client.event async def on_raw_reaction_remove(payload): - adapter_self._append_reaction_journal(payload, "remove") + await adapter_self._emit_reaction_journal(payload, "remove") # Register slash commands if self._slash_commands: @@ -1852,10 +1852,16 @@ def _next_reaction_seq(self) -> int: if not line: continue try: - s = int(json.loads(line).get("seq", 0)) + obj = json.loads(line) + if not isinstance(obj, dict): + continue # a non-dict line carries no seq + s = int(obj.get("seq", 0)) if s > seq: seq = s - except (ValueError, TypeError): + except (ValueError, TypeError, AttributeError): + # malformed line / non-int seq — skip it, never let + # one bad line abort seeding (which would leave + # _reaction_seq unset and silence the journal). continue except OSError: seq = 0 @@ -1863,6 +1869,16 @@ def _next_reaction_seq(self) -> int: self._reaction_seq = seq return seq + async def _emit_reaction_journal(self, payload: Any, action: str) -> None: + """Async wrapper: offload the blocking journal write to a thread so a + burst of reaction events can never stall the gateway's asyncio loop.""" + try: + loop = asyncio.get_running_loop() + await loop.run_in_executor( + None, self._append_reaction_journal, payload, action) + except Exception as e: # noqa: BLE001 - never let it bubble into the loop + logger.debug("[%s] reaction-journal emit soft-fail: %s", self.name, e) + def _append_reaction_journal(self, payload: Any, action: str) -> None: """Append one raw reaction transition to the configured journal, in the reaction_state core's schema. Best-effort: a journal hiccup must never @@ -1888,7 +1904,10 @@ def _append_reaction_journal(self, payload: Any, action: str) -> None: "seq": self._next_reaction_seq(), "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } - os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + # makedirs once per process (the dir persists after the first write). + if not getattr(self, "_reaction_journal_dir_created", False): + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + self._reaction_journal_dir_created = True with open(path, "a", encoding="utf-8") as fh: fh.write(json.dumps(event, ensure_ascii=False) + "\n") except Exception as e: # noqa: BLE001 - never let a journal error kill the loop diff --git a/tests/gateway/test_discord_reaction_journal.py b/tests/gateway/test_discord_reaction_journal.py index 052a7377712c..a000bb1a54c5 100644 --- a/tests/gateway/test_discord_reaction_journal.py +++ b/tests/gateway/test_discord_reaction_journal.py @@ -89,6 +89,25 @@ def test_seq_resumes_above_existing_journal_max(tmp_path): assert s._next_reaction_seq() == 9 +def test_seq_seeding_survives_non_dict_lines(tmp_path): + # P1 regression: a valid-JSON-but-non-dict line (bare string/number/array) must + # NOT raise AttributeError out of seeding — which would leave _reaction_seq unset + # and permanently silence the journal for the rest of the process. The bad lines + # are skipped; the real max seq still wins. + journal = tmp_path / "reactions.jsonl" + journal.write_text( + json.dumps("just a string") + "\n" # non-dict + + json.dumps([1, 2, 3]) + "\n" # non-dict + + "42\n" # bare number + + json.dumps({"seq": 5}) + "\n", # the real one + encoding="utf-8") + s = _stub(journal) + assert s._next_reaction_seq() == 6 # 5+1, not crashed, not reset to 1 + # and a subsequent append actually writes (journal not silenced) + s._append_reaction_journal(_Payload("C1", "M1", "U1", _Emoji("✅")), "add") + assert _lines(journal)[-1]["action"] == "add" + + def test_no_path_is_noop(tmp_path): journal = tmp_path / "reactions.jsonl" s = types.SimpleNamespace(name="discord", _reaction_journal_path=None) From 7bc4275e69946054b30fb34cbeb5c092c0c116bb Mon Sep 17 00:00:00 2001 From: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> Date: Tue, 30 Jun 2026 11:44:01 -0700 Subject: [PATCH 4/5] fix(discord): make reaction_journal env-driven like the rest of the adapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ground-truthed the config path: the Discord adapter is env-driven by convention (_apply_yaml_config translates discord.* YAML keys to DISCORD_* env vars; it seeds nothing into PlatformConfig.extra). Reading config.extra.get('reaction_journal') would therefore have been silently dead — the config key never reaches the adapter. Fix: - read DISCORD_REACTION_JOURNAL env (fallback to extra for nested platforms.discord.extra users), - add the config.yaml discord.reaction_journal -> DISCORD_REACTION_JOURNAL bridge in _apply_yaml_config (env takes precedence, matching every other key), - tests for the bridge + precedence; make the cross-repo core-ingest test path-agnostic (glob for tools/reaction_state.py) since the deployed layout moved to versioned symlink dirs. --- plugins/platforms/discord/adapter.py | 15 +++++++- .../gateway/test_discord_reaction_journal.py | 36 ++++++++++++++++--- 2 files changed, 46 insertions(+), 5 deletions(-) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 485039e2cc43..0662af811c97 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -981,7 +981,14 @@ async def connect(self, *, is_reconnect: bool = False) -> bool: # downstream consumer durably capture raw reaction transitions # (reaction_state / seed_triage). Default off → no behavior change # and no extra gateway traffic for anyone who hasn't opted in. - self._reaction_journal_path = self.config.extra.get("reaction_journal") or None + # Env-driven like the rest of this adapter: config.yaml + # ``discord.reaction_journal`` is translated to DISCORD_REACTION_JOURNAL + # by _apply_yaml_config (the apply_yaml_config_fn hook). + self._reaction_journal_path = ( + os.getenv("DISCORD_REACTION_JOURNAL") + or self.config.extra.get("reaction_journal") + or None + ) if self._reaction_journal_path: intents.reactions = True @@ -7418,6 +7425,12 @@ def _apply_yaml_config(yaml_cfg: dict, discord_cfg: dict) -> dict | None: if isinstance(ic, list): ic = ",".join(str(v) for v in ic) os.environ["DISCORD_IGNORED_CHANNELS"] = str(ic) + # reaction_journal: opt-in path to append raw reaction transitions to, in the + # reaction_state core's journal schema (durable triage state). Empty/unset = + # feature off (no reactions intent requested). + rj = discord_cfg.get("reaction_journal") + if rj is not None and not os.getenv("DISCORD_REACTION_JOURNAL"): + os.environ["DISCORD_REACTION_JOURNAL"] = str(rj) # allowed_channels: if set, bot ONLY responds in these channels (whitelist) ac = discord_cfg.get("allowed_channels") if ac is not None and not os.getenv("DISCORD_ALLOWED_CHANNELS"): diff --git a/tests/gateway/test_discord_reaction_journal.py b/tests/gateway/test_discord_reaction_journal.py index a000bb1a54c5..0abd6d117bc2 100644 --- a/tests/gateway/test_discord_reaction_journal.py +++ b/tests/gateway/test_discord_reaction_journal.py @@ -15,6 +15,28 @@ from plugins.platforms.discord.adapter import DiscordAdapter +def test_yaml_config_bridges_reaction_journal_to_env(monkeypatch): + # config.yaml `discord.reaction_journal: ` must reach the adapter, which + # reads it from the DISCORD_REACTION_JOURNAL env var (the adapter is env-driven + # by convention). Without this bridge the feature is silently dead. + from plugins.platforms.discord.adapter import _apply_yaml_config + monkeypatch.delenv("DISCORD_REACTION_JOURNAL", raising=False) + yaml_cfg = {"discord": {"reaction_journal": "/tmp/r.jsonl"}} + _apply_yaml_config(yaml_cfg, yaml_cfg["discord"]) + import os + assert os.environ.get("DISCORD_REACTION_JOURNAL") == "/tmp/r.jsonl" + + +def test_yaml_config_env_takes_precedence(monkeypatch): + # An explicit env var must survive a config.yaml value (matches every other + # discord key's precedence contract). + import os + monkeypatch.setenv("DISCORD_REACTION_JOURNAL", "/preset") + from plugins.platforms.discord.adapter import _apply_yaml_config + _apply_yaml_config({"discord": {"reaction_journal": "/yaml"}}, {"reaction_journal": "/yaml"}) + assert os.environ["DISCORD_REACTION_JOURNAL"] == "/preset" + + def _stub(journal_path): """A bare object carrying just what the journal methods touch, with the two real DiscordAdapter methods bound to it (avoids the full adapter ctor).""" @@ -134,12 +156,18 @@ def test_journal_is_ingestible_by_reaction_state_core(tmp_path): import importlib.util from importlib.machinery import SourceFileLoader import sys - core_path = (tmp_path.home() / ".hermes" / "greenhouse" / "deployed" - / "reaction-state" / "current") - if not core_path.exists(): + # The core lives in the greenhouse-tools repo, not this checkout. Find any + # real reaction_state.py (worktree or deployed version dir); skip if absent. + gh = tmp_path.home() / ".hermes" / "greenhouse" + core_path = None + if gh.exists(): + for cand in sorted(gh.glob("**/tools/reaction_state.py")): + if cand.is_file(): + core_path = cand + break + if core_path is None: import pytest pytest.skip("reaction_state core not present in this checkout") - # the core file has no .py extension; load it explicitly via SourceFileLoader loader = SourceFileLoader("reaction_state", str(core_path)) spec = importlib.util.spec_from_loader("reaction_state", loader) rs = importlib.util.module_from_spec(spec) From 9e9e14754bcb165ace012cb0738d16ce3c2e219b Mon Sep 17 00:00:00 2001 From: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> Date: Tue, 30 Jun 2026 11:46:22 -0700 Subject: [PATCH 5/5] fix(discord): lock reaction-seq generator against thread-pool race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Greptile P-level: _next_reaction_seq is now invoked from run_in_executor (the async-offload fix), so its read-increment-write on self._reaction_seq raced — two concurrent burst events could mint the SAME seq, and the reaction_state core drops a duplicate/stale seq → silent event loss. Guard the read-increment-write with a per-instance threading.Lock (lazily created under a module-level init lock so the lazy-init is itself race-free). Added a 16-thread/50-iter concurrency test asserting all seqs are unique + a clean monotonic run (RED-proven: the lock-free version mints ~750/800 duplicates). --- plugins/platforms/discord/adapter.py | 75 ++++++++++++------- .../gateway/test_discord_reaction_journal.py | 28 +++++++ 2 files changed, 75 insertions(+), 28 deletions(-) diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py index 0662af811c97..504bc3fabd62 100644 --- a/plugins/platforms/discord/adapter.py +++ b/plugins/platforms/discord/adapter.py @@ -26,6 +26,10 @@ logger = logging.getLogger(__name__) +# Module-level guard for lazily creating each adapter's per-instance reaction-seq +# lock (the lazy-init itself must be race-free across the journal thread pool). +_REACTION_SEQ_INIT_LOCK = threading.Lock() + class _Snowflake: """Minimal object exposing ``.id`` — satisfies discord.py's Snowflake @@ -1846,35 +1850,50 @@ def _reactions_enabled(self) -> bool: def _next_reaction_seq(self) -> int: """Monotonic per-key-safe sequence for journal events. Seeded once from the existing journal's last seq so a gateway restart never rewinds the - counter (which would make the core reject post-restart events as stale).""" - seq = getattr(self, "_reaction_seq", None) - if seq is None: - seq = 0 - path = getattr(self, "_reaction_journal_path", None) - try: - if path and os.path.exists(path): - with open(path, "r", encoding="utf-8") as fh: - for line in fh: - line = line.strip() - if not line: - continue - try: - obj = json.loads(line) - if not isinstance(obj, dict): - continue # a non-dict line carries no seq - s = int(obj.get("seq", 0)) - if s > seq: - seq = s - except (ValueError, TypeError, AttributeError): - # malformed line / non-int seq — skip it, never let - # one bad line abort seeding (which would leave - # _reaction_seq unset and silence the journal). - continue - except OSError: + counter (which would make the core reject post-restart events as stale). + + Thread-safe: called from a thread pool (via run_in_executor), so the + read-increment-write is guarded by a lock — otherwise two concurrent + burst events could mint the SAME seq and the core would drop one as a + duplicate/stale (silent loss).""" + lock = getattr(self, "_reaction_seq_lock", None) + if lock is None: + # First-call init is itself racy; create the lock under a class-level + # guard so all threads converge on one lock instance. + with _REACTION_SEQ_INIT_LOCK: + lock = getattr(self, "_reaction_seq_lock", None) + if lock is None: + lock = threading.Lock() + self._reaction_seq_lock = lock + with lock: + seq = getattr(self, "_reaction_seq", None) + if seq is None: seq = 0 - seq += 1 - self._reaction_seq = seq - return seq + path = getattr(self, "_reaction_journal_path", None) + try: + if path and os.path.exists(path): + with open(path, "r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + obj = json.loads(line) + if not isinstance(obj, dict): + continue # a non-dict line carries no seq + s = int(obj.get("seq", 0)) + if s > seq: + seq = s + except (ValueError, TypeError, AttributeError): + # malformed line / non-int seq — skip it, never + # let one bad line abort seeding (which would + # leave _reaction_seq unset, silencing writes). + continue + except OSError: + seq = 0 + seq += 1 + self._reaction_seq = seq + return seq async def _emit_reaction_journal(self, payload: Any, action: str) -> None: """Async wrapper: offload the blocking journal write to a thread so a diff --git a/tests/gateway/test_discord_reaction_journal.py b/tests/gateway/test_discord_reaction_journal.py index 0abd6d117bc2..fd25b7f2cae8 100644 --- a/tests/gateway/test_discord_reaction_journal.py +++ b/tests/gateway/test_discord_reaction_journal.py @@ -130,6 +130,34 @@ def test_seq_seeding_survives_non_dict_lines(tmp_path): assert _lines(journal)[-1]["action"] == "add" +def test_concurrent_seq_calls_are_unique(tmp_path): + # The seq generator is called from a thread pool (run_in_executor). Two + # concurrent burst events must NEVER mint the same seq, or the core drops one + # as a duplicate/stale → silent loss. Hammer it from many threads and assert + # every returned seq is distinct. + import threading as _t + journal = tmp_path / "reactions.jsonl" + s = _stub(journal) + out = [] + out_lock = _t.Lock() + barrier = _t.Barrier(16) + + def worker(): + barrier.wait() # maximize overlap on the read-increment-write + for _ in range(50): + v = s._next_reaction_seq() + with out_lock: + out.append(v) + + threads = [_t.Thread(target=worker) for _ in range(16)] + for th in threads: + th.start() + for th in threads: + th.join() + assert len(out) == len(set(out)), "duplicate seq minted under concurrency" + assert sorted(out) == list(range(1, len(out) + 1)), "seqs not a clean monotonic run" + + def test_no_path_is_noop(tmp_path): journal = tmp_path / "reactions.jsonl" s = types.SimpleNamespace(name="discord", _reaction_journal_path=None)