feat(plugins): add pre_agent_dispatch plugin hook

gateway/run.py

@@ -21397,7 +21397,52 @@ def _approval_notify_sync(approval_data: dict) -> None:
                     _conversation_kwargs["moa_config"] = moa_config
                 if _persist_user_timestamp_override is not None:
                     _conversation_kwargs["persist_user_timestamp"] = _persist_user_timestamp_override
-                result = agent.run_conversation(_api_run_message, **_conversation_kwargs)
+
+                # ── pre_agent_dispatch plugin hook ──────────────────
+                from hermes_cli.plugins import dispatch_pre_agent as _dispatch_pre_agent
+
+                _hook_msg = (
+                    message if isinstance(message, str)
+                    else str(message)
+                )
+                _hook_kwargs: dict = dict(
+                    message=_hook_msg,
+                    session_key=session_key or "",
+                    source=source,
+                    gateway=self,
+                    history=list(agent_history),
+                )
+                # Pass the stream consumer's delta callback so router plugins
+                # can stream orchestrator output progressively instead of
+                # making the user wait in silence.
+                if _stream_consumer is not None:
+                    _hook_kwargs["stream_callback"] = _stream_consumer.on_delta
+                _hook_result = _dispatch_pre_agent(**_hook_kwargs)
+                _hook_action = _hook_result.get("action", "allow")
+
+                if _hook_action == "skip":
+                    result = {
+                        "final_response": "",
+                        "messages": list(agent_history),
+                        "interrupted": False,
+                    }
+                elif _hook_action == "route":
+                    result = {
+                        "final_response": _hook_result.get("result", ""),
+                        "messages": list(agent_history) + [
+                            {"role": "user", "content": _hook_msg},
+                            {"role": "assistant",
+                             "content": _hook_result.get("result", "")},
+                        ],
+                        "interrupted": False,
+                    }
+                else:
+                    if _hook_action == "rewrite":
+                        _api_run_message = _hook_result.get("text", _api_run_message)
+                    # ── Normal dispatch (allow, rewrite, or unrecognised) ─
+                    result = agent.run_conversation(
+                        _api_run_message, **_conversation_kwargs)
+                # ── End pre_agent_dispatch hook ──────────────────────
             finally:
                 unregister_gateway_notify(_approval_session_key)
                 # Cancel any pending clarify entries so blocked agent

hermes_cli/config.py

@@ -2364,6 +2364,27 @@ def _ensure_hermes_home_managed(home: Path):
         "subagent_auto_approve": False,
     },
 
+    # Flash → Orchestrator gateway-level router.
+    # Pre-turn classifier that routes complex tasks to the orchestrator profile.
+    # Simple tasks stay on the default Flash model — fast and cheap.
+    "router": {
+        "enabled": False,          # master on/off switch (opt-in)
+        "classifier": {
+            "task": "triage_specifier",  # aux task for the classification call
+            "model": "",                 # explicit override (empty = use task default)
+        },
+        "orchestrator": {
+            "profile": "orchestrator",   # profile name for heavy tasks
+            "timeout": 600,              # subprocess timeout in seconds
+            "pass_history": True,        # whether to pass session history
+        },
+        "rules": {
+            "always_simple": [],         # regex patterns that always stay local
+            "always_complex": [],        # regex patterns that always escalate
+            "threshold": 0.5,            # minimum confidence to route to Pro
+        },
+    },
+
     # Ephemeral prefill messages file — JSON list of {role, content} dicts
     # injected at the start of every API call for few-shot priming.
     # Never saved to sessions, logs, or trajectories.

hermes_cli/plugins.py

@@ -171,6 +171,16 @@ def _install_plugin_debug_handler(force: bool = False) -> None:
     #   {"action": "allow"}  /  None             -> normal dispatch
     # Kwargs: event: MessageEvent, gateway: GatewayRunner, session_store.
     "pre_gateway_dispatch",
+    # Pre-agent-dispatch hook. Fired once per turn just BEFORE the agent
+    # processes a user message (CLI, gateway, or TUI). Plugins may return
+    # a dict to influence flow:
+    #   {"action": "allow"}  /  None              -> normal agent dispatch
+    #   {"action": "skip",    "reason": "..."}    -> drop message (no reply)
+    #   {"action": "rewrite", "text": "..."}      -> replace message text, continue
+    #   {"action": "route",   "result": "..."}    -> bypass agent, use this text as final response
+    # Kwargs: message: str, session_key: str, source: SessionSource | None,
+    #   history: list[dict] | None, gateway: GatewayRunner | None.
+    "pre_agent_dispatch",
     # Approval lifecycle hooks. Fired by tools/approval.py when a dangerous
     # command needs an approval decision -- fires for CLI-interactive prompts,
     # gateway/ACP approvals, and smart-mode auxiliary-LLM decisions.
@@ -2464,3 +2474,76 @@ def get_plugin_toolsets() -> List[tuple]:
         result.append((ts_key, label, desc))
 
     return result
+
+
+# ── Shared helper for pre_agent_dispatch hook ─────────────────────────
+
+def dispatch_pre_agent(
+    message: str,
+    session_key: str = "",
+    source=None,
+    gateway=None,
+    history: list | None = None,
+    stream_callback=None,
+) -> dict:
+    """Run pre_agent_dispatch hooks and return a result dict.
+
+    Callers should check the ``action`` key:
+
+    ``{"action": "skip"}``
+        Drop the message completely — no reply, no agent run.
+    ``{"action": "route", "result": str}``
+        Bypass the agent entirely; use ``result`` as the final response.
+    ``{"action": "rewrite", "text": str}``
+        Replace the message text with ``text``, then run the agent normally.
+    ``{"action": "allow"}``
+        Normal agent dispatch (hooks had no comment or chose not to intercept).
+
+    On error (hook failure, import error, etc.) the function returns
+    ``{"action": "allow"}`` — fail-closed to normal dispatch.
+
+    When ``stream_callback`` is provided, it is forwarded to every hook as a
+    ``stream_callback`` kwarg.  A hook that produces a ``route`` result may
+    call the callback repeatedly during the routing subprocess so the user
+    sees progressive output while the orchestrator works.
+    """
+    try:
+        discover_plugins()
+        _hook_kwargs: dict = dict(
+            message=message,
+            session_key=session_key,
+            source=source,
+            history=history or [],
+            gateway=gateway,
+        )
+        if stream_callback is not None:
+            _hook_kwargs["stream_callback"] = stream_callback
+        _hook_results = invoke_hook(
+            "pre_agent_dispatch",
+            **_hook_kwargs,
+        )
+        for _hr in (_hook_results or []):
+            if not isinstance(_hr, dict):
+                continue
+            action = _hr.get("action")
+            if action == "skip":
+                return {"action": "skip"}
+            if action == "route":
+                result_text = _hr.get("result", "") or ""
+                result: dict = {"action": "route", "result": result_text}
+                # Pass through streamed flag so callers can avoid duplicating
+                # output that was already shown via stream_callback.
+                if _hr.get("streamed"):
+                    result["streamed"] = True
+                return result
+            if action == "rewrite":
+                new_text = _hr.get("text", "")
+                if isinstance(new_text, str) and new_text:
+                    return {"action": "rewrite", "text": new_text}
+                continue  # empty rewrite is a no-op, check next hook
+        return {"action": "allow"}
+    except Exception:
+        logger = logging.getLogger(__name__)
+        logger.warning("pre_agent_dispatch hook failed, falling back to local agent")
+        logger.debug("pre_agent_dispatch hook traceback:", exc_info=True)
+        return {"action": "allow"}