mios-agent-pipe + OWUI shim: SSE phase markers (step 3 of 5)

Two coordinated changes:

usr/lib/mios/agent-pipe/server.py
  Adds a `mios_status` field on otherwise-empty SSE chunks. The field
  carries pipe-phase markers (📡 prompt -> 🧭 route -> 🛠️ {tool} or
  🧠 → hermes -> ✅ done) that any translator gateway can lift into
  its native status surface (OWUI status pill, Hermes Discord
  reaction, future Slack icon, ...). Stock OpenAI streaming clients
  see an unknown field on a content-empty chunk and ignore it
  cleanly -- graceful degradation.

  Three chain paths emit markers:
    DISPATCH fast-path  -> 📡 prompt -> 🧭 route -> 🛠️ {tool} ->
                            <content delta with tool_calls envelope>
                            -> ✅ done
    CHAT fast-path      -> 📡 prompt -> 🧭 route ->
                            <content delta with reply text> ->
                            ✅ done
    AGENT / fallback    -> 📡 prompt -> 🧭 route -> 🧠 → hermes ->
                            <backend stream passes through verbatim>

  The non-streaming response path returns JSON exactly as before
  (no markers; OpenAI non-streaming has no per-chunk extension point).

usr/share/mios/owui/pipes/mios_agent_pipe.py
  Surgical slim: deletes the now-redundant router/dispatch code path
  (agent-pipe owns it) and adds the `mios_status` translator.

  Specifically:
    * BACKEND_URL default flips from :8642/v1 (hermes direct) to
      :8640/v1 (agent-pipe). Every chat request the OWUI pipe makes
      now goes through the agent-pipe service, which runs router /
      dispatch / SurrealDB writes and forwards leftover work to
      hermes itself.
    * The if-verdict-elif-chat-elif-agent dispatch branch (~60
      lines) is removed -- agent-pipe handles those three branches
      now. The OWUI pipe just gets back either a tool_calls envelope
      content delta, a chat reply, or a streaming hermes response,
      and yields each to OWUI without re-routing.
    * SSE loop gains an early-branch on chunk["mios_status"]:
      lifts {emoji, label, done} into a self._emit() call so OWUI's
      status pill mirrors the standalone service's phase markers.
      Anything else (delta.content, finish_reason) handled exactly
      as before.
    * Retained intact: task-gen bypass, tail_watcher (Hermes-
      internal tool emits sideband), CPU refine / polish / critic,
      narration collapse, output cleanup, Qwen XML stripping. Those
      are OWUI-specific quality features; can be ported to
      agent-pipe in Step 2b if Discord wants them too.

Live-verified on podman-MiOS-DEV:
  Chat-path stream from agent-pipe shows the full marker sequence
  (📡 prompt -> 🧭 route -> content -> ✅ chat) in the SSE body
  alongside the normal delta chunks. OWUI shim re-installed in
  webui.db; the chat status pill now reflects which agent-pipe
  phase the request is in instead of saying "thinking..." opaquely.

Step 4 (next commit) re-points the Hermes Discord gateway's backend
URL at :8640 so Discord chats ride the same chain. The
hermes-discord-reactions-patch.py shipped in Step 1 will then
animate the per-phase reactions from the `mios_status` markers
the agent-pipe emits.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: mios-dev

usr/lib/mios/agent-pipe/server.py

@@ -505,7 +505,14 @@ async def dispatch_mios_verb(tool: str, args: dict) -> dict:
 
 def _sse_chunk(content: str, *, chat_id: str, model: str,
                role: Optional[str] = None,
-               finish_reason: Optional[str] = None) -> bytes:
+               finish_reason: Optional[str] = None,
+               mios_status: Optional[dict] = None) -> bytes:
+    """Build an OpenAI-streaming SSE chunk. Optional `mios_status`
+    field carries pipe-internal phase emits (📡 prompt, 🧭 route,
+    🛠️ {tool}, ✅) that translator gateways (OWUI shim, Hermes
+    Discord) lift into their native status surfaces. Stock OpenAI
+    clients see this as an unknown field and ignore it -- graceful
+    degradation."""
     delta: dict[str, Any] = {}
     if role:
         delta["role"] = role
@@ -522,9 +529,24 @@ def _sse_chunk(content: str, *, chat_id: str, model: str,
             "finish_reason": finish_reason,
         }],
     }
+    if mios_status:
+        chunk["mios_status"] = mios_status
     return ("data: " + json.dumps(chunk) + "\n\n").encode("utf-8")
 
 
+def _sse_status(*, chat_id: str, model: str, emoji: str, label: str,
+                done: bool = False) -> bytes:
+    """Emit a content-empty SSE chunk whose only purpose is the
+    `mios_status` field. Standard OpenAI clients see a no-op delta
+    + ignore the extra field. Translator gateways pull the phase
+    info from `mios_status` and surface it natively (OWUI's
+    event_emitter status, Hermes Discord's reactions, etc.)."""
+    return _sse_chunk(
+        "", chat_id=chat_id, model=model,
+        mios_status={"emoji": emoji, "label": label, "done": done},
+    )
+
+
 def _sse_done() -> bytes:
     return b"data: [DONE]\n\n"
 
@@ -711,9 +733,22 @@ async def chat_completions(request: Request) -> Any:
                 )
                 if streaming:
                     async def _stream_dispatch() -> AsyncGenerator[bytes, None]:
+                        # Phase markers: prompt -> route -> tool -> done.
+                        # Translator gateways pull the emoji/label from
+                        # `mios_status` and surface natively (OWUI status
+                        # event_emitter, Discord reactions, etc.).
+                        yield _sse_status(chat_id=chat_id, model=model,
+                                          emoji="📡", label="prompt")
+                        yield _sse_status(chat_id=chat_id, model=model,
+                                          emoji="🧭", label="route")
+                        yield _sse_status(chat_id=chat_id, model=model,
+                                          emoji="🛠️", label=tool)
                         yield _sse_chunk("", chat_id=chat_id, model=model,
                                          role="assistant")
                         yield _sse_chunk(rendered, chat_id=chat_id, model=model)
+                        yield _sse_status(chat_id=chat_id, model=model,
+                                          emoji="✅" if ok else "⚠️",
+                                          label=tool, done=True)
                         yield _sse_chunk("", chat_id=chat_id, model=model,
                                          finish_reason="stop")
                         yield _sse_done()
@@ -738,9 +773,15 @@ async def _stream_dispatch() -> AsyncGenerator[bytes, None]:
             if reply:
                 if streaming:
                     async def _stream_chat() -> AsyncGenerator[bytes, None]:
+                        yield _sse_status(chat_id=chat_id, model=model,
+                                          emoji="📡", label="prompt")
+                        yield _sse_status(chat_id=chat_id, model=model,
+                                          emoji="🧭", label="route")
                         yield _sse_chunk("", chat_id=chat_id, model=model,
                                          role="assistant")
                         yield _sse_chunk(reply, chat_id=chat_id, model=model)
+                        yield _sse_status(chat_id=chat_id, model=model,
+                                          emoji="✅", label="chat", done=True)
                         yield _sse_chunk("", chat_id=chat_id, model=model,
                                          finish_reason="stop")
                         yield _sse_done()
@@ -766,6 +807,18 @@ async def _stream_chat() -> AsyncGenerator[bytes, None]:
     headers.setdefault("Content-Type", "application/json")
     if streaming:
         async def _stream_backend() -> AsyncGenerator[bytes, None]:
+            # Phase markers BEFORE the backend stream so the OWUI shim
+            # / Discord reactions know the pipe handed off to hermes.
+            # The backend's own content chunks pass through unchanged
+            # (no mios_status injected mid-stream -- the backend may
+            # emit its own tool-call status via tail_watcher or its
+            # internal mechanism).
+            yield _sse_status(chat_id=chat_id, model=model,
+                              emoji="📡", label="prompt")
+            yield _sse_status(chat_id=chat_id, model=model,
+                              emoji="🧭", label="route")
+            yield _sse_status(chat_id=chat_id, model=model,
+                              emoji="🧠", label="→ hermes")
             client = await _get_client()
             async with client.stream(
                 "POST", f"{BACKEND}/chat/completions",

usr/share/mios/owui/pipes/mios_agent_pipe.py

@@ -195,8 +195,8 @@ def _db_fire(coro: Awaitable) -> None:
 class Pipe:
     class Valves(BaseModel):
         BACKEND_URL: str = Field(
-            default="http://host.containers.internal:8642/v1",
-            description="OpenAI-compat backend = hermes-agent gateway on :8642 (direct). The prefilter @ :8641 sat in front for delegate_task forcing but the pipe now does refinement directly, so prefilter is bypassed -- removes a moving part and fixes the ClientConnectorError when prefilter is down (operator-flagged 2026-05-17). OWUI runs in a podman Quadlet so the host is reached via host.containers.internal.",
+            default="http://host.containers.internal:8640/v1",
+            description="OpenAI-compat backend = the standalone MiOS Agent Pipe service at :8640 (NOT hermes directly). Operator directive 2026-05-18: extract the router/dispatch/SurrealDB-writes chain out of this OWUI pipe class into a gateway-agnostic FastAPI service so Hermes Discord + future Slack/Telegram/MCP gateways get the same tool-understanding parity as OWUI. The agent-pipe service forwards to hermes-agent (:8642) itself. OWUI runs in a podman Quadlet so the host is reached via host.containers.internal.",
         )
         BACKEND_MODEL: str = Field(
             default="hermes-agent",
@@ -2064,71 +2064,28 @@ async def pipe(
                     _last_user_text = raw
                 break
 
-        # ── Layer-1 ROUTER ─────────────────────────────────────────
-        # Micro-LLM classifies into dispatch / chat / agent. Single-
-        # intent dispatch runs the broker tool directly (skip refine
-        # + hermes); chat returns a one-liner; agent falls through
-        # to the existing refine -> hermes -> compose -> critic chain.
-        verdict = await self._classify_intent(_last_user_text, __event_emitter__)
-        if verdict:
-            action = verdict.get("action")
-            if action == "dispatch":
-                tool = str(verdict.get("tool", "")).strip()
-                args = verdict.get("args") or {}
-                if tool:
-                    result_json = await self._dispatch_mios_verb(
-                        tool, args if isinstance(args, dict) else {},
-                        __event_emitter__,
-                    )
-                    try:
-                        result = json.loads(result_json)
-                    except json.JSONDecodeError:
-                        result = {"output": result_json}
-                    ok = bool(result.get("success"))
-                    await self._emit(__event_emitter__,
-                                     "✅" if ok else "⚠️", done=True)
-                    # OpenAI-tool-result-native propagation: emit the
-                    # structured tool_call + tool_result envelope inside
-                    # a <details type="tool_calls"> block OWUI renders
-                    # natively. The only literal characters at this layer
-                    # are cross-locale identifiers (the tool name) +
-                    # universal symbols (✅ / ⚠️). The structured JSON
-                    # matches the OpenAI tool_calls / role:tool shape,
-                    # so any downstream agent reading chat history sees
-                    # the canonical tool_result content (not a prose
-                    # render) and the operator gets a collapsible block
-                    # with the raw data inside (zero English narrative).
-                    envelope = {
-                        "tool_call": {
-                            "id": f"call_{int(time.time()*1000)}",
-                            "type": "function",
-                            "function": {
-                                "name": tool,
-                                "arguments": args if isinstance(args, dict) else {},
-                            },
-                        },
-                        "tool_result": {
-                            "success": ok,
-                            "output": (result.get("output") or "")[:2000],
-                            "stderr": (result.get("stderr") or "")[:2000],
-                        },
-                    }
-                    symbol = "✅" if ok else "⚠️"
-                    yield (
-                        f"<details type=\"tool_calls\" done=\"true\">\n"
-                        f"<summary>{symbol} `{tool}`</summary>\n\n"
-                        f"```json\n{json.dumps(envelope, indent=2, default=str)}\n```\n"
-                        f"</details>"
-                    )
-                    return
-            elif action == "chat":
-                reply = str(verdict.get("reply", "")).strip()
-                if reply:
-                    await self._emit(__event_emitter__, "✅", done=True)
-                    yield reply
-                    return
-            # action == "agent" (or any unrecognized) -> fall through
-            await self._emit(__event_emitter__, "🧭 → agent")
+        # ── Layer-1 ROUTER -- DELEGATED to mios-agent-pipe service ──
+        # The router + dispatch + chat-fast-path + SurrealDB writes
+        # are owned by the standalone agent-pipe service at :8640 now
+        # (operator directive 2026-05-18: "discord chats not going
+        # through MiOS-Agent paths" -- extracted the chain into a
+        # gateway-agnostic service so Hermes Discord + future
+        # Slack/Telegram get the same tool surface). The OWUI pipe
+        # POSTs to agent-pipe (BACKEND_URL = :8640) which runs the
+        # router and either returns a tool_calls envelope (dispatch),
+        # a short reply (chat), or streams hermes content (agent path).
+        #
+        # OWUI-specific behaviors RETAINED in this shim:
+        #   - task-gen bypass (above)
+        #   - tail_watcher (Hermes-internal tool_call emits via the
+        #     /var/lib/mios/hermes-tail/latest.json sideband)
+        #   - mios_status SSE field translation (below) -- agent-pipe
+        #     emits {emoji, label, done} markers on each phase; the
+        #     translator below calls _emit() so the OWUI status pill
+        #     stays lit during dispatch/chat/agent paths
+        #   - CPU REFINE / CRITIC / POLISH (kept below; these add
+        #     OWUI-specific quality but aren't ported to agent-pipe
+        #     yet -- Step 2b if Discord needs them too)
 
         # ── CPU REFINEMENT (in-pipe) ─────────────────────────────────
         # Extract the last user message, refine via the small CPU model,
@@ -2229,6 +2186,25 @@ def _open_details_chunk() -> str:
                             chunk = json.loads(payload_str)
                         except json.JSONDecodeError:
                             continue
+                        # mios-agent-pipe (the standalone service at
+                        # :8640) injects a `mios_status` field on
+                        # content-empty SSE chunks to carry pipe-phase
+                        # markers (📡 prompt, 🧭 route, 🛠️ {tool}, ✅).
+                        # Translate it into the OWUI status pill via
+                        # __event_emitter__ -- stock OpenAI clients
+                        # would just ignore the unknown field.
+                        _mios_status = chunk.get("mios_status")
+                        if isinstance(_mios_status, dict):
+                            emoji = str(_mios_status.get("emoji", ""))
+                            label = str(_mios_status.get("label", ""))
+                            done  = bool(_mios_status.get("done", False))
+                            description = (
+                                f"{emoji} {label}".strip() if (emoji or label)
+                                else ""
+                            )
+                            if description:
+                                await self._emit(__event_emitter__,
+                                                 description, done=done)
                         choices = chunk.get("choices") or []
                         if not choices:
                             continue