diff --git a/hermes_cli/main.py b/hermes_cli/main.py
index b71c59f38353..434cf25e8e7f 100644
--- a/hermes_cli/main.py
+++ b/hermes_cli/main.py
@@ -12805,7 +12805,7 @@ def _dispatch_secrets(args):  # noqa: ANN001
                     plugin_parser.set_defaults(func=cmd_info["handler_fn"])
                 seen_plugin_commands.add(cmd_info["name"])
 
-            discover_plugins()
+            discover_plugins(cli_command=_first_positional_argv())
             for cmd_info in get_plugin_manager()._cli_commands.values():
                 if cmd_info["name"] in seen_plugin_commands:
                     continue
diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py
index d5e4b3ff8c1c..f79044441cf9 100644
--- a/hermes_cli/plugins.py
+++ b/hermes_cli/plugins.py
@@ -1227,7 +1227,12 @@ def __init__(self) -> None:
     # Public
     # -----------------------------------------------------------------------
 
-    def discover_and_load(self, force: bool = False) -> None:
+    def discover_and_load(
+        self,
+        force: bool = False,
+        *,
+        cli_command: Optional[str] = None,
+    ) -> None:
         """Scan all plugin sources and load each plugin found.
 
         When ``force`` is true, clear cached discovery state first so config
@@ -1235,6 +1240,7 @@ def discover_and_load(self, force: bool = False) -> None:
         sessions without requiring a full agent restart.
         """
         if self._discovered and not force:
+            self._load_deferred_platform_for_cli_command(cli_command)
             return
         # Safe mode (--safe-mode / HERMES_SAFE_MODE=1): troubleshooting run
         # with all customizations disabled. Skip plugin discovery entirely so
@@ -1268,6 +1274,7 @@ def discover_and_load(self, force: bool = False) -> None:
         except BaseException:
             self._discovered = False
             raise
+        self._load_deferred_platform_for_cli_command(cli_command)
 
     def _discover_and_load_inner(self) -> None:
         """The actual discovery sweep — see :meth:`discover_and_load`."""
@@ -1700,6 +1707,21 @@ def _loader(_manifest: PluginManifest = manifest) -> None:
             )
             self._load_plugin(manifest)
 
+    def _load_deferred_platform_for_cli_command(
+        self,
+        cli_command: Optional[str],
+    ) -> None:
+        if not cli_command or cli_command in self._cli_commands:
+            return
+        for loaded in list(self._plugins.values()):
+            if not loaded.deferred:
+                continue
+            manifest = loaded.manifest
+            if self._platform_name_from_manifest(manifest) != cli_command:
+                continue
+            self._load_plugin(manifest)
+            return
+
     def _load_plugin(self, manifest: PluginManifest) -> None:
         """Import a plugin module and call its ``register(ctx)`` function."""
         loaded = LoadedPlugin(manifest=manifest)
@@ -1992,13 +2014,17 @@ def get_plugin_manager() -> PluginManager:
     return _plugin_manager
 
 
-def discover_plugins(force: bool = False) -> None:
+def discover_plugins(
+    force: bool = False,
+    *,
+    cli_command: Optional[str] = None,
+) -> None:
     """Discover and load all plugins.
 
     Default behavior is idempotent. Pass ``force=True`` to rescan plugin
     manifests and reload state in the current process.
     """
-    get_plugin_manager().discover_and_load(force=force)
+    get_plugin_manager().discover_and_load(force=force, cli_command=cli_command)
 
 
 def invoke_hook(hook_name: str, **kwargs: Any) -> List[Any]:
diff --git a/plugins/platforms/photon/adapter.py b/plugins/platforms/photon/adapter.py
index d6e627f667bc..e28119860354 100644
--- a/plugins/platforms/photon/adapter.py
+++ b/plugins/platforms/photon/adapter.py
@@ -64,6 +64,7 @@
 from gateway.platforms.helpers import strip_markdown
 
 from .auth import load_project_credentials
+from .state import PhotonStateStore
 
 logger = logging.getLogger(__name__)
 
@@ -184,6 +185,15 @@ def _markdown_enabled() -> bool:
     }
 
 
+def _timestamp_from_iso(value: Any) -> Optional[float]:
+    if not isinstance(value, str) or not value:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
+    except ValueError:
+        return None
+
+
 # ---------------------------------------------------------------------------
 # Adapter
 
@@ -254,6 +264,8 @@ def __init__(self, config: PlatformConfig):
         # react action default to "the message that triggered me" without
         # requiring the model to thread message ids through tool calls.
         self._last_inbound_by_chat: Dict[str, str] = {}
+        self._photon_state = PhotonStateStore()
+        self._hydrate_persistent_state()
         # Last time we sent a typing indicator per chat, for cooldown gating.
         self._typing_last_sent: Dict[str, float] = {}
 
@@ -272,6 +284,26 @@ def __init__(self, config: PlatformConfig):
             else os.getenv("PHOTON_MENTION_PATTERNS")
         )
 
+    def _hydrate_persistent_state(self) -> None:
+        state = self._photon_state.load()
+        sent = state.get("sent_messages") or {}
+        if isinstance(sent, dict):
+            for message_id, meta in sent.items():
+                if not isinstance(message_id, str):
+                    continue
+                ts = time.time()
+                if isinstance(meta, dict):
+                    ts = _timestamp_from_iso(meta.get("sent_at")) or ts
+                self._sent_message_ids[message_id] = ts
+        inbound = state.get("last_inbound_by_chat") or {}
+        if isinstance(inbound, dict):
+            for chat_key, meta in inbound.items():
+                if not isinstance(chat_key, str) or not isinstance(meta, dict):
+                    continue
+                message_id = meta.get("message_id")
+                if isinstance(message_id, str) and message_id:
+                    self._last_inbound_by_chat[chat_key] = message_id
+
     # -- Group-mention gating (parity with BlueBubbles) -------------------
 
     @staticmethod
@@ -1122,7 +1154,13 @@ async def stop_typing(self, chat_id: str) -> None:
     _SENT_IDS_MAX = 1000
     _LAST_INBOUND_CHATS_MAX = 200
 
-    def _record_sent_message(self, message_id: Optional[str]) -> None:
+    def _record_sent_message(
+        self,
+        message_id: Optional[str],
+        *,
+        chat_id: Optional[str] = None,
+        kind: str = "text",
+    ) -> None:
         if not message_id:
             return
         sent = self._sent_message_ids
@@ -1132,6 +1170,13 @@ def _record_sent_message(self, message_id: Optional[str]) -> None:
         if len(sent) > self._SENT_IDS_MAX:
             for old in list(sent.keys())[: len(sent) - self._SENT_IDS_MAX]:
                 del sent[old]
+        chat_key = self._normalize_chat_key(chat_id) if chat_id else None
+        self._photon_state.record_sent_message(
+            message_id,
+            chat_key=chat_key,
+            space_id=chat_id,
+            kind=kind,
+        )
 
     # A DM space is addressable two ways — the chat GUID (`any;-;+1555...`)
     # that inbound events carry, and the bare E.164 phone that home-channel
@@ -1160,6 +1205,7 @@ def _record_last_inbound(
                 : len(last) - self._LAST_INBOUND_CHATS_MAX
             ]:
                 del last[old]
+        self._photon_state.record_last_inbound(key, message_id, space_id=chat_id)
 
     def _reactions_enabled(self) -> bool:
         return os.getenv("PHOTON_REACTIONS", "false").strip().lower() in {
@@ -1170,29 +1216,101 @@ async def _add_reaction(
         self, chat_id: str, message_id: str, emoji: str
     ) -> bool:
         """Tapback ``emoji`` onto a message. Soft-fails (False), never raises."""
+        chat_key = self._normalize_chat_key(chat_id)
+        self._photon_state.record_audit(
+            action="react",
+            status="started",
+            chat_key=chat_key,
+            message_id=message_id,
+        )
         try:
-            await self._sidecar_call(
+            data = await self._sidecar_call(
                 "/react",
                 {"spaceId": chat_id, "messageId": message_id, "emoji": emoji},
             )
+            reaction_id = data.get("reactionId")
+            self._photon_state.record_reaction_added(
+                chat_key,
+                message_id,
+                emoji,
+                str(reaction_id) if reaction_id else None,
+            )
+            self._photon_state.record_audit(
+                action="react",
+                status="succeeded",
+                chat_key=chat_key,
+                message_id=message_id,
+                reaction_id=str(reaction_id) if reaction_id else None,
+            )
             return True
         except Exception as e:
+            self._photon_state.record_audit(
+                action="react",
+                status="failed",
+                chat_key=chat_key,
+                message_id=message_id,
+                error_class=e.__class__.__name__,
+                error=e,
+            )
             logger.debug("[photon] add_reaction failed: %s", e)
             return False
 
     async def _remove_reaction(self, chat_id: str, message_id: str) -> bool:
         """Retract our tapback from a message. Soft-fails (False), never raises.
 
-        The sidecar tracks one reaction handle per target message; after a
-        sidecar restart the handle is gone and removal is best-effort (the
-        stale tapback self-heals when the next reaction replaces it).
+        The sidecar tracks one live reaction handle per target message. Hermes
+        also persists the returned reaction id so a restarted sidecar can try
+        rehydrating and unsending the reaction message.
         """
+        chat_key = self._normalize_chat_key(chat_id)
+        slot = self._photon_state.reaction_for(chat_key, message_id)
+        if slot is None and chat_key != chat_id:
+            slot = self._photon_state.reaction_for(chat_id, message_id)
+        reaction_id = slot.get("reaction_id") if isinstance(slot, dict) else None
+        self._photon_state.record_audit(
+            action="unreact",
+            status="started",
+            chat_key=chat_key,
+            message_id=message_id,
+            reaction_id=str(reaction_id) if reaction_id else None,
+        )
         try:
             await self._sidecar_call(
-                "/unreact", {"spaceId": chat_id, "messageId": message_id},
+                "/unreact",
+                {
+                    "spaceId": chat_id,
+                    "messageId": message_id,
+                    "reactionId": str(reaction_id) if reaction_id else None,
+                },
+            )
+            self._photon_state.record_reaction_removed(
+                chat_key, message_id, succeeded=True
+            )
+            if chat_key != chat_id:
+                self._photon_state.record_reaction_removed(
+                    chat_id, message_id, succeeded=True
+                )
+            self._photon_state.record_audit(
+                action="unreact",
+                status="succeeded",
+                chat_key=chat_key,
+                message_id=message_id,
+                reaction_id=str(reaction_id) if reaction_id else None,
             )
             return True
         except Exception as e:
+            self._photon_state.record_reaction_removed(
+                chat_key, message_id, succeeded=False
+            )
+            self._photon_state.record_audit(
+                action="unreact",
+                status="failed",
+                chat_key=chat_key,
+                message_id=message_id,
+                reaction_id=str(reaction_id) if reaction_id else None,
+                error_class=e.__class__.__name__,
+                error=e,
+            )
             logger.debug("[photon] remove_reaction failed: %s", e)
             return False
 
@@ -1256,9 +1374,9 @@ async def on_processing_start(self, event: MessageEvent) -> None:
         """Tapback 👀 on the triggering message while the agent works."""
         if not self._reactions_enabled():
             return
-        chat_id = getattr(event.source, "chat_id", None)
-        message_id = getattr(event, "message_id", None)
-        if chat_id and message_id:
+        target = self._processing_reaction_target(event)
+        if target:
+            chat_id, message_id = target
             await self._add_reaction(chat_id, message_id, "\U0001f440")
 
     async def on_processing_complete(
@@ -1272,10 +1390,10 @@ async def on_processing_complete(
         """
         if not self._reactions_enabled():
             return
-        chat_id = getattr(event.source, "chat_id", None)
-        message_id = getattr(event, "message_id", None)
-        if not chat_id or not message_id:
+        target = self._processing_reaction_target(event)
+        if not target:
             return
+        chat_id, message_id = target
         await self._remove_reaction(chat_id, message_id)
         if outcome == ProcessingOutcome.SUCCESS:
             await self._add_reaction(chat_id, message_id, "\U0001f44d")
@@ -1283,6 +1401,25 @@ async def on_processing_complete(
             await self._add_reaction(chat_id, message_id, "\U0001f44e")
         # CANCELLED: leave the message unreacted.
 
+    @staticmethod
+    def _processing_reaction_target(
+        event: MessageEvent,
+    ) -> Optional[tuple[str, str]]:
+        """Return the message target for processing tapbacks, if reactable."""
+        chat_id = getattr(event.source, "chat_id", None)
+        message_id = getattr(event, "message_id", None)
+        if not chat_id or not message_id:
+            return None
+        raw = getattr(event, "raw_message", None)
+        if isinstance(raw, dict):
+            content = raw.get("content")
+            if isinstance(content, dict) and content.get("type") == "reaction":
+                return None
+        text = getattr(event, "text", None)
+        if isinstance(text, str) and text.startswith("reaction:"):
+            return None
+        return str(chat_id), str(message_id)
+
     async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
         """Return whatever we know about a Spectrum space id.
 
@@ -1390,12 +1527,36 @@ async def _sidecar_send(self, space_id: str, text: str) -> SendResult:
         # keeps accepting the body during a half-upgraded restart.
         if _markdown_enabled():
             body["format"] = "markdown"
+        chat_key = self._normalize_chat_key(space_id)
+        self._photon_state.record_audit(
+            action="send",
+            status="started",
+            chat_key=chat_key,
+        )
         try:
             data = await self._sidecar_call("/send", body)
         except Exception as e:
+            self._photon_state.record_audit(
+                action="send",
+                status="failed",
+                chat_key=chat_key,
+                error_class=e.__class__.__name__,
+                error=e,
+            )
             return SendResult(success=False, error=str(e))
-        self._record_sent_message(data.get("messageId"))
-        return SendResult(success=True, message_id=data.get("messageId"))
+        message_id = data.get("messageId")
+        self._record_sent_message(
+            message_id,
+            chat_id=space_id,
+            kind="markdown" if body.get("format") == "markdown" else "text",
+        )
+        self._photon_state.record_audit(
+            action="send",
+            status="succeeded",
+            chat_key=chat_key,
+            message_id=message_id,
+        )
+        return SendResult(success=True, message_id=message_id)
 
     async def _sidecar_send_attachment(
         self,
@@ -1420,6 +1581,13 @@ async def _sidecar_send_attachment(
         # send_*_file / cron callers may pass arbitrary strings.
         safe_path = self.validate_media_delivery_path(str(path))
         if not safe_path:
+            self._photon_state.record_audit(
+                action="send",
+                status="failed",
+                chat_key=self._normalize_chat_key(space_id),
+                error_class="ValueError",
+                error=f"unsafe or missing attachment path: {path}",
+            )
             return SendResult(
                 success=False, error=f"unsafe or missing attachment path: {path}"
             )
@@ -1439,12 +1607,32 @@ async def _sidecar_send_attachment(
             body["mimeType"] = mime_type
         if caption:
             body["caption"] = caption
+        chat_key = self._normalize_chat_key(space_id)
+        self._photon_state.record_audit(
+            action="send",
+            status="started",
+            chat_key=chat_key,
+        )
         try:
             data = await self._sidecar_call("/send-attachment", body)
         except Exception as e:
+            self._photon_state.record_audit(
+                action="send",
+                status="failed",
+                chat_key=chat_key,
+                error_class=e.__class__.__name__,
+                error=e,
+            )
             return SendResult(success=False, error=str(e))
-        self._record_sent_message(data.get("messageId"))
-        return SendResult(success=True, message_id=data.get("messageId"))
+        message_id = data.get("messageId")
+        self._record_sent_message(message_id, chat_id=space_id, kind=body["kind"])
+        self._photon_state.record_audit(
+            action="send",
+            status="succeeded",
+            chat_key=chat_key,
+            message_id=message_id,
+        )
+        return SendResult(success=True, message_id=message_id)
 
     async def _sidecar_call(self, path: str, body: Dict[str, Any]) -> Dict[str, Any]:
         # Guard: adapter not yet connected (no sidecar address known).
@@ -1580,12 +1768,48 @@ async def _standalone_send(
 ) -> Dict[str, Any]:
     if not HTTPX_AVAILABLE:
         return {"error": "httpx not installed"}
+    state = PhotonStateStore()
+    state.load()
+    chat_key = PhotonAdapter._normalize_chat_key(chat_id) if chat_id else chat_id
+
+    def _audit(
+        status: str,
+        *,
+        message_id: Optional[str] = None,
+        error: Any = None,
+        error_class: Optional[str] = None,
+    ) -> None:
+        state.record_audit(
+            action="send",
+            status=status,
+            chat_key=chat_key,
+            message_id=message_id,
+            error=error,
+            error_class=error_class,
+        )
+
+    def _record_success(data: Dict[str, Any], *, kind: str) -> Optional[str]:
+        message_id = data.get("messageId")
+        state.record_sent_message(
+            message_id,
+            chat_key=chat_key,
+            space_id=chat_id,
+            kind=kind,
+        )
+        _audit("succeeded", message_id=message_id)
+        return message_id if isinstance(message_id, str) else None
+
     port = _coerce_port(
         (pconfig.extra or {}).get("sidecar_port") or os.getenv("PHOTON_SIDECAR_PORT"),
         _DEFAULT_SIDECAR_PORT,
     )
     token = os.getenv("PHOTON_SIDECAR_TOKEN")
     if not token:
+        _audit(
+            "failed",
+            error_class="RuntimeError",
+            error="missing PHOTON_SIDECAR_TOKEN",
+        )
         return {
             "error": (
                 "Photon standalone send requires a running sidecar with "
@@ -1606,15 +1830,23 @@ async def _standalone_send(
                 }
                 if _markdown_enabled():
                     send_body["format"] = "markdown"
+                _audit("started")
                 resp = await client.post(
                     f"{base}/send", json=send_body, headers=headers,
                 )
                 if resp.status_code != 200:
-                    return {"error": f"sidecar returned {resp.status_code}: {resp.text[:200]}"}
+                    error = f"sidecar returned {resp.status_code}: {resp.text[:200]}"
+                    _audit("failed", error_class="RuntimeError", error=error)
+                    return {"error": error}
                 data = resp.json() or {}
                 if not data.get("ok"):
-                    return {"error": data.get("error") or "sidecar reported failure"}
-                last_message_id = data.get("messageId")
+                    error = data.get("error") or "sidecar reported failure"
+                    _audit("failed", error_class="RuntimeError", error=error)
+                    return {"error": error}
+                last_message_id = _record_success(
+                    data,
+                    kind="markdown" if send_body.get("format") == "markdown" else "text",
+                )
 
             # 2. Each attachment as a separate /send-attachment call.
             #    media_files is List[Tuple[path, is_voice]] (see
@@ -1625,6 +1857,11 @@ async def _standalone_send(
                 safe_path = BasePlatformAdapter.validate_media_delivery_path(str(media_path))
                 if not safe_path:
                     logger.warning("[photon] standalone send skipping unsafe path")
+                    _audit(
+                        "failed",
+                        error_class="ValueError",
+                        error=f"unsafe or missing attachment path: {media_path}",
+                    )
                     continue
                 guessed, _ = mimetypes.guess_type(safe_path)
                 att_body: Dict[str, Any] = {
@@ -1634,18 +1871,26 @@ async def _standalone_send(
                 }
                 if guessed:
                     att_body["mimeType"] = guessed
+                _audit("started")
                 resp = await client.post(
                     f"{base}/send-attachment", json=att_body, headers=headers,
                 )
                 if resp.status_code != 200:
-                    return {"error": f"sidecar returned {resp.status_code}: {resp.text[:200]}"}
+                    error = f"sidecar returned {resp.status_code}: {resp.text[:200]}"
+                    _audit("failed", error_class="RuntimeError", error=error)
+                    return {"error": error}
                 data = resp.json() or {}
                 if not data.get("ok"):
-                    return {"error": data.get("error") or "sidecar reported failure"}
-                last_message_id = data.get("messageId") or last_message_id
+                    error = data.get("error") or "sidecar reported failure"
+                    _audit("failed", error_class="RuntimeError", error=error)
+                    return {"error": error}
+                last_message_id = (
+                    _record_success(data, kind=att_body["kind"]) or last_message_id
+                )
 
         return {"success": True, "message_id": last_message_id}
     except Exception as e:
+        _audit("failed", error_class=e.__class__.__name__, error=e)
         return {"error": f"Photon standalone send failed: {e}"}
 
 
diff --git a/plugins/platforms/photon/cli.py b/plugins/platforms/photon/cli.py
index 89e1c6bc8bc4..59a98fe83fb5 100644
--- a/plugins/platforms/photon/cli.py
+++ b/plugins/platforms/photon/cli.py
@@ -314,9 +314,44 @@ def _cmd_status(_args: argparse.Namespace) -> int:
     print(f"  node binary         : {node_bin or '✗ missing (install Node 18+)'}")
     print(f"  sidecar deps        : {'✓ installed' if sidecar_installed else '✗ run `hermes photon install-sidecar`'}")
     print(f"  telemetry           : {'on' if _telemetry_enabled() else 'off'} (`hermes photon telemetry on|off`)")
+    _print_state_summary(print)
     return 0
 
 
+def _print_state_summary(emit) -> None:
+    from .state import PhotonStateStore
+
+    store = PhotonStateStore()
+    store.load()
+    health = store.health()
+    emit(f"  state file          : {health['path']}")
+    if health.get("load_error"):
+        emit(
+            "  state unavailable/corrupt; using empty runtime state "
+            f"({health['load_error']})"
+        )
+    if health.get("write_error"):
+        emit(f"  state write failure : {health['write_error']}")
+    emit(f"  state schema        : v{health['schema_version']}")
+    emit(
+        "  state counts        : "
+        f"{health['sent_messages']} sent, "
+        f"{health['last_inbound_chats']} inbound chats, "
+        f"{health['active_reactions']} active reactions, "
+        f"{health['audit_entries']} audit entries"
+    )
+    failure = health.get("last_failure")
+    if isinstance(failure, dict):
+        parts = [
+            str(failure.get("at") or "unknown time"),
+            str(failure.get("action") or "unknown action"),
+            "failed",
+        ]
+        if failure.get("error"):
+            parts.append(f"({failure['error']})")
+        emit(f"  last state failure  : {' '.join(parts)}")
+
+
 def _refresh_status_numbers() -> None:
     phone, assigned = photon_auth.load_user_numbers()
     if phone and assigned:
diff --git a/plugins/platforms/photon/state.py b/plugins/platforms/photon/state.py
new file mode 100644
index 000000000000..2d7b17ee2af5
--- /dev/null
+++ b/plugins/platforms/photon/state.py
@@ -0,0 +1,370 @@
+"""Persistent Photon correlation state.
+
+This store intentionally keeps only bounded delivery/reaction metadata.  It is
+not a message archive and should never hold message bodies, attachment bytes, or
+Photon credentials.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+from hermes_constants import get_hermes_home
+from utils import atomic_json_write
+
+logger = logging.getLogger(__name__)
+
+SCHEMA_VERSION = 1
+DEFAULT_SENT_MAX = 1000
+DEFAULT_LAST_INBOUND_MAX = 200
+DEFAULT_REACTIONS_MAX = 512
+DEFAULT_AUDIT_MAX = 500
+DEFAULT_RETENTION_SECONDS = 48 * 3600
+
+
+def photon_state_path() -> Path:
+    return get_hermes_home() / "plugins" / "photon" / "state.json"
+
+
+def _now_iso() -> str:
+    return datetime.now(tz=timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def _parse_ts(value: Any) -> float:
+    if isinstance(value, (int, float)):
+        return float(value)
+    if not isinstance(value, str) or not value:
+        return 0.0
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
+    except ValueError:
+        return 0.0
+
+
+def _short_error(error: Any, *, limit: int = 300) -> Optional[str]:
+    if error is None:
+        return None
+    text = str(error)
+    return text[:limit]
+
+
+class PhotonStateStore:
+    """Small plugin-local JSON store for Photon send/reaction correlation."""
+
+    def __init__(
+        self,
+        path: Optional[Path] = None,
+        *,
+        sent_max: int = DEFAULT_SENT_MAX,
+        last_inbound_max: int = DEFAULT_LAST_INBOUND_MAX,
+        reactions_max: int = DEFAULT_REACTIONS_MAX,
+        audit_max: int = DEFAULT_AUDIT_MAX,
+        retention_seconds: int = DEFAULT_RETENTION_SECONDS,
+    ) -> None:
+        self.path = path or photon_state_path()
+        self.sent_max = sent_max
+        self.last_inbound_max = last_inbound_max
+        self.reactions_max = reactions_max
+        self.audit_max = audit_max
+        self.retention_seconds = retention_seconds
+        self.load_error: Optional[str] = None
+        self.write_error: Optional[str] = None
+        self._state = self._empty_state()
+
+    @staticmethod
+    def _empty_state() -> Dict[str, Any]:
+        return {
+            "schema_version": SCHEMA_VERSION,
+            "updated_at": None,
+            "sent_messages": {},
+            "last_inbound_by_chat": {},
+            "reactions": {},
+            "audit": [],
+        }
+
+    def load(self) -> Dict[str, Any]:
+        self.load_error = None
+        if not self.path.exists():
+            self._state = self._empty_state()
+            return self.snapshot()
+        try:
+            raw = self.path.read_text(encoding="utf-8")
+            payload = json.loads(raw)
+            if not isinstance(payload, dict):
+                raise ValueError("state root is not an object")
+            if payload.get("schema_version") != SCHEMA_VERSION:
+                raise ValueError("unsupported schema version")
+            self._state = self._normalize(payload)
+        except Exception as exc:
+            self.load_error = str(exc)
+            logger.warning(
+                "[photon] ignoring unreadable persistent state at %s: %s",
+                self.path,
+                exc,
+            )
+            self._state = self._empty_state()
+        return self.snapshot()
+
+    def snapshot(self) -> Dict[str, Any]:
+        return {
+            "schema_version": self._state["schema_version"],
+            "updated_at": self._state.get("updated_at"),
+            "sent_messages": dict(self._state["sent_messages"]),
+            "last_inbound_by_chat": dict(self._state["last_inbound_by_chat"]),
+            "reactions": dict(self._state["reactions"]),
+            "audit": list(self._state["audit"]),
+            "path": str(self.path),
+            "load_error": self.load_error,
+            "write_error": self.write_error,
+        }
+
+    def health(self) -> Dict[str, Any]:
+        state = self.snapshot()
+        reactions = state["reactions"]
+        failures = [
+            item for item in state["audit"]
+            if isinstance(item, dict) and item.get("status") == "failed"
+        ]
+        return {
+            "path": state["path"],
+            "schema_version": state["schema_version"],
+            "load_error": state["load_error"],
+            "write_error": state["write_error"],
+            "sent_messages": len(state["sent_messages"]),
+            "last_inbound_chats": len(state["last_inbound_by_chat"]),
+            "active_reactions": sum(
+                1 for item in reactions.values()
+                if isinstance(item, dict) and not item.get("removed_at")
+            ),
+            "audit_entries": len(state["audit"]),
+            "last_failure": failures[-1] if failures else None,
+        }
+
+    def record_sent_message(
+        self,
+        message_id: Optional[str],
+        *,
+        chat_key: Optional[str] = None,
+        space_id: Optional[str] = None,
+        kind: str = "text",
+    ) -> None:
+        if not message_id:
+            return
+        self._state["sent_messages"][str(message_id)] = {
+            "chat_key": chat_key,
+            "space_id": space_id,
+            "sent_at": _now_iso(),
+            "kind": kind,
+        }
+        self._persist()
+
+    def record_last_inbound(
+        self,
+        chat_key: Optional[str],
+        message_id: Optional[str],
+        *,
+        space_id: Optional[str] = None,
+    ) -> None:
+        if not chat_key or not message_id:
+            return
+        self._state["last_inbound_by_chat"][str(chat_key)] = {
+            "message_id": str(message_id),
+            "space_id": space_id,
+            "seen_at": _now_iso(),
+        }
+        self._persist()
+
+    def record_reaction_added(
+        self,
+        space_id: Optional[str],
+        message_id: Optional[str],
+        emoji: str,
+        reaction_id: Optional[str],
+    ) -> None:
+        if not space_id or not message_id:
+            return
+        key = self.reaction_key(space_id, message_id)
+        self._state["reactions"][key] = {
+            "reaction_id": reaction_id,
+            "emoji": emoji,
+            "created_at": _now_iso(),
+            "removed_at": None,
+        }
+        self._persist()
+
+    def record_reaction_removed(
+        self,
+        space_id: Optional[str],
+        message_id: Optional[str],
+        *,
+        succeeded: bool,
+    ) -> None:
+        if not space_id or not message_id:
+            return
+        key = self.reaction_key(space_id, message_id)
+        slot = self._state["reactions"].get(key)
+        if not isinstance(slot, dict):
+            return
+        if succeeded:
+            slot["removed_at"] = _now_iso()
+        self._persist()
+
+    def reaction_for(
+        self, space_id: Optional[str], message_id: Optional[str]
+    ) -> Optional[Dict[str, Any]]:
+        if not space_id or not message_id:
+            return None
+        slot = self._state["reactions"].get(self.reaction_key(space_id, message_id))
+        if not isinstance(slot, dict) or slot.get("removed_at"):
+            return None
+        return dict(slot)
+
+    def record_audit(
+        self,
+        *,
+        action: str,
+        status: str,
+        chat_key: Optional[str] = None,
+        message_id: Optional[str] = None,
+        reaction_id: Optional[str] = None,
+        error_class: Optional[str] = None,
+        error: Any = None,
+    ) -> None:
+        self._state["audit"].append({
+            "at": _now_iso(),
+            "action": action,
+            "status": status,
+            "chat_key": chat_key,
+            "message_id": message_id,
+            "reaction_id": reaction_id,
+            "error_class": error_class,
+            "error": _short_error(error),
+        })
+        self._persist()
+
+    @staticmethod
+    def reaction_key(space_id: str, message_id: str) -> str:
+        return f"{space_id}\0{message_id}"
+
+    def _persist(self) -> None:
+        self.write_error = None
+        self._state = self._normalize(self._state)
+        self._state["updated_at"] = _now_iso()
+        try:
+            atomic_json_write(
+                self.path,
+                self._state,
+                indent=2,
+                mode=0o600,
+                sort_keys=True,
+            )
+        except Exception as exc:
+            self.write_error = str(exc)
+            logger.warning("[photon] failed to persist state at %s: %s", self.path, exc)
+
+    def _normalize(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        now = time.time()
+        cutoff = now - self.retention_seconds
+        state = self._empty_state()
+        state["updated_at"] = payload.get("updated_at")
+
+        sent = payload.get("sent_messages")
+        if isinstance(sent, dict):
+            for message_id, value in sent.items():
+                if not isinstance(message_id, str) or not isinstance(value, dict):
+                    continue
+                ts = _parse_ts(value.get("sent_at"))
+                if ts and ts < cutoff:
+                    continue
+                state["sent_messages"][message_id] = {
+                    "chat_key": _string_or_none(value.get("chat_key")),
+                    "space_id": _string_or_none(value.get("space_id")),
+                    "sent_at": value.get("sent_at") if isinstance(value.get("sent_at"), str) else _now_iso(),
+                    "kind": _string_or_none(value.get("kind")) or "text",
+                }
+
+        inbound = payload.get("last_inbound_by_chat")
+        if isinstance(inbound, dict):
+            for chat_key, value in inbound.items():
+                if not isinstance(chat_key, str) or not isinstance(value, dict):
+                    continue
+                message_id = _string_or_none(value.get("message_id"))
+                if not message_id:
+                    continue
+                ts = _parse_ts(value.get("seen_at"))
+                if ts and ts < cutoff:
+                    continue
+                state["last_inbound_by_chat"][chat_key] = {
+                    "message_id": message_id,
+                    "space_id": _string_or_none(value.get("space_id")),
+                    "seen_at": value.get("seen_at") if isinstance(value.get("seen_at"), str) else _now_iso(),
+                }
+
+        reactions = payload.get("reactions")
+        if isinstance(reactions, dict):
+            for key, value in reactions.items():
+                if not isinstance(key, str) or not isinstance(value, dict):
+                    continue
+                created_ts = _parse_ts(value.get("created_at"))
+                removed_ts = _parse_ts(value.get("removed_at"))
+                newest_ts = max(created_ts, removed_ts)
+                if newest_ts and newest_ts < cutoff:
+                    continue
+                state["reactions"][key] = {
+                    "reaction_id": _string_or_none(value.get("reaction_id")),
+                    "emoji": _string_or_none(value.get("emoji")) or "",
+                    "created_at": value.get("created_at") if isinstance(value.get("created_at"), str) else _now_iso(),
+                    "removed_at": value.get("removed_at") if isinstance(value.get("removed_at"), str) else None,
+                }
+
+        audit = payload.get("audit")
+        if isinstance(audit, list):
+            for item in audit:
+                if not isinstance(item, dict):
+                    continue
+                at = item.get("at") if isinstance(item.get("at"), str) else _now_iso()
+                state["audit"].append({
+                    "at": at,
+                    "action": _string_or_none(item.get("action")) or "unknown",
+                    "status": _string_or_none(item.get("status")) or "unknown",
+                    "chat_key": _string_or_none(item.get("chat_key")),
+                    "message_id": _string_or_none(item.get("message_id")),
+                    "reaction_id": _string_or_none(item.get("reaction_id")),
+                    "error_class": _string_or_none(item.get("error_class")),
+                    "error": _short_error(item.get("error")),
+                })
+
+        state["sent_messages"] = _trim_mapping(
+            state["sent_messages"], self.sent_max, "sent_at"
+        )
+        state["last_inbound_by_chat"] = _trim_mapping(
+            state["last_inbound_by_chat"], self.last_inbound_max, "seen_at"
+        )
+        state["reactions"] = _trim_mapping(
+            state["reactions"], self.reactions_max, "created_at"
+        )
+        state["audit"] = state["audit"][-self.audit_max:]
+        return state
+
+
+def _string_or_none(value: Any) -> Optional[str]:
+    if value is None:
+        return None
+    text = str(value)
+    return text if text else None
+
+
+def _trim_mapping(
+    mapping: Dict[str, Dict[str, Any]], max_items: int, timestamp_key: str
+) -> Dict[str, Dict[str, Any]]:
+    if len(mapping) <= max_items:
+        return dict(mapping)
+    items = sorted(
+        mapping.items(),
+        key=lambda item: _parse_ts(item[1].get(timestamp_key)),
+    )
+    return dict(items[-max_items:])
diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py
index 0df9790c31ab..f1f13a019538 100644
--- a/tests/hermes_cli/test_plugins.py
+++ b/tests/hermes_cli/test_plugins.py
@@ -15,6 +15,7 @@
     PluginContext,
     PluginManager,
     PluginManifest,
+    LoadedPlugin,
     get_plugin_command_handler,
     get_plugin_commands,
     get_pre_tool_call_block_message,
@@ -484,6 +485,32 @@ def test_force_rediscover_clears_all_plugin_registries(self, monkeypatch):
         assert mgr._aux_tasks == {}
         assert mgr._slack_action_handlers == []
 
+    def test_deferred_bundled_platform_cli_loads_on_matching_command(self, tmp_path, monkeypatch):
+        mgr = PluginManager()
+        manifest = PluginManifest(
+            name="photon-platform",
+            version="0.1.0",
+            source="bundled",
+            path=tmp_path,
+            kind="platform",
+            key="platforms/photon",
+        )
+        loaded = LoadedPlugin(manifest=manifest, enabled=True, deferred=True)
+        mgr._plugins[manifest.key] = loaded
+        mgr._discovered = True
+        loaded_names: list[str] = []
+
+        def fake_load_plugin(item):
+            loaded_names.append(item.name)
+            mgr._cli_commands["photon"] = {"name": "photon", "plugin": item.name}
+
+        monkeypatch.setattr(mgr, "_load_plugin", fake_load_plugin)
+
+        mgr.discover_and_load(cli_command="photon")
+
+        assert loaded_names == ["photon-platform"]
+        assert "photon" in mgr._cli_commands
+
 
 # ── TestPluginLoading ──────────────────────────────────────────────────────
 
diff --git a/tests/plugins/platforms/photon/test_markdown.py b/tests/plugins/platforms/photon/test_markdown.py
index 6e803d653179..560daf7205a2 100644
--- a/tests/plugins/platforms/photon/test_markdown.py
+++ b/tests/plugins/platforms/photon/test_markdown.py
@@ -13,6 +13,7 @@
 from gateway.config import PlatformConfig
 from plugins.platforms.photon import adapter as photon_adapter
 from plugins.platforms.photon.adapter import PhotonAdapter
+from plugins.platforms.photon.state import PhotonStateStore
 
 _MD = "**bold** and `code`"
 
@@ -127,3 +128,6 @@ async def post(self, url: str, json: Dict[str, Any], headers=None):
 
     assert result.get("success") is True
     assert posted[0][1]["format"] == "markdown"
+    state = PhotonStateStore().load()
+    assert state["sent_messages"]["m-9"]["chat_key"] == "+15551234567"
+    assert state["sent_messages"]["m-9"]["kind"] == "markdown"
diff --git a/tests/plugins/platforms/photon/test_outbound_media.py b/tests/plugins/platforms/photon/test_outbound_media.py
index 09d4402a148d..64ab6cd3f213 100644
--- a/tests/plugins/platforms/photon/test_outbound_media.py
+++ b/tests/plugins/platforms/photon/test_outbound_media.py
@@ -16,6 +16,7 @@
 from gateway.config import PlatformConfig
 from plugins.platforms.photon import adapter as photon_adapter
 from plugins.platforms.photon.adapter import PhotonAdapter
+from plugins.platforms.photon.state import PhotonStateStore
 
 
 def _make_adapter(monkeypatch: pytest.MonkeyPatch) -> PhotonAdapter:
@@ -240,7 +241,7 @@ async def post(self, url: str, json: Dict[str, Any], headers=None):
     cfg = PlatformConfig(enabled=True, token="", extra={})
     result = await photon_adapter._standalone_send(
         cfg,
-        "any;-;+1",
+        "any;-;+15551234567",
         "hello",
         media_files=[(str(img), False)],
     )
@@ -253,3 +254,6 @@ async def post(self, url: str, json: Dict[str, Any], headers=None):
     assert posted[1][1]["path"] == str(img)
     assert posted[1][1]["kind"] == "attachment"
     assert posted[1][1]["mimeType"] == "image/png"
+    state = PhotonStateStore().load()
+    assert state["sent_messages"]["m-9"]["chat_key"] == "+15551234567"
+    assert state["sent_messages"]["m-9"]["kind"] == "attachment"
diff --git a/tests/plugins/platforms/photon/test_reactions.py b/tests/plugins/platforms/photon/test_reactions.py
index adf356a2779e..0d56e23237e2 100644
--- a/tests/plugins/platforms/photon/test_reactions.py
+++ b/tests/plugins/platforms/photon/test_reactions.py
@@ -67,6 +67,26 @@ def _message_event(adapter: PhotonAdapter) -> MessageEvent:
     )
 
 
+def _reaction_message_event(adapter: PhotonAdapter) -> MessageEvent:
+    return MessageEvent(
+        text="reaction:added:\U0001f44d",
+        message_type=MessageType.TEXT,
+        source=adapter.build_source(
+            chat_id="+15551234567",
+            chat_name="+15551234567",
+            chat_type="dm",
+            user_id="+15551234567",
+            user_name=None,
+        ),
+        message_id="reaction-evt-1",
+        reply_to_message_id="bot-msg-1",
+        reply_to_text="the bot's earlier reply",
+        reply_to_is_own_message=True,
+        raw_message=_reaction_event(emoji="\U0001f44d"),
+        timestamp=datetime.now(tz=timezone.utc),
+    )
+
+
 def _reaction_event(
     emoji: str = "❤️",
     target_id: str = "bot-msg-1",
@@ -123,7 +143,14 @@ async def test_remove_reaction_posts_unreact(monkeypatch: pytest.MonkeyPatch) ->
 
     assert ok is True
     assert calls == [
-        ("/unreact", {"spaceId": "+15551234567", "messageId": "target-msg-1"})
+        (
+            "/unreact",
+            {
+                "spaceId": "+15551234567",
+                "messageId": "target-msg-1",
+                "reactionId": None,
+            },
+        )
     ]
 
 
@@ -217,6 +244,21 @@ async def test_processing_cancelled_only_removes(
     assert [path for path, _ in calls] == ["/unreact"]
 
 
+@pytest.mark.asyncio
+async def test_processing_hooks_skip_inbound_reaction_events(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("PHOTON_REACTIONS", "true")
+    adapter = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(adapter)
+    event = _reaction_message_event(adapter)
+
+    await adapter.on_processing_start(event)
+    await adapter.on_processing_complete(event, ProcessingOutcome.SUCCESS)
+
+    assert calls == []
+
+
 # -- Inbound reaction routing ------------------------------------------------
 
 @pytest.mark.asyncio
@@ -275,6 +317,106 @@ async def test_inbound_reaction_sent_ids_fallback(
     assert len(captured) == 1
 
 
+@pytest.mark.asyncio
+async def test_inbound_reaction_sent_ids_fallback_survives_restart(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Persisted sent ids still route tapbacks when provider direction is null."""
+    first = _make_adapter(monkeypatch)
+    first._record_sent_message("bot-msg-1", chat_id="+15551234567")
+
+    restarted = _make_adapter(monkeypatch)
+    captured = _capture_handled(restarted, monkeypatch)
+
+    await restarted._dispatch_inbound(
+        _reaction_event(target_id="bot-msg-1", target_direction=None)
+    )
+
+    assert len(captured) == 1
+
+
+@pytest.mark.asyncio
+async def test_latest_inbound_target_survives_restart_for_add_reaction(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    first = _make_adapter(monkeypatch)
+    _capture_handled(first, monkeypatch)
+    await first._dispatch_inbound({
+        "messageId": "inbound-before-restart",
+        "platform": "iMessage",
+        "space": {"id": "+15551234567", "type": "dm", "phone": "+15551234567"},
+        "sender": {"id": "+15551234567"},
+        "content": {"type": "text", "text": "hello"},
+        "timestamp": "2026-06-11T10:00:00.000Z",
+    })
+
+    restarted = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(restarted)
+
+    result = await restarted.add_reaction("+15551234567", _THUMBS_UP)
+
+    assert result == {"success": True, "message_id": "inbound-before-restart"}
+    assert calls[0] == (
+        "/react",
+        {
+            "spaceId": "+15551234567",
+            "messageId": "inbound-before-restart",
+            "emoji": _THUMBS_UP,
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_reaction_id_persisted_for_unreact_after_restart(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    first = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(first)
+
+    assert await first._add_reaction("+15551234567", "target-msg-1", _EYES)
+    assert calls[0][0] == "/react"
+
+    restarted = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(restarted)
+
+    assert await restarted._remove_reaction("+15551234567", "target-msg-1")
+    assert calls == [
+        (
+            "/unreact",
+            {
+                "spaceId": "+15551234567",
+                "messageId": "target-msg-1",
+                "reactionId": "react-1",
+            },
+        )
+    ]
+
+
+@pytest.mark.asyncio
+async def test_persisted_reaction_lookup_accepts_dm_alias_after_restart(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    first = _make_adapter(monkeypatch)
+    _capture_sidecar(first)
+
+    assert await first._add_reaction("any;-;+15551234567", "target-msg-1", _EYES)
+
+    restarted = _make_adapter(monkeypatch)
+    calls = _capture_sidecar(restarted)
+
+    assert await restarted._remove_reaction("+15551234567", "target-msg-1")
+    assert calls == [
+        (
+            "/unreact",
+            {
+                "spaceId": "+15551234567",
+                "messageId": "target-msg-1",
+                "reactionId": "react-1",
+            },
+        )
+    ]
+
+
 @pytest.mark.asyncio
 async def test_inbound_reaction_on_foreign_message_dropped(
     monkeypatch: pytest.MonkeyPatch,
diff --git a/tests/plugins/platforms/photon/test_state.py b/tests/plugins/platforms/photon/test_state.py
new file mode 100644
index 000000000000..a8771729bcd8
--- /dev/null
+++ b/tests/plugins/platforms/photon/test_state.py
@@ -0,0 +1,160 @@
+"""Persistent Photon state tests."""
+from __future__ import annotations
+
+import json
+import stat
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from plugins.platforms.photon.state import PhotonStateStore
+
+
+def _iso(seconds_ago: int = 0) -> str:
+    dt = datetime.now(tz=timezone.utc) - timedelta(seconds=seconds_ago)
+    return dt.isoformat().replace("+00:00", "Z")
+
+
+def test_load_missing_state_returns_empty(tmp_path: Path) -> None:
+    store = PhotonStateStore(tmp_path / "missing.json")
+
+    state = store.load()
+
+    assert state["sent_messages"] == {}
+    assert state["last_inbound_by_chat"] == {}
+    assert state["reactions"] == {}
+    assert state["audit"] == []
+    assert state["load_error"] is None
+
+
+def test_load_corrupt_state_fails_open(tmp_path: Path) -> None:
+    path = tmp_path / "state.json"
+    path.write_text("{not-json", encoding="utf-8")
+    store = PhotonStateStore(path)
+
+    state = store.load()
+
+    assert state["sent_messages"] == {}
+    assert state["load_error"]
+
+
+def test_record_methods_create_private_atomic_snapshot(tmp_path: Path) -> None:
+    path = tmp_path / "plugins" / "photon" / "state.json"
+    store = PhotonStateStore(path)
+    store.load()
+
+    store.record_sent_message(
+        "msg-1", chat_key="+15551234567", space_id="any;-;+15551234567"
+    )
+    store.record_last_inbound("+15551234567", "inbound-1", space_id="any;-;+1555")
+    store.record_reaction_added("any;-;+1555", "inbound-1", "like", "reaction-1")
+    store.record_audit(
+        action="send",
+        status="failed",
+        chat_key="+15551234567",
+        message_id="msg-1",
+        error_class="RuntimeError",
+        error="x" * 1000,
+    )
+
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    assert payload["sent_messages"]["msg-1"]["chat_key"] == "+15551234567"
+    assert payload["last_inbound_by_chat"]["+15551234567"]["message_id"] == "inbound-1"
+    assert store.reaction_for("any;-;+1555", "inbound-1")["reaction_id"] == "reaction-1"
+    assert len(payload["audit"][0]["error"]) == 300
+    assert stat.S_IMODE(path.stat().st_mode) == 0o600
+
+
+def test_state_prunes_by_count_and_age(tmp_path: Path) -> None:
+    path = tmp_path / "state.json"
+    payload = {
+        "schema_version": 1,
+        "updated_at": _iso(),
+        "sent_messages": {
+            "old": {"sent_at": _iso(7200), "kind": "text"},
+            "keep-1": {"sent_at": _iso(20), "kind": "text"},
+            "keep-2": {"sent_at": _iso(10), "kind": "text"},
+            "keep-3": {"sent_at": _iso(5), "kind": "text"},
+        },
+        "last_inbound_by_chat": {
+            "old-chat": {"message_id": "old", "seen_at": _iso(7200)},
+            "chat-1": {"message_id": "m1", "seen_at": _iso(10)},
+        },
+        "reactions": {},
+        "audit": [
+            {"at": _iso(3), "action": "send", "status": "started"},
+            {"at": _iso(2), "action": "send", "status": "succeeded"},
+            {"at": _iso(1), "action": "react", "status": "succeeded"},
+        ],
+    }
+    path.write_text(json.dumps(payload), encoding="utf-8")
+    store = PhotonStateStore(
+        path,
+        sent_max=2,
+        last_inbound_max=10,
+        audit_max=2,
+        retention_seconds=3600,
+    )
+
+    state = store.load()
+
+    assert set(state["sent_messages"]) == {"keep-2", "keep-3"}
+    assert set(state["last_inbound_by_chat"]) == {"chat-1"}
+    assert [item["action"] for item in state["audit"]] == ["send", "react"]
+
+
+def test_reaction_removed_is_not_returned(tmp_path: Path) -> None:
+    store = PhotonStateStore(tmp_path / "state.json")
+    store.load()
+    store.record_reaction_added("space", "message", "like", "reaction")
+
+    store.record_reaction_removed("space", "message", succeeded=True)
+
+    assert store.reaction_for("space", "message") is None
+    assert store.health()["active_reactions"] == 0
+
+
+def test_failed_reaction_removal_keeps_active_slot(tmp_path: Path) -> None:
+    store = PhotonStateStore(tmp_path / "state.json")
+    store.load()
+    store.record_reaction_added("space", "message", "like", "reaction")
+
+    store.record_reaction_removed("space", "message", succeeded=False)
+
+    assert store.reaction_for("space", "message")["reaction_id"] == "reaction"
+    assert store.health()["active_reactions"] == 1
+
+
+def test_state_does_not_persist_message_content_or_secrets(tmp_path: Path) -> None:
+    store = PhotonStateStore(tmp_path / "state.json")
+    store.load()
+
+    store.record_sent_message("msg-secret", chat_key="+1", space_id="space")
+    store.record_audit(
+        action="send",
+        status="failed",
+        chat_key="+1",
+        message_id="msg-secret",
+        error="sidecar rejected send",
+    )
+
+    raw = (tmp_path / "state.json").read_text(encoding="utf-8")
+    assert "hello world" not in raw
+    assert "PHOTON_PROJECT_SECRET" not in raw
+    assert "test-project-secret" not in raw
+    assert "attachment-bytes" not in raw
+    assert "msg-secret" in raw
+
+
+def test_write_failure_is_fail_open(tmp_path: Path, monkeypatch) -> None:
+    store = PhotonStateStore(tmp_path / "state.json")
+    store.load()
+
+    def boom(*_args, **_kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr("plugins.platforms.photon.state.atomic_json_write", boom)
+
+    store.record_sent_message("msg-1", chat_key="+1", space_id="space")
+
+    assert store.write_error == "disk full"
+    assert store.snapshot()["sent_messages"]["msg-1"]["space_id"] == "space"
diff --git a/tests/plugins/platforms/photon/test_status.py b/tests/plugins/platforms/photon/test_status.py
new file mode 100644
index 000000000000..799feb81ed65
--- /dev/null
+++ b/tests/plugins/platforms/photon/test_status.py
@@ -0,0 +1,73 @@
+"""Photon CLI status state-summary tests."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from plugins.platforms.photon import cli
+from plugins.platforms.photon.state import PhotonStateStore
+
+
+def test_status_prints_state_counts(tmp_path: Path, monkeypatch) -> None:
+    home = tmp_path / "hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    store = PhotonStateStore()
+    store.load()
+    store.record_sent_message("msg-1", chat_key="+1", space_id="space")
+    store.record_last_inbound("+1", "inbound-1", space_id="space")
+    store.record_reaction_added("space", "inbound-1", "like", "reaction-1")
+    store.record_audit(action="send", status="succeeded", chat_key="+1")
+    rendered: list[str] = []
+
+    cli._print_state_summary(rendered.append)
+
+    output = "\n".join(rendered)
+    assert "state schema        : v1" in output
+    assert "1 sent, 1 inbound chats, 1 active reactions, 1 audit entries" in output
+    assert str(home / "plugins" / "photon" / "state.json") in output
+
+
+def test_status_handles_corrupt_state(tmp_path: Path, monkeypatch) -> None:
+    home = tmp_path / "hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    path = home / "plugins" / "photon" / "state.json"
+    path.parent.mkdir(parents=True)
+    path.write_text("{bad", encoding="utf-8")
+    rendered: list[str] = []
+
+    cli._print_state_summary(rendered.append)
+
+    assert any("unavailable/corrupt" in line for line in rendered)
+
+
+def test_status_does_not_print_stored_message_text(tmp_path: Path, monkeypatch) -> None:
+    home = tmp_path / "hermes"
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    path = home / "plugins" / "photon" / "state.json"
+    path.parent.mkdir(parents=True)
+    path.write_text(
+        json.dumps({
+            "schema_version": 1,
+            "updated_at": "2026-06-30T00:00:00Z",
+            "sent_messages": {
+                "msg-1": {
+                    "chat_key": "+1",
+                    "space_id": "space",
+                    "sent_at": "2026-06-30T00:00:00Z",
+                    "kind": "text",
+                    "text": "do not show this",
+                }
+            },
+            "last_inbound_by_chat": {},
+            "reactions": {},
+            "audit": [],
+        }),
+        encoding="utf-8",
+    )
+    rendered: list[str] = []
+
+    cli._print_state_summary(rendered.append)
+
+    output = "\n".join(rendered)
+    assert "do not show this" not in output
+    assert "msg-1" not in output