feat: add context-engine runtime identity diagnostics

agent/context_engine.py

@@ -26,9 +26,53 @@
 """
 
 from abc import ABC, abstractmethod
+import inspect
+from pathlib import Path
 from typing import Any, Dict, List
 
 
+def describe_context_engine_origin(engine: "ContextEngine") -> Dict[str, Any]:
+    """Return best-effort origin metadata for the active context engine."""
+    info: Dict[str, Any] = {
+        "origin": "unknown",
+        "module": "",
+        "path": None,
+        "module_path": None,
+    }
+
+    if engine is None:
+        return info
+
+    module_name = getattr(engine.__class__, "__module__", "") or ""
+    info["module"] = module_name
+
+    stamped_origin = getattr(engine, "_hermes_context_engine_origin", "") or ""
+    stamped_path = getattr(engine, "_hermes_context_engine_path", None)
+    if stamped_origin:
+        info["origin"] = stamped_origin
+    if stamped_path:
+        info["path"] = str(stamped_path)
+
+    try:
+        class_file = Path(inspect.getfile(engine.__class__)).resolve()
+        info["module_path"] = str(class_file)
+        if info["path"] is None:
+            info["path"] = str(class_file)
+        if info["origin"] == "unknown":
+            parts = class_file.parts
+            if "plugins" in parts and "context_engine" in parts:
+                info["origin"] = "repo-shipped"
+            elif module_name.startswith("agent."):
+                info["origin"] = "built-in"
+    except Exception:
+        pass
+
+    if info["origin"] == "unknown" and getattr(engine, "name", "") == "compressor":
+        info["origin"] = "built-in"
+
+    return info
+
+
 class ContextEngine(ABC):
     """Base class all context engines must implement."""
 

hermes_cli/plugins.py

@@ -316,6 +316,12 @@ def register_context_engine(self, engine) -> None:
                 self.manifest.name,
             )
             return
+        try:
+            engine._hermes_context_engine_origin = f"plugin:{self.manifest.source or 'unknown'}"
+            if self.manifest.path:
+                engine._hermes_context_engine_path = self.manifest.path
+        except Exception:
+            pass
         self._manager._context_engine = engine
         logger.info(
             "Plugin '%s' registered context engine: %s",

plugins/context_engine/__init__.py

@@ -30,6 +30,18 @@
 _CONTEXT_ENGINE_PLUGINS_DIR = Path(__file__).parent
 
 
+def _stamp_engine_metadata(engine, *, origin: str, path: Path | None) -> Optional["ContextEngine"]:
+    if engine is None:
+        return None
+    try:
+        engine._hermes_context_engine_origin = origin
+        if path is not None:
+            engine._hermes_context_engine_path = str(path.resolve())
+    except Exception:
+        pass
+    return engine
+
+
 def discover_context_engines() -> List[Tuple[str, str, bool]]:
     """Scan plugins/context_engine/ for available engines.
 
@@ -178,7 +190,7 @@ def _load_engine_from_dir(engine_dir: Path) -> Optional["ContextEngine"]:
         try:
             mod.register(collector)
             if collector.engine:
-                return collector.engine
+                return _stamp_engine_metadata(collector.engine, origin="repo-shipped", path=engine_dir)
         except Exception as e:
             logger.debug("register() failed for %s: %s", name, e)
 
@@ -189,7 +201,7 @@ def _load_engine_from_dir(engine_dir: Path) -> Optional["ContextEngine"]:
         if (isinstance(attr, type) and issubclass(attr, ContextEngine)
                 and attr is not ContextEngine):
             try:
-                return attr()
+                return _stamp_engine_metadata(attr(), origin="repo-shipped", path=engine_dir)
             except Exception:
                 pass
 

run_agent.py

@@ -1480,6 +1480,7 @@ def __init__(
             _engine_name = _ctx_cfg.get("engine", "compressor") or "compressor"
         except Exception:
             pass
+        self._configured_context_engine_name = _engine_name
 
         if _engine_name != "compressor":
             # Try loading from plugins/context_engine/<name>/
@@ -2556,6 +2557,74 @@ def _run_review():
         t = threading.Thread(target=_run_review, daemon=True, name="bg-review")
         t.start()
 
+    def get_context_engine_runtime_identity(self) -> Dict[str, Any]:
+        """Return host ↔ context-engine runtime identity for diagnostics."""
+        from agent.context_engine import describe_context_engine_origin
+
+        engine = getattr(self, "context_compressor", None)
+        engine_status = {}
+        if engine is not None:
+            try:
+                engine_status = engine.get_status() or {}
+            except Exception:
+                engine_status = {}
+
+        lifecycle = engine_status.get("lifecycle") if isinstance(engine_status, dict) else None
+        lifecycle = lifecycle if isinstance(lifecycle, dict) else {}
+
+        plugin_session_id = (
+            lifecycle.get("current_session_id")
+            or engine_status.get("current_session_id")
+            or getattr(engine, "_session_id", None)
+        )
+        plugin_conversation_id = (
+            lifecycle.get("conversation_id")
+            or engine_status.get("conversation_id")
+            or getattr(engine, "_conversation_id", None)
+        )
+        plugin_last_finalized_session_id = (
+            lifecycle.get("last_finalized_session_id")
+            or engine_status.get("last_finalized_session_id")
+        )
+
+        state_db_known = None
+        if getattr(self, "_session_db", None) is not None:
+            try:
+                state_db_known = self._session_db.get_session(self.session_id) is not None
+            except Exception:
+                state_db_known = None
+
+        warnings: list[str] = []
+        if state_db_known is False:
+            warnings.append("state_db_missing_host_session")
+        if plugin_session_id and plugin_session_id != self.session_id:
+            warnings.append("plugin_session_mismatch")
+
+        slash_commands: list[str] = []
+        try:
+            from hermes_cli.plugins import get_plugin_commands
+
+            slash_commands = sorted(get_plugin_commands().keys())
+        except Exception:
+            slash_commands = []
+
+        identity = describe_context_engine_origin(engine)
+        if getattr(engine, "name", "") == "lcm" and identity["origin"].startswith("plugin:") and "lcm" not in slash_commands:
+            warnings.append("missing_expected_slash_command:/lcm")
+
+        identity.update({
+            "configured_engine": getattr(self, "_configured_context_engine_name", getattr(engine, "name", "compressor")),
+            "active_engine": getattr(engine, "name", None),
+            "host_session_id": self.session_id,
+            "host_session_known_to_state_db": state_db_known,
+            "plugin_session_id": plugin_session_id,
+            "plugin_conversation_id": plugin_conversation_id,
+            "plugin_last_finalized_session_id": plugin_last_finalized_session_id,
+            "slash_commands": slash_commands,
+            "warnings": warnings,
+        })
+        return identity
+
     def _apply_persist_user_message_override(self, messages: List[Dict]) -> None:
         """Rewrite the current-turn user message before persistence/return.
 

tests/run_agent/test_context_engine_runtime_identity.py

@@ -0,0 +1,117 @@
+from unittest.mock import MagicMock, patch
+
+from agent.context_engine import ContextEngine
+
+
+class _StubEngine(ContextEngine):
+    @property
+    def name(self) -> str:
+        return "stub"
+
+    def update_from_response(self, usage):
+        pass
+
+    def should_compress(self, prompt_tokens=None):
+        return False
+
+    def compress(self, messages, current_tokens=None):
+        return messages
+
+
+class _StubLCMEngine(_StubEngine):
+    @property
+    def name(self) -> str:
+        return "lcm"
+
+
+def test_runtime_identity_reports_builtin_context_engine_and_known_state_db(tmp_path):
+    from hermes_state import SessionDB
+
+    db = SessionDB(db_path=tmp_path / "state.db")
+    cfg = {"context": {"engine": "compressor"}, "agent": {}}
+
+    with (
+        patch("hermes_cli.config.load_config", return_value=cfg),
+        patch("run_agent.get_tool_definitions", return_value=[]),
+        patch("run_agent.check_toolset_requirements", return_value={}),
+        patch("run_agent.OpenAI"),
+    ):
+        from run_agent import AIAgent
+
+        agent = AIAgent(
+            api_key="test-key-1234567890",
+            base_url="https://openrouter.ai/api/v1",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+            session_db=db,
+            session_id="host-session-1",
+        )
+
+    identity = agent.get_context_engine_runtime_identity()
+
+    assert identity["configured_engine"] == "compressor"
+    assert identity["active_engine"] == "compressor"
+    assert identity["origin"] == "built-in"
+    assert identity["host_session_id"] == "host-session-1"
+    assert identity["host_session_known_to_state_db"] is True
+    assert identity["plugin_session_id"] is None
+    assert identity["warnings"] == []
+
+
+def test_runtime_identity_reports_plugin_binding_mismatch_and_missing_lcm_command():
+    engine = _StubLCMEngine()
+    engine.update_model = MagicMock()
+    engine._hermes_context_engine_origin = "plugin:user"
+    engine._hermes_context_engine_path = "/tmp/hermes/plugins/hermes-lcm"
+    engine.get_status = MagicMock(return_value={
+        "engine": "lcm",
+        "conversation_id": "conv-123",
+        "lifecycle": {
+            "current_session_id": "plugin-session-9",
+            "last_finalized_session_id": "plugin-session-8",
+            "conversation_id": "conv-123",
+        },
+    })
+
+    cfg = {"context": {"engine": "lcm"}, "agent": {}}
+    fake_session_db = MagicMock()
+    fake_session_db.get_session.return_value = None
+
+    with (
+        patch("hermes_cli.config.load_config", return_value=cfg),
+        patch("plugins.context_engine.load_context_engine", return_value=None),
+        patch("hermes_cli.plugins.get_plugin_context_engine", return_value=engine),
+        patch("hermes_cli.plugins.get_plugin_commands", return_value={}),
+        patch("agent.model_metadata.get_model_context_length", return_value=131_072),
+        patch("run_agent.get_tool_definitions", return_value=[]),
+        patch("run_agent.check_toolset_requirements", return_value={}),
+        patch("run_agent.OpenAI"),
+    ):
+        from run_agent import AIAgent
+
+        agent = AIAgent(
+            model="openrouter/auto",
+            api_key="test-key-1234567890",
+            base_url="https://openrouter.ai/api/v1",
+            quiet_mode=True,
+            skip_context_files=True,
+            skip_memory=True,
+            session_db=fake_session_db,
+            session_id="host-session-2",
+        )
+
+    identity = agent.get_context_engine_runtime_identity()
+
+    assert identity["configured_engine"] == "lcm"
+    assert identity["active_engine"] == "lcm"
+    assert identity["origin"] == "plugin:user"
+    assert identity["path"] == "/tmp/hermes/plugins/hermes-lcm"
+    assert identity["host_session_id"] == "host-session-2"
+    assert identity["host_session_known_to_state_db"] is False
+    assert identity["plugin_session_id"] == "plugin-session-9"
+    assert identity["plugin_conversation_id"] == "conv-123"
+    assert identity["plugin_last_finalized_session_id"] == "plugin-session-8"
+    assert "state_db_missing_host_session" in identity["warnings"]
+    assert "plugin_session_mismatch" in identity["warnings"]
+    assert "missing_expected_slash_command:/lcm" in identity["warnings"]