diff --git a/pyproject.toml b/pyproject.toml
index fda72b0..44ca719 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -55,6 +55,7 @@ asyncio_mode = "auto"
 norecursedirs = ["tests/integration/eval"]
 markers = [
     "v2_only: tests for the lifecycle v2 refactor (will fail until migration is complete)",
+    "unverified: e2e intent specs not yet reconciled against the rig; non-blocking (xfail) until verified before the next beta tag",
 ]
 
 [tool.ruff]
diff --git a/stacklets/core/bot-runner/microbot.py b/stacklets/core/bot-runner/microbot.py
index 80076c0..c9647cc 100644
--- a/stacklets/core/bot-runner/microbot.py
+++ b/stacklets/core/bot-runner/microbot.py
@@ -54,6 +54,7 @@
 import json
 import time
 from pathlib import Path
+from urllib.parse import quote
 
 import aiohttp
 import markdown
@@ -78,6 +79,12 @@
 # attached to the specific message instead of a separate timeline event.
 EYES = "\U0001F440"
 
+# Terminal outcome glyphs added alongside the 👀 when work finishes: ✅
+# on success, ❌ on failure. They give an at-a-glance result in the main
+# timeline, since the detailed filing reply now lives in a thread.
+CHECK = "\U00002705"
+CROSS = "\U0000274C"
+
 
 class MicroBot:
     """Base class for lightweight Matrix bots.
@@ -1065,7 +1072,7 @@ def _is_bot_mentioned(self, event) -> bool:
         body = getattr(event, "body", "") or ""
         return self.user_id in body
 
-    def _should_react(self, ctx: RoomContext, *, mentioned: bool) -> bool:
+    async def _should_react(self, ctx: RoomContext, *, mentioned: bool) -> bool:
         """Decide whether the bot acts on the current event at all.
 
         Two cases are never gated — the bot always reacts:
@@ -1078,36 +1085,173 @@ def _should_react(self, ctx: RoomContext, *, mentioned: bool) -> bool:
             message to be aimed at.
 
         Everything else (group rooms with 3+ members, no mention) is
-        subject to ``_room_mode_allows_react`` — the single seam a
-        subclass overrides when it wants per-room mode config. The
-        framework default lets every event through.
+        subject to ``_room_mode_allows_react``, which reads the room's
+        config. Async because that read hits the room state.
         """
         if mentioned:
             return True
         if ctx.is_dm:
             return True
-        return self._room_mode_allows_react(ctx)
+        return await self._room_mode_allows_react(ctx)
 
-    def _room_mode_allows_react(self, ctx: RoomContext) -> bool:
+    async def _room_mode_allows_react(self, ctx: RoomContext) -> bool:
         """The configurable branch of ``_should_react``.
 
-        Group-room behavior is the only thing rooms might want to gate.
-        Anticipated shape — once a subclass / config lands:
+        Reads the room's ``process`` mode from its config state event:
+
+          * ``react`` — the bot ignores plain messages; the only trigger
+            is an explicit user reaction (e.g. 🔖 to save). Returns False.
+          * ``auto`` / unset — the bot processes messages as they arrive
+            (the default). Returns True.
+
+        Set per room with ``!config process auto|react``.
+        """
+        cfg = await self.get_room_config(ctx.room_id)
+        return cfg.get("process") != "react"
+
+    # ── Per-room config (the bot's room account data) ────────────────────
+    #
+    # Per-room bot settings live in the bot's *room account data*, not a
+    # room state event. Writing room state needs power level 50 (a room
+    # moderator), which a bot invited to a family room does not have, so a
+    # state-event write is silently rejected. Account data is the bot's
+    # own: writable regardless of power level, server-stored (survives
+    # restarts), and read fresh — no local cache. nio has no high-level
+    # account-data setter, so we hit the REST endpoint directly with the
+    # bot's token. The friendly `!config` command is the write interface.
+
+    ROOM_CONFIG_TYPE = "dev.famstack.room"
+
+    # Declarative registry of room-config options. One entry drives
+    # validation, the `!config` status view, and the set acknowledgement,
+    # so they never drift. `default` is the value assumed when the option
+    # is unset; `describe` maps each allowed value to a one-line meaning.
+    # Subclasses extend this dict to add their own options.
+    CONFIG_OPTIONS: dict[str, dict] = {
+        "process": {
+            "default": "auto",
+            "describe": {
+                "auto": "file everything as it arrives",
+                "react": "act only when you react 🔖 to a message",
+            },
+        },
+    }
+
+    def _room_config_url(self, room_id: str) -> str:
+        return (
+            f"{self.homeserver}/_matrix/client/v3/user/"
+            f"{quote(self.user_id)}/rooms/{quote(room_id)}"
+            f"/account_data/{self.ROOM_CONFIG_TYPE}"
+        )
+
+    def _auth_headers(self) -> dict:
+        return {"Authorization": f"Bearer {self._client.access_token}"}
+
+    async def get_room_config(self, room_id: str) -> dict:
+        """The bot's config for this room (its ``dev.famstack.room`` room
+        account data), or ``{}`` when unset. Read fresh; best-effort, so a
+        404 or a transient error reads as "no config" rather than crashing
+        routing."""
+        try:
+            async with self._ensure_http().get(
+                self._room_config_url(room_id), headers=self._auth_headers(),
+            ) as resp:
+                if resp.status == 200:
+                    data = await resp.json()
+                    return data if isinstance(data, dict) else {}
+                return {}
+        except Exception as e:
+            logger.debug("[{}] room config read failed in {}: {}",
+                         self.name, room_id, e)
+            return {}
+
+    async def set_room_config(self, room_id: str, **updates) -> bool:
+        """Merge ``updates`` into the room's config. Returns True on a
+        confirmed write — the caller reports honestly rather than acking a
+        write that never landed."""
+        merged = {**await self.get_room_config(room_id), **updates}
+        try:
+            async with self._ensure_http().put(
+                self._room_config_url(room_id),
+                headers=self._auth_headers(), json=merged,
+            ) as resp:
+                if resp.status >= 400:
+                    logger.warning("[{}] room config write failed in {}: {} {}",
+                                   self.name, room_id, resp.status,
+                                   await resp.text())
+                    return False
+                return True
+        except Exception as e:
+            logger.warning("[{}] room config write error in {}: {}",
+                           self.name, room_id, e)
+            return False
 
-            [room_modes]
-            "#family:home.local"  = "mention"  # only when @-tagged
-            "#friends:home.local" = "off"      # ignore entirely
-            "#open-chat:home"     = "always"   # current default
+    async def _maybe_handle_config_command(self, room, event) -> bool:
+        """Handle a ``!config`` room-config command. Returns True when the
+        message was a config command (and is now handled), so the caller
+        stops routing it as anything else.
 
-        With the two always-on cases handled upstream, "mention" mode
-        collapses to "ignore" here (mentions never reach this branch).
-        The framework default — react in every group room — is the
-        least-surprise baseline; bots that want to be quieter override
-        this method.
+        Grammar:
+          ``!config``                 show the current config + options
+          ``!config <option> <value>``  set an option (from CONFIG_OPTIONS)
+
+        Any room member may set it; families are high-trust. Dispatched
+        ahead of the room-mode gate so a room can always be switched back
+        out of react mode.
         """
-        del ctx
+        body = (getattr(event, "body", "") or "").strip()
+        if not body.startswith("!config"):
+            return False
+        reply_to = getattr(event, "event_id", None)
+        parts = body.split()
+
+        # `!config <option> <value>` — set, if the option/value are known.
+        if len(parts) >= 3:
+            key, value = parts[1], parts[2]
+            opt = self.CONFIG_OPTIONS.get(key)
+            if opt and value in opt["describe"]:
+                if await self.set_room_config(room.room_id, **{key: value}):
+                    ack = f"✅ **{key}** is now **{value}**: {opt['describe'][value]}."
+                else:
+                    ack = ("⚠️ Couldn't save that setting (homeserver error). "
+                           "The room config is unchanged.")
+                await self._send(room.room_id, ack, reply_to)
+                return True
+
+        # Bare `!config`, or an unrecognized option/value: show the status
+        # view, which doubles as the help (it lists every option + values).
+        await self._send(
+            room.room_id, await self._render_config(room.room_id), reply_to,
+        )
         return True
 
+    async def _render_config(self, room_id: str) -> str:
+        """The `!config` status view: current value of each option and the
+        values it can take, built from CONFIG_OPTIONS so it never drifts.
+        Greeting-flavored but compact."""
+        cfg = await self.get_room_config(room_id)
+        lines = ["⚙️ **Room config**", ""]
+        for key, opt in self.CONFIG_OPTIONS.items():
+            current = cfg.get(key, opt["default"])
+            lines.append(f"**{key}**: {current}")
+            for value, desc in opt["describe"].items():
+                mark = "▸" if value == current else "·"
+                lines.append(f"  {mark} {value}: {desc}")
+            lines.append("")
+        lines.append(
+            "Set one with `!config <option> <value>`, "
+            "for example `!config process react`."
+        )
+        return "\n".join(lines)
+
+    @staticmethod
+    def normalize_emoji(key: str) -> str:
+        """Strip the emoji variation selector so a reaction key like
+        ``'👍️'`` (👍 + U+FE0F) compares equal to ``'👍'``. Reaction keys
+        frequently carry the selector; matching a binding without
+        normalizing silently misses those reactions."""
+        return (key or "").replace("\uFE0F", "").strip()
+
     @staticmethod
     def strip_mention(
         body: str, bot_user_id: str, *, formatted_body: str | None = None,
diff --git a/stacklets/docs/bot/archivist.py b/stacklets/docs/bot/archivist.py
index 5332e30..4a61729 100644
--- a/stacklets/docs/bot/archivist.py
+++ b/stacklets/docs/bot/archivist.py
@@ -39,6 +39,7 @@
 from PIL import Image
 from nio import (
     AsyncClient,
+    ReactionEvent,
     RoomMessageMedia,
     RoomMessageImage,
     RoomMessageFile,
@@ -49,7 +50,7 @@
 from capture_tags import CaptureTagCache
 from extractors import TextExtractor, UrlExtractor
 from git_mirror import GitMirror
-from microbot import EYES, MicroBot
+from microbot import CHECK, CROSS, EYES, MicroBot
 from pdf_analysis import (
     DEFAULT_REFORMAT_MAX_PDF_PAGES,
     DEFAULT_VISION_MAX_PDF_PAGES,
@@ -350,6 +351,9 @@ def register_callbacks(self, client: AsyncClient) -> None:
             (RoomMessageMedia, RoomMessageImage, RoomMessageFile, RoomMessageAudio),
         )
         self.add_event_callback(self._on_text, RoomMessageText)
+        # User → bot reactions: 🔖 to save the reacted message. The drain
+        # delivers reactions as typed ReactionEvents (verified on the rig).
+        self.add_event_callback(self._on_reaction, ReactionEvent)
 
     async def start(self) -> None:
         logger.info("[archivist] Config: paperless={} openai={} language={} classify={} reformat={}",
@@ -1271,6 +1275,14 @@ async def _reply_for_outcome(
         That chain depends on chat-only inputs (openai_url, the
         translator) so it lives here, not in the pipeline.
         """
+        # Terminal glyph alongside the 👀: ❌ when nothing was filed
+        # (upload/OCR failed), ✅ otherwise (filed, even if classification
+        # was partial or the document was already on file).
+        if o.status in ("upload_failed", "ocr_failed"):
+            await self._react(room_id, reply_to, CROSS)
+        else:
+            await self._react(room_id, reply_to, CHECK)
+
         if o.status == "upload_failed":
             await self._answer(room_id, self.t("upload_failed", name=o.display_name), reply_to)
             return
@@ -1363,7 +1375,7 @@ async def _on_file(self, room, event) -> None:
         # always sees the intro before any other reply from the bot.
         await self._send_room_welcome_if_needed(room, ctx)
         mentioned = self._is_bot_mentioned(event)
-        if not self._should_react(ctx, mentioned=mentioned):
+        if not await self._should_react(ctx, mentioned=mentioned):
             logger.debug(
                 "[archivist] skipping file from {} in {} per room mode",
                 event.sender, ctx.room_id,
@@ -1482,6 +1494,11 @@ async def _on_text(self, room, event: RoomMessageText) -> None:
         query = event.body.strip()
         if not query:
             return
+        # Room-config commands (`!config ...`) are handled before the
+        # room-mode gate, so a room can always be switched back out of
+        # react mode, and before routing so they never read as a capture.
+        if await self._maybe_handle_config_command(room, event):
+            return
         query_lower = query.lower()
         reply_to = event.event_id
 
@@ -1498,7 +1515,7 @@ async def _on_text(self, room, event: RoomMessageText) -> None:
             event.sender, ctx.room_id, ctx.alias, ctx.is_dm,
             is_documents, len(ctx.members), mentioned,
         )
-        if not self._should_react(ctx, mentioned=mentioned):
+        if not await self._should_react(ctx, mentioned=mentioned):
             logger.debug(
                 "[archivist] skipping {} in {} per room mode",
                 event.sender, ctx.room_id,
@@ -1668,6 +1685,81 @@ async def _on_text(self, room, event: RoomMessageText) -> None:
                 room.room_id, query[:60],
             )
 
+    # ── Reactions: user → bot per-message routing ────────────────────────
+    #
+    # A small hard-coded registry maps a (normalized) emoji to the handler
+    # method that runs when a family member drops it on a message. To add a
+    # binding — 🗑 redact, 👎 reclassify — add one entry here and write the
+    # `_react_*` method; the dispatcher needs no changes. Handlers receive
+    # the already-fetched target event, so they decide for themselves
+    # whether a bot-authored target is valid (bookmark says no, redact yes).
+
+    def _reaction_handlers(self) -> dict:
+        return {
+            "🔖": self._react_bookmark,
+            "📌": self._react_bookmark,
+        }
+
+    async def _on_reaction(self, room, event) -> None:
+        """Dispatch a user's reaction to its registered handler.
+
+        Generic and binding-agnostic: ignore the bot's own (and other
+        bots') reactions, normalize the emoji, look up the handler, fetch
+        the reacted message once, and hand it off. Idempotency and any
+        bot-target policy live in the handlers, keyed on the target event
+        id so a drain replay dedups rather than acting twice.
+        """
+        if event.sender == self.user_id or self.is_bot_user(event.sender):
+            return
+        emoji = self.normalize_emoji(getattr(event, "key", ""))
+        handler = self._reaction_handlers().get(emoji)
+        if handler is None:
+            return
+        target_id = getattr(event, "reacts_to", None)
+        if not target_id:
+            return
+        try:
+            resp = await self._client.room_get_event(room.room_id, target_id)
+        except Exception as e:
+            logger.debug("[archivist] reaction target fetch failed: {}", e)
+            return
+        target = getattr(resp, "event", None)
+        if target is None:
+            return
+        await handler(room, event, target, target_id)
+
+    async def _react_bookmark(self, room, event, target, target_id) -> None:
+        """🔖 / 📌 — bookmark the reacted message into the room: the same
+        capture auto-mode would make, but on demand. The only capture path
+        in a `!config process react` room.
+
+        Never bookmarks a bot message (a filing, a welcome). Attribution
+        is the message author, not the reactor — we're saving their
+        content — and the capture is keyed on the target event id so a
+        replay or a second reactor dedups downstream.
+        """
+        if self.is_bot_user(getattr(target, "sender", "")):
+            return
+        body = (getattr(target, "body", "") or "").strip()
+        if not body:
+            # v1 bookmarks text/URL messages; file uploads are a follow-up.
+            return
+        author = target.sender
+        if _is_just_url(body):
+            await self._handle_capture(
+                room.room_id, body, author, target_id, capture_id=target_id,
+            )
+        elif (embedded_url := _first_url(body)) is not None:
+            hint = body.replace(embedded_url, "", 1).strip(" \t\n\r:.,;!?—-")
+            await self._handle_capture(
+                room.room_id, embedded_url, author, target_id,
+                capture_id=target_id, user_hint=hint or None,
+            )
+        else:
+            await self._handle_text_capture(
+                room.room_id, body, author, target_id, capture_id=target_id,
+            )
+
     # ── Inbound source events (mail bot, future ingest channels) ──────────
 
     async def _handle_source_message(self, room, event, source: dict) -> None:
@@ -2055,6 +2147,13 @@ async def _reply_for_capture(
         `capture.*` envelope as metadata so the user can reply-to-
         correct the same way they do with document filings.
         """
+        # Terminal glyph alongside the 👀: ❌ on a genuine failure, ✅
+        # otherwise. `empty` did nothing, so it gets no glyph.
+        if o.status in ("extract_failed", "no_mirror"):
+            await self._react(room_id, reply_to, CROSS)
+        elif o.status != "empty":
+            await self._react(room_id, reply_to, CHECK)
+
         if o.status == "empty":
             return
         if o.status == "extract_failed":
diff --git a/tests/integration/test_room_modes_e2e.py b/tests/integration/test_room_modes_e2e.py
new file mode 100644
index 0000000..07c4fb8
--- /dev/null
+++ b/tests/integration/test_room_modes_e2e.py
@@ -0,0 +1,177 @@
+"""Per-room process modes + 🔖 bookmark reactions — end-to-end INTENT.
+
+This file is an executable specification of how the archivist should
+behave around per-room config and user-driven reactions. It is the
+contract; the implementation is expected to satisfy it.
+
+> STATUS: UNVERIFIED against the integration rig. The behavior below was
+> verified by hand against a running instance on 2026-06-26, but this
+> test has not yet been run green through `stacktests` (server
+> test.local). It is marked `unverified` + `xfail(strict=False)` so it
+> neither blocks the suite nor is trusted until reconciled. Run it at the
+> end of the iteration and fix the test (or the implementation) before
+> tagging the next beta. Do not silently delete it to make CI green.
+
+The intent, in one place:
+
+  Rooms have a `process` mode, set in chat with `!config`:
+
+    `!config`                 -> show the room's current config + options
+    `!config process auto`    -> file everything as it arrives (default)
+    `!config process react`   -> ignore plain messages; act only on a
+                                 user's reaction
+
+  In a `react` room the bot stays quiet until a family member reacts 🔖
+  (or 📌) to a message, which bookmarks it (the same capture `auto` mode
+  would have made). In an `auto` room the bot processes messages as they
+  land. Either way, a processed message ends up marked 👀 (picked up) and
+  then ✅ (filed) or ❌ (failed) so the outcome is visible at a glance in
+  the timeline, while the detailed reply lives in a thread.
+
+  Room mode is stored in the bot's own room account data, so it works
+  without granting the bot any room power level (see
+  project_bot_power_level_rule).
+
+Why a group room: a 2-member room with the bot is a DM, which always
+reacts regardless of mode. We add Marge so the room has 3 members and
+the mode gate actually applies.
+"""
+from __future__ import annotations
+
+import time
+
+import pytest
+from nio import AsyncClient
+
+from tests.integration.matrix import event_type, fetch_room_events
+
+ARCHIVIST = "@archivist-bot:test.local"
+MARGE = "@marge:test.local"
+EYES, CHECK = "👀", "✅"
+
+pytestmark = [pytest.mark.unverified]
+
+
+def _norm(key: str) -> str:
+    return (key or "").replace("\uFE0F", "").strip()
+
+
+def _bot_reacted(events, *, key: str, target: str) -> bool:
+    """Did the archivist add reaction `key` to event `target`?"""
+    return any(
+        event_type(e) == "m.reaction"
+        and getattr(e, "sender", None) == ARCHIVIST
+        and getattr(e, "reacts_to", None) == target
+        and _norm(getattr(e, "key", "")) == key
+        for e in events
+    )
+
+
+def _bot_said(events, needle: str) -> bool:
+    """Did the archivist post a message containing `needle`?"""
+    return any(
+        getattr(e, "sender", None) == ARCHIVIST
+        and needle.lower() in (getattr(e, "body", "") or "").lower()
+        for e in events
+    )
+
+
+async def _send(client, room_id: str, body: str) -> str:
+    r = await client.room_send(
+        room_id, "m.room.message", {"msgtype": "m.text", "body": body},
+    )
+    return r.event_id
+
+
+async def _react(client, room_id: str, target: str, key: str) -> None:
+    await client.room_send(room_id, "m.reaction", {"m.relates_to": {
+        "rel_type": "m.annotation", "event_id": target, "key": key}})
+
+
+@pytest.mark.xfail(
+    reason="UNVERIFIED intent spec: not yet run green against the rig. "
+           "Reconcile before the next beta tag.",
+    strict=False,
+)
+async def test_room_modes_and_bookmark_reactions(homer, matrix):
+    """The full contract, driven as Homer in a 3-member group room.
+
+    1. `!config process react` is acknowledged and persists.
+    2. In react mode a bare URL is NOT processed (no 👀, no filing).
+    3. 🔖 on that message bookmarks it: the bot marks it 👀 then ✅.
+    4. `!config process auto` is acknowledged.
+    5. In auto mode a bare URL is processed without any reaction.
+    """
+    marge_creds = matrix["marge"]
+    marge = AsyncClient(marge_creds.homeserver, marge_creds.user_id)
+    marge.access_token = marge_creds.access_token
+    marge.device_id = marge_creds.device_id
+    try:
+        # 3-member group room: Homer + Marge + the archivist.
+        created = await homer.room_create(
+            name=f"modes-e2e-{int(time.time())}", invite=[MARGE, ARCHIVIST],
+        )
+        room = created.room_id
+        await marge.join(room)
+
+        # The bot auto-accepts the invite and posts a welcome on join;
+        # wait for any sign of it (cold start can take ~40s).
+        seen: list = []
+        for _ in range(13):
+            seen += await fetch_room_events(homer, room, duration=10)
+            if any(getattr(e, "sender", None) == ARCHIVIST for e in seen):
+                break
+        assert any(getattr(e, "sender", None) == ARCHIVIST for e in seen), \
+            "archivist never joined or responded in the room"
+
+        # 1. Switch the room to react mode.
+        await _send(homer, room, "!config process react")
+        ack = await fetch_room_events(homer, room, duration=30)
+        assert _bot_said(ack, "is now") and _bot_said(ack, "react"), \
+            "expected an acknowledgement that the room is now in react mode"
+
+        # `!config` (bare) shows the current config + options.
+        await _send(homer, room, "!config")
+        status = await fetch_room_events(homer, room, duration=30)
+        assert _bot_said(status, "room config") and _bot_said(status, "process"), \
+            "bare !config should print the current config and its options"
+
+        # 2. React mode gates auto-processing: a bare URL is left alone.
+        e1 = await _send(homer, room, "https://en.wikipedia.org/wiki/Camping")
+        gated = await fetch_room_events(homer, room, duration=40)
+        assert not _bot_reacted(gated, key=EYES, target=e1), \
+            "react mode must not auto-process a plain message"
+        assert not _bot_said(gated, "captured"), \
+            "react mode must not file a plain message"
+
+        # 3. 🔖 triggers the capture: 👀 then ✅ on the bookmarked message.
+        await _react(homer, room, e1, "🔖")
+        done: list = []
+        for _ in range(12):
+            done += await fetch_room_events(homer, room, duration=10)
+            if _bot_reacted(done, key=CHECK, target=e1):
+                break
+        assert _bot_reacted(done, key=EYES, target=e1), \
+            "🔖 should make the bot pick up the message (👀)"
+        assert _bot_reacted(done, key=CHECK, target=e1), \
+            "a successful bookmark should be marked ✅"
+
+        # 4. Switch back to auto mode.
+        await _send(homer, room, "!config process auto")
+        ack2 = await fetch_room_events(homer, room, duration=30)
+        assert _bot_said(ack2, "is now") and _bot_said(ack2, "auto"), \
+            "expected an acknowledgement that the room is now in auto mode"
+
+        # 5. Auto mode processes a bare URL with no reaction needed.
+        e2 = await _send(homer, room, "https://en.wikipedia.org/wiki/Hiking")
+        auto: list = []
+        for _ in range(12):
+            auto += await fetch_room_events(homer, room, duration=10)
+            if _bot_reacted(auto, key=CHECK, target=e2):
+                break
+        assert _bot_reacted(auto, key=EYES, target=e2), \
+            "auto mode should pick up a plain message (👀)"
+        assert _bot_reacted(auto, key=CHECK, target=e2), \
+            "auto mode should file it and mark ✅"
+    finally:
+        await marge.close()
diff --git a/tests/stacklets/test_archivist_routing.py b/tests/stacklets/test_archivist_routing.py
index b8973e1..cc08314 100644
--- a/tests/stacklets/test_archivist_routing.py
+++ b/tests/stacklets/test_archivist_routing.py
@@ -205,10 +205,10 @@ class TestShouldReact:
     mention) goes through `_room_mode_allows_react` which is the
     placeholder for future config."""
 
-    def test_mention_in_group_room_always_reacts(self, tmp_path):
+    async def test_mention_in_group_room_always_reacts(self, tmp_path):
         """An @-tag must never be ignored, regardless of room mode.
-        Even if a future config marks this room as off, the mention
-        bypasses the mode lookup entirely."""
+        Even if the room is configured react-only, the mention bypasses
+        the mode lookup entirely."""
         bot = _build_bot(tmp_path)
         room = _room(
             canonical_alias="#family-chat:server",
@@ -216,22 +216,25 @@ def test_mention_in_group_room_always_reacts(self, tmp_path):
         )
         ctx = bot._room_context(room)
         # Pin the contract by forcing the mode gate to deny — mention
-        # must still win. The day modes ship, this test catches any
-        # regression that routes mentions through the mode lookup.
-        bot._room_mode_allows_react = lambda _ctx: False
-        assert bot._should_react(ctx, mentioned=True) is True
+        # must still win, never routing through the mode lookup.
+        async def _deny(_ctx):
+            return False
+        bot._room_mode_allows_react = _deny
+        assert await bot._should_react(ctx, mentioned=True) is True
 
-    def test_dm_always_reacts(self, tmp_path):
+    async def test_dm_always_reacts(self, tmp_path):
         """A 2-member room with the bot is a private chat. There's
         nobody else for the message to be aimed at, so the mode gate
         doesn't apply."""
         bot = _build_bot(tmp_path)
         room = _room(members=[BOT_ID, "@homer:server"])
         ctx = bot._room_context(room)
-        bot._room_mode_allows_react = lambda _ctx: False
-        assert bot._should_react(ctx, mentioned=False) is True
+        async def _deny(_ctx):
+            return False
+        bot._room_mode_allows_react = _deny
+        assert await bot._should_react(ctx, mentioned=False) is True
 
-    def test_documents_room_with_mention_reacts(self, tmp_path):
+    async def test_documents_room_with_mention_reacts(self, tmp_path):
         # Mention beats every other consideration, including docs-room
         # routing — the upstream handlers still see ctx.is_documents_room
         # and dispatch accordingly.
@@ -241,12 +244,11 @@ def test_documents_room_with_mention_reacts(self, tmp_path):
             members=[BOT_ID, "@homer:server", "@marge:server"],
         )
         ctx = bot._room_context(room)
-        assert bot._should_react(ctx, mentioned=True) is True
+        assert await bot._should_react(ctx, mentioned=True) is True
 
-    def test_group_room_no_mention_consults_mode(self, tmp_path):
-        """The only branch that talks to the future mode lookup. Today
-        the lookup defaults to True; this test pins that wiring so an
-        accidental rewrite that hard-codes True in `_should_react`
+    async def test_group_room_no_mention_consults_mode(self, tmp_path):
+        """The only branch that talks to the mode lookup. Pin the wiring
+        so an accidental rewrite that hard-codes True in `_should_react`
         bypasses the seam."""
         bot = _build_bot(tmp_path)
         room = _room(
@@ -254,24 +256,194 @@ def test_group_room_no_mention_consults_mode(self, tmp_path):
             members=[BOT_ID, "@homer:server", "@marge:server", "@bart:server"],
         )
         ctx = bot._room_context(room)
-        # Today: gate returns True, so should_react returns True.
-        assert bot._should_react(ctx, mentioned=False) is True
-        # Tomorrow: gate returns False → should_react must respect it.
-        bot._room_mode_allows_react = lambda _ctx: False
-        assert bot._should_react(ctx, mentioned=False) is False
-
-    def test_default_room_mode_is_react(self, tmp_path):
-        """`_room_mode_allows_react` returns True today (no modes
-        configured). This test pins that default so the day a mode
-        config lands, the default-on behavior is explicit not
-        accidental."""
+        async def _allow(_ctx):
+            return True
+        bot._room_mode_allows_react = _allow
+        assert await bot._should_react(ctx, mentioned=False) is True
+        async def _deny(_ctx):
+            return False
+        bot._room_mode_allows_react = _deny
+        assert await bot._should_react(ctx, mentioned=False) is False
+
+    async def test_room_mode_reads_process_config(self, tmp_path):
+        """`_room_mode_allows_react` reflects the room's `process` config:
+        unset/auto → react to messages; `react` → ignore plain messages
+        (reactions become the only trigger)."""
         bot = _build_bot(tmp_path)
         room = _room(
             canonical_alias="#whatever:server",
             members=[BOT_ID, "@a:server", "@b:server"],
         )
         ctx = bot._room_context(room)
-        assert bot._room_mode_allows_react(ctx) is True
+
+        async def _auto(_room_id):
+            return {}
+        bot.get_room_config = _auto
+        assert await bot._room_mode_allows_react(ctx) is True
+
+        async def _react(_room_id):
+            return {"process": "react"}
+        bot.get_room_config = _react
+        assert await bot._room_mode_allows_react(ctx) is False
+
+
+class TestReactionDispatch:
+    """`_on_reaction` routes a user's emoji to a registered handler. v1
+    binding: 🔖 / 📌 bookmark the reacted message (the same capture
+    auto-mode makes), attributed to the message author and keyed on the
+    target event id so a drain replay or a second reactor dedups."""
+
+    def _bot(self, tmp_path, *, target):
+        bot = _build_bot(tmp_path)
+        cap, txt = [], []
+
+        async def _cap(room_id, url, sender, reply_to, *,
+                       capture_id=None, user_hint=None):
+            cap.append({"url": url, "sender": sender, "reply_to": reply_to,
+                        "capture_id": capture_id, "hint": user_hint})
+
+        async def _txt(room_id, text, sender, reply_to, *, capture_id=None):
+            txt.append({"text": text, "sender": sender, "reply_to": reply_to,
+                        "capture_id": capture_id})
+
+        bot._handle_capture = _cap
+        bot._handle_text_capture = _txt
+
+        async def _get_event(room_id, event_id):
+            return SimpleNamespace(event=target)
+
+        bot._client = SimpleNamespace(room_get_event=_get_event)
+        return bot, cap, txt
+
+    @staticmethod
+    def _reaction(key="🔖", reacts_to="$tgt", sender="@homer:server"):
+        return SimpleNamespace(
+            key=key, reacts_to=reacts_to, sender=sender,
+            source={"content": {}},
+        )
+
+    @staticmethod
+    def _target(body, sender="@marge:server"):
+        return SimpleNamespace(
+            sender=sender, body=body, source={"content": {"body": body}},
+        )
+
+    async def test_bookmark_url_message_captures(self, tmp_path):
+        bot, cap, txt = self._bot(
+            tmp_path, target=self._target("https://example.com/gear"),
+        )
+        await bot._on_reaction(_room(room_id="!r:server"), self._reaction())
+        assert len(cap) == 1 and not txt
+        assert cap[0]["url"] == "https://example.com/gear"
+        assert cap[0]["sender"] == "@marge:server"   # message author, not reactor
+        assert cap[0]["capture_id"] == "$tgt"         # idempotent on target id
+
+    async def test_bookmark_text_message_captures_as_note(self, tmp_path):
+        bot, cap, txt = self._bot(
+            tmp_path, target=self._target("remember the boiler service in March"),
+        )
+        await bot._on_reaction(_room(), self._reaction())
+        assert len(txt) == 1 and not cap
+        assert txt[0]["text"].startswith("remember the boiler")
+        assert txt[0]["capture_id"] == "$tgt"
+
+    async def test_bookmark_embedded_url_passes_hint(self, tmp_path):
+        bot, cap, txt = self._bot(
+            tmp_path, target=self._target("camping gear list https://example.com/x"),
+        )
+        await bot._on_reaction(_room(), self._reaction())
+        assert len(cap) == 1
+        assert cap[0]["url"] == "https://example.com/x"
+        assert cap[0]["hint"] == "camping gear list"
+
+    async def test_variation_selector_emoji_still_dispatches(self, tmp_path):
+        bot, cap, _ = self._bot(tmp_path, target=self._target("https://example.com"))
+        await bot._on_reaction(_room(), self._reaction(key="🔖\uFE0F"))
+        assert len(cap) == 1
+
+    async def test_pushpin_emoji_dispatches(self, tmp_path):
+        bot, cap, _ = self._bot(tmp_path, target=self._target("https://example.com"))
+        await bot._on_reaction(_room(), self._reaction(key="📌"))
+        assert len(cap) == 1
+
+    async def test_unregistered_emoji_ignored(self, tmp_path):
+        bot, cap, txt = self._bot(tmp_path, target=self._target("https://example.com"))
+        await bot._on_reaction(_room(), self._reaction(key="👍"))
+        assert not cap and not txt
+
+    async def test_bot_reactor_ignored(self, tmp_path):
+        bot, cap, txt = self._bot(tmp_path, target=self._target("https://example.com"))
+        await bot._on_reaction(
+            _room(), self._reaction(sender="@archivist-bot:server"),
+        )
+        assert not cap and not txt
+
+    async def test_bot_authored_target_not_bookmarked(self, tmp_path):
+        # 🔖 on the bot's own filing must not re-capture the filing.
+        bot, cap, txt = self._bot(
+            tmp_path,
+            target=self._target("Filed: passport", sender="@archivist-bot:server"),
+        )
+        await bot._on_reaction(_room(), self._reaction())
+        assert not cap and not txt
+
+
+class TestOutcomeGlyph:
+    """After a capture/filing finishes, the bot marks the source message
+    with a terminal glyph alongside the 👀: ✅ when something was filed,
+    ❌ on a genuine failure, nothing for a silent drop. The detailed
+    reply lives in a thread, so this is the at-a-glance timeline signal."""
+
+    CHECK = "✅"
+    CROSS = "❌"
+
+    def _bot(self, tmp_path):
+        bot = _build_bot(tmp_path)
+        reacts = []
+
+        async def _react(room_id, eid, emoji):
+            reacts.append(emoji)
+
+        async def _noop(*a, **k):
+            return None
+
+        bot._react = _react
+        bot._answer = _noop
+        bot._send = _noop
+        return bot, reacts
+
+    async def test_capture_success_checks(self, tmp_path, monkeypatch):
+        bot, reacts = self._bot(tmp_path)
+        monkeypatch.setattr("archivist.render_capture_reply", lambda *a, **k: "x")
+        o = SimpleNamespace(
+            status="captured", source_title_hint="t", classification={},
+            display_link="http://x", transcript=None, envelope=None,
+        )
+        await bot._reply_for_capture("!r:server", o, "$tgt")
+        assert reacts == [self.CHECK]
+
+    async def test_capture_extract_failed_crosses(self, tmp_path):
+        bot, reacts = self._bot(tmp_path)
+        o = SimpleNamespace(status="extract_failed", failure_reason="url")
+        await bot._reply_for_capture("!r:server", o, "$tgt")
+        assert reacts == [self.CROSS]
+
+    async def test_capture_empty_gets_no_glyph(self, tmp_path):
+        bot, reacts = self._bot(tmp_path)
+        await bot._reply_for_capture("!r:server", SimpleNamespace(status="empty"), "$tgt")
+        assert reacts == []
+
+    async def test_filing_success_checks(self, tmp_path):
+        bot, reacts = self._bot(tmp_path)
+        o = SimpleNamespace(status="filed_no_details", display_name="x", link="y")
+        await bot._reply_for_outcome("!r:server", o, "$tgt")
+        assert reacts == [self.CHECK]
+
+    async def test_filing_ocr_failed_crosses(self, tmp_path):
+        bot, reacts = self._bot(tmp_path)
+        o = SimpleNamespace(status="ocr_failed", display_name="x")
+        await bot._reply_for_outcome("!r:server", o, "$tgt")
+        assert reacts == [self.CROSS]
 
 
 class TestPastePredicate:
diff --git a/tests/stacklets/test_archivist_source.py b/tests/stacklets/test_archivist_source.py
index 3ca4e25..01fedb1 100644
--- a/tests/stacklets/test_archivist_source.py
+++ b/tests/stacklets/test_archivist_source.py
@@ -85,6 +85,10 @@ async def _none():
     return None
 
 
+async def _true():
+    return True
+
+
 @pytest.mark.asyncio
 async def test_email_source_folds_through_capture(tmp_path):
     bot = _bot(tmp_path)
@@ -303,7 +307,7 @@ async def rec_binary(**kwargs):
     bot._room_context = lambda room: SimpleNamespace(room_id=room.room_id)
     bot._send_room_welcome_if_needed = lambda *_a, **_k: _none()
     bot._is_bot_mentioned = lambda _e: False
-    bot._should_react = lambda *_a, **_k: True
+    bot._should_react = lambda *_a, **_k: _true()
     bot._is_documents_room = lambda _ctx: False
     bot._download_media = lambda _url: _bytes()
     bot._send = lambda *_a, **_k: _none()
diff --git a/tests/stacklets/test_microbot.py b/tests/stacklets/test_microbot.py
index c307acd..b50cda3 100644
--- a/tests/stacklets/test_microbot.py
+++ b/tests/stacklets/test_microbot.py
@@ -678,3 +678,221 @@ async def test_none_when_fetch_raises(self, tmp_path):
         bot, client = _bare_bot(tmp_path)
         client.get_event_raises = ConnectionError("synapse down")
         assert await bot._reply_parent_envelope("!r:server", self._reply_to("$x")) is None
+
+
+# ── Per-room config + emoji + !config command ────────────────────────────
+
+
+class _FakeResp:
+    """Minimal aiohttp-response stand-in (async context manager)."""
+
+    def __init__(self, status, data=None):
+        self.status = status
+        self._data = data
+
+    async def json(self):
+        return self._data
+
+    async def text(self):
+        return str(self._data)
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+
+class _FakeHttp:
+    """Test double for the bot's aiohttp session backing room account
+    data: an in-memory key→json store. ``raise_on`` forces a transport
+    error to exercise the best-effort read path. We stub the network
+    boundary here, not nio — the real REST round-trip is covered e2e."""
+
+    def __init__(self):
+        self.store: dict[str, dict] = {}
+        self.raise_on: set[str] = set()
+
+    def get(self, url, headers=None):
+        if "get" in self.raise_on:
+            raise RuntimeError("homeserver hiccup")
+        if url in self.store:
+            return _FakeResp(200, self.store[url])
+        return _FakeResp(404)
+
+    def put(self, url, headers=None, json=None):
+        if "put" in self.raise_on:
+            raise RuntimeError("homeserver hiccup")
+        self.store[url] = json
+        return _FakeResp(200, json)
+
+
+class TestRoomConfig:
+    """`get_room_config` / `set_room_config` read+write the bot's room
+    account data over the Matrix REST API. No local cache — a read hits
+    the homeserver every time, and a 404 or transient error reads as "no
+    config" rather than crashing routing."""
+
+    @staticmethod
+    def _bot(tmp_path):
+        bot, _ = _bare_bot(tmp_path)
+        bot._client.access_token = "tok"
+        bot._http = _FakeHttp()
+        return bot, bot._http
+
+    @pytest.mark.asyncio
+    async def test_get_returns_empty_when_unset(self, tmp_path):
+        bot, _ = self._bot(tmp_path)
+        assert await bot.get_room_config("!r:server") == {}
+
+    @pytest.mark.asyncio
+    async def test_set_then_get_roundtrips(self, tmp_path):
+        bot, http = self._bot(tmp_path)
+        assert await bot.set_room_config("!r:server", process="react") is True
+        assert await bot.get_room_config("!r:server") == {"process": "react"}
+        # Stored under the room-scoped account-data URL for this bot.
+        assert any("/account_data/dev.famstack.room" in u for u in http.store)
+
+    @pytest.mark.asyncio
+    async def test_set_merges_into_existing(self, tmp_path):
+        bot, _ = self._bot(tmp_path)
+        await bot.set_room_config("!r:server", process="react")
+        await bot.set_room_config("!r:server", other="x")
+        assert await bot.get_room_config("!r:server") == {
+            "process": "react", "other": "x",
+        }
+
+    @pytest.mark.asyncio
+    async def test_get_swallows_read_error(self, tmp_path):
+        bot, http = self._bot(tmp_path)
+        http.raise_on.add("get")
+        assert await bot.get_room_config("!r:server") == {}
+
+    @pytest.mark.asyncio
+    async def test_set_reports_failure(self, tmp_path):
+        bot, http = self._bot(tmp_path)
+        http.raise_on.add("put")
+        assert await bot.set_room_config("!r:server", process="react") is False
+
+
+class TestNormalizeEmoji:
+    """Reaction keys often carry the U+FE0F variation selector; matching
+    a binding without normalizing silently misses those reactions."""
+
+    def test_strips_variation_selector(self):
+        assert MicroBot.normalize_emoji("\U0001F44D\uFE0F") == "\U0001F44D"
+
+    def test_plain_emoji_unchanged(self):
+        assert MicroBot.normalize_emoji("\U0001F516") == "\U0001F516"
+
+    def test_none_is_safe(self):
+        assert MicroBot.normalize_emoji(None) == ""
+
+
+class TestConfigCommand:
+    """`!config process auto|react` writes the room mode. Handled (returns
+    True) so the caller stops routing it; non-config text passes through
+    (returns False)."""
+
+    @staticmethod
+    def _bot(tmp_path):
+        bot, client = _bare_bot(tmp_path)
+        bot._client.access_token = "tok"
+        bot._http = _FakeHttp()
+        return bot, client
+
+    @staticmethod
+    def _evt(body):
+        return SimpleNamespace(body=body, event_id="$e", source={"content": {}})
+
+    @pytest.mark.asyncio
+    async def test_sets_react_mode(self, tmp_path):
+        bot, client = self._bot(tmp_path)
+        room = SimpleNamespace(room_id="!r:server")
+        handled = await bot._maybe_handle_config_command(
+            room, self._evt("!config process react"),
+        )
+        assert handled is True
+        assert await bot.get_room_config("!r:server") == {"process": "react"}
+        # Ack describes the value, sourced from the registry.
+        ack = client.sends[-1][2]["body"]
+        assert "react" in ack and "react 🔖" in ack
+
+    @pytest.mark.asyncio
+    async def test_bare_config_shows_status_and_options(self, tmp_path):
+        bot, client = self._bot(tmp_path)
+        room = SimpleNamespace(room_id="!r:server")
+        handled = await bot._maybe_handle_config_command(
+            room, self._evt("!config"),
+        )
+        assert handled is True
+        body = client.sends[-1][2]["body"]
+        # Current value (default when unset) + every allowed value + how to set.
+        assert "process" in body
+        assert "auto" in body and "react" in body
+        assert "!config process react" in body
+
+    @pytest.mark.asyncio
+    async def test_bare_config_marks_current_value(self, tmp_path):
+        bot, client = self._bot(tmp_path)
+        await bot.set_room_config("!r:server", process="react")
+        room = SimpleNamespace(room_id="!r:server")
+        await bot._maybe_handle_config_command(room, self._evt("!config"))
+        body = client.sends[-1][2]["body"]
+        assert "▸ react" in body   # current value is marked
+        assert "· auto" in body    # the other value is not
+
+    @pytest.mark.asyncio
+    async def test_sets_auto_mode(self, tmp_path):
+        bot, _ = self._bot(tmp_path)
+        room = SimpleNamespace(room_id="!r:server")
+        await bot._maybe_handle_config_command(
+            room, self._evt("!config process auto"),
+        )
+        assert (await bot.get_room_config("!r:server"))["process"] == "auto"
+
+    @pytest.mark.asyncio
+    async def test_non_config_message_passes_through(self, tmp_path):
+        bot, _ = self._bot(tmp_path)
+        room = SimpleNamespace(room_id="!r:server")
+        handled = await bot._maybe_handle_config_command(
+            room, self._evt("just chatting"),
+        )
+        assert handled is False
+
+    @pytest.mark.asyncio
+    async def test_unknown_subcommand_shows_status(self, tmp_path):
+        bot, client = self._bot(tmp_path)
+        room = SimpleNamespace(room_id="!r:server")
+        handled = await bot._maybe_handle_config_command(
+            room, self._evt("!config wat"),
+        )
+        assert handled is True              # consumed, not routed onward
+        assert await bot.get_room_config("!r:server") == {}  # nothing written
+        # Falls back to the status view (which lists the valid options).
+        assert "Room config" in client.sends[-1][2]["body"]
+
+    @pytest.mark.asyncio
+    async def test_invalid_value_not_written(self, tmp_path):
+        bot, client = self._bot(tmp_path)
+        room = SimpleNamespace(room_id="!r:server")
+        handled = await bot._maybe_handle_config_command(
+            room, self._evt("!config process loud"),
+        )
+        assert handled is True
+        assert await bot.get_room_config("!r:server") == {}  # bad value rejected
+        assert "Room config" in client.sends[-1][2]["body"]
+
+    @pytest.mark.asyncio
+    async def test_write_failure_reported_not_acked(self, tmp_path):
+        # A failed write must not be acked as success — the regression the
+        # power-level eval exposed (state-event write silently rejected).
+        bot, client = self._bot(tmp_path)
+        bot._http.raise_on.add("put")
+        room = SimpleNamespace(room_id="!r:server")
+        handled = await bot._maybe_handle_config_command(
+            room, self._evt("!config process react"),
+        )
+        assert handled is True
+        body = client.sends[-1][2].get("body", "")
+        assert "Couldn't save" in body and "react mode" not in body