feat(discord): opt-in raw-reaction journal for durable triage state

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
@@ -976,6 +980,22 @@ 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.
+            # 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
+
             # 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 +1201,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):
+                    await adapter_self._emit_reaction_journal(payload, "add")
+
+                @self._client.event
+                async def on_raw_reaction_remove(payload):
+                    await adapter_self._emit_reaction_journal(payload, "remove")
+
             # Register slash commands
             if self._slash_commands:
                 self._register_slash_commands()
@@ -1813,6 +1847,98 @@ 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).
+
+        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
+                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
+        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
+        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()),
+            }
+            # 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
+            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():
@@ -7318,6 +7444,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"):

tests/gateway/test_discord_reaction_journal.py

@@ -0,0 +1,216 @@
+"""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 test_yaml_config_bridges_reaction_journal_to_env(monkeypatch):
+    # config.yaml `discord.reaction_journal: <path>` 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)."""
+    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_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_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)
+    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
+    # 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")
+    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