"""Default-off Hermes Gateway bridge for browser-originated chat turns."""
from __future__ import annotations

import json
import logging
import os
import threading
import time
import uuid
import urllib.error
import urllib.parse
import urllib.request
from typing import Any

from api.config import (
    AGENT_INSTANCES,
    CANCEL_FLAGS,
    PENDING_GOAL_CONTINUATION,
    STREAM_GOAL_RELATED,
    STREAMS,
    STREAMS_LOCK,
    STREAM_LAST_EVENT_ID,
    STREAM_LIVE_TOOL_CALLS,
    STREAM_PARTIAL_TEXT,
    STREAM_REASONING_TEXT,
    _get_session_agent_lock,
    _parse_provider_qualified_model_id,
    clear_session_writeback_owner_if_owned,
    coerce_reasoning_effort_for_model,
    gateway_approval_unavailable_reason,
    gateway_supports_approval,
    peek_stream,
    register_active_run,
    unregister_active_run,
    unregister_stream_owner,
    update_active_run,
)
from api.helpers import _redact_text, redact_session_data
from api.models import clear_process_wakeup_pause, get_session, merge_session_messages_append_only
from api.run_journal import RunJournalWriter, bound_run_journal_snapshot_args
from api.turn_journal import append_turn_journal_event_for_stream

logger = logging.getLogger(__name__)

# Maps stream_id -> gateway run_id for approval response relay.
_STREAM_RUN_IDS: dict[str, str] = {}
_STREAM_RUN_LIFECYCLE: dict[str, dict[str, Any]] = {}
_STREAM_RUN_STARTING_CONDITION = threading.Condition()
GATEWAY_RUN_ID_WAIT_TIMEOUT = 5.0
# stream_id -> (base_url, api_key) of the Gateway the stream's run lives on, for Stop
# and approval replies; a reattached run's Gateway need not be the process profile's.
_STREAM_ENDPOINTS: dict[str, tuple[str, str]] = {}


def gateway_run_endpoint(run_id: str) -> tuple[str, str]:
    """URL and key of the Gateway that owns run_id; the process Gateway if no live stream holds it."""
    run_id = str(run_id or "").strip()
    with _STREAM_RUN_STARTING_CONDITION:
        for stream_id, mapped in list(_STREAM_RUN_IDS.items()):
            if run_id and mapped == run_id and stream_id in _STREAM_ENDPOINTS:
                return _STREAM_ENDPOINTS[stream_id]
    from api.config import get_config

    return _gateway_base_url(get_config()), _gateway_api_key()


def _mark_gateway_run_starting(stream_id: str) -> None:
    with _STREAM_RUN_STARTING_CONDITION:
        _STREAM_RUN_IDS.pop(stream_id, None)
        _STREAM_RUN_LIFECYCLE[stream_id] = {
            "phase": "pending",
            "run_id": "",
            "waiters": 0,
            "owner_done": False,
        }


def _publish_gateway_run_id(stream_id: str, run_id: str) -> None:
    with _STREAM_RUN_STARTING_CONDITION:
        _STREAM_RUN_IDS[stream_id] = run_id
        state = _STREAM_RUN_LIFECYCLE.get(stream_id) or {}
        _STREAM_RUN_LIFECYCLE[stream_id] = {
            "phase": "ready",
            "run_id": run_id,
            "waiters": int(state.get("waiters") or 0),
            "owner_done": bool(state.get("owner_done")),
        }
        _STREAM_RUN_STARTING_CONDITION.notify_all()


def _finish_gateway_run_starting(stream_id: str, *, result: str = "failed") -> None:
    with _STREAM_RUN_STARTING_CONDITION:
        state = _STREAM_RUN_LIFECYCLE.get(stream_id) or {}
        if str(state.get("phase") or "").strip().lower() == "ready":
            return
        _STREAM_RUN_IDS.pop(stream_id, None)
        _STREAM_RUN_LIFECYCLE[stream_id] = {
            "phase": "fallback" if result == "fallback" else "failed",
            "run_id": "",
            "waiters": int(state.get("waiters") or 0),
            "owner_done": bool(state.get("owner_done")),
        }
        _STREAM_RUN_STARTING_CONDITION.notify_all()


def _retire_gateway_run_starting_if_done(stream_id: str) -> bool:
    state = _STREAM_RUN_LIFECYCLE.get(stream_id)
    if not state:
        return False
    if int(state.get("waiters") or 0) > 0:
        return False
    if not bool(state.get("owner_done")):
        return False
    _STREAM_RUN_LIFECYCLE.pop(stream_id, None)
    _STREAM_RUN_IDS.pop(stream_id, None)
    return True


def _clear_gateway_run_starting(stream_id: str) -> None:
    with _STREAM_RUN_STARTING_CONDITION:
        state = _STREAM_RUN_LIFECYCLE.get(stream_id)
        if state:
            state["owner_done"] = True
        _retire_gateway_run_starting_if_done(stream_id)
        _STREAM_RUN_STARTING_CONDITION.notify_all()


def gateway_run_id_pending(stream_id: str) -> bool:
    with _STREAM_RUN_STARTING_CONDITION:
        return str((_STREAM_RUN_LIFECYCLE.get(stream_id) or {}).get("phase") or "").strip().lower() == "pending"


def wait_for_gateway_run_id(stream_id: str, timeout: float) -> tuple[bool, str | None]:
    deadline = time.monotonic() + max(0.0, float(timeout))
    with _STREAM_RUN_STARTING_CONDITION:
        state = _STREAM_RUN_LIFECYCLE.get(stream_id)
        if state:
            state["waiters"] = int(state.get("waiters") or 0) + 1
        try:
            while True:
                state = _STREAM_RUN_LIFECYCLE.get(stream_id)
                phase = str((state or {}).get("phase") or "").strip().lower()
                if phase == "fallback":
                    return False, None
                if phase == "failed":
                    return True, None
                run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
                if phase == "ready":
                    stored_run_id = str((state or {}).get("run_id") or "").strip()
                    return True, run_id or stored_run_id or None
                if run_id:
                    return True, run_id
                if not state:
                    return False, None
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    return True, None
                _STREAM_RUN_STARTING_CONDITION.wait(timeout=remaining)
        finally:
            state = _STREAM_RUN_LIFECYCLE.get(stream_id)
            if state:
                waiters = max(0, int(state.get("waiters") or 0) - 1)
                state["waiters"] = waiters
                if _retire_gateway_run_starting_if_done(stream_id):
                    _STREAM_RUN_STARTING_CONDITION.notify_all()

_WEBUI_CHAT_BACKEND_ENV = "HERMES_WEBUI_CHAT_BACKEND"
_WEBUI_GATEWAY_BASE_URL_ENV = "HERMES_WEBUI_GATEWAY_BASE_URL"
_WEBUI_GATEWAY_API_KEY_ENV = "HERMES_WEBUI_GATEWAY_API_KEY"
_WEBUI_GATEWAY_USE_RUNS_API_ENV = "HERMES_WEBUI_GATEWAY_USE_RUNS_API"
_GATEWAY_CHAT_BACKENDS = {"gateway", "api_server", "api-server"}
# Backend tag of the in-process WebUI runtime. Local workers register their
# active run with it; cache-only Steer only enqueues on this explicit value.
WEBUI_LOCAL_CHAT_BACKEND = "legacy"


def _gateway_model_field(model: str | None) -> str:
    """Return the bare model name to put in a gateway request body.

    The picker and ``_resolve_compatible_session_model_state`` intentionally
    keep the full ``@provider:model`` string for internal routing (#1253), but
    the gateway expects a bare model name and carries the provider separately.
    Sent verbatim, ``@ollama-cloud:minimax-m2.7`` is forwarded to the upstream
    provider API, which 404s on the ``@``-prefixed string (#6722).

    Parsing is delegated to ``config._parse_provider_qualified_model_id()`` so
    a multi-segment custom provider ID (``@custom:backup:model-a``) yields the
    real model (``model-a``) instead of a positional-split fragment.
    """
    if not model:
        return ""
    value = str(model).strip()
    parsed = _parse_provider_qualified_model_id(value)
    if parsed:
        return str(parsed[0] or "").strip()
    return value


# Total byte-silence budget (seconds) for the gateway SSE socket, applied via
# ``urlopen(timeout=...)``. A stream that emits ANY byte within the window never
# trips it, so a genuinely alive (if slow) token stream is untouched; only more
# than this much *total* byte-silence is treated as a dead/stalled gateway.
#
# This is a TERMINAL budget, not a per-read grace: CPython's ``socket.makefile``
# latches ``_timeout_occurred`` on the first ``socket.timeout``, after which every
# further read raises a bare ``OSError`` — the connection cannot be resumed. So a
# read timeout ends the turn (surfacing Stop if pressed). It replaces the old flat
# 600s timeout, under which a half-open gateway (TCP open, zero bytes) pinned the
# worker for the full 10 minutes and ignored Stop (cancel is only re-checked
# between SSE lines). We KEEP the 600s default budget: there is no Gateway
# protocol heartbeat guaranteeing sub-600s progress bytes, so a legitimately
# long/fully-silent server-side tool call must not be terminated early — reducing
# the default below 600s would kill currently-working turns (gate finding, #5789).
# The win here is that a read timeout is now TERMINAL and Stop-honoring (the old
# flat timeout ignored Stop on a half-open gateway); the budget itself stays 600s
# for backward compatibility. Deployments that want a tighter dead-gateway cap can
# lower ``HERMES_WEBUI_GATEWAY_READ_TIMEOUT``.
_GATEWAY_READ_TIMEOUT_ENV = "HERMES_WEBUI_GATEWAY_READ_TIMEOUT"
_GATEWAY_READ_TIMEOUT_DEFAULT = 600.0


def _gateway_read_timeout_secs() -> float:
    """Total byte-silence budget for gateway SSE reads (default 600s, env-tunable)."""
    raw = os.environ.get(_GATEWAY_READ_TIMEOUT_ENV)
    if raw:
        try:
            val = float(raw)
            if val > 0:
                return val
        except (TypeError, ValueError):
            pass
    return _GATEWAY_READ_TIMEOUT_DEFAULT


def _iter_sse_lines_cancellable(resp, cancel_event):
    """Yield raw SSE lines from ``resp``, unblocking cleanly on a read timeout.

    ``resp``'s socket carries a read timeout (``urlopen(timeout=...)``). A read
    that blocks past it raises ``socket.timeout``, and that timeout is TERMINAL:
    CPython's ``socket.makefile`` latches ``_timeout_occurred`` on the first
    timeout, so every subsequent read raises a bare ``OSError`` ("cannot read
    from timed out object") — there is no multi-read grace to reclaim. So on a
    read timeout (or the poisoned-socket ``OSError``/any read error) this either
    surfaces the user's Stop or tears the stalled turn down:

      - cancel set -> yield ``b""`` (the caller's ``if cancel_event.is_set()``
        branch emits its cancel event), then stop. This is why the old flat 600s
        pin — where a stalled gateway ignored Stop until it eventually errored —
        is gone: Stop is honored within one timeout window.
      - otherwise -> re-raise, so the caller's error handling reports the stall.

    A stream that keeps emitting bytes within the timeout window never trips it,
    so a genuinely alive (if slow) token stream is untouched. Emitting ``b""`` is
    safe: the SSE loops decode it to an empty line and ``continue`` (same as a
    real blank line).

    Iterates ``resp`` via the iterator protocol so a real ``HTTPResponse`` and the
    test fakes (which implement ``__iter__``) behave identically.
    """
    resp_iter = iter(resp)
    while True:
        try:
            raw_line = next(resp_iter)
        except StopIteration:
            return  # EOF
        except OSError:
            # socket.timeout / TimeoutError are OSError subclasses, as is the
            # post-timeout poisoned-socket "cannot read" error. All are terminal
            # for this connection.
            if cancel_event.is_set():
                yield b""  # let the caller emit its cancel event
                return
            raise
        yield raw_line


def webui_chat_backend_mode(config_data=None, environ: dict[str, str] | None = None) -> str:
    """Return the explicitly selected browser chat backend.

    The default remains the in-process WebUI runtime. Only explicit gateway
    values opt browser chat into the Hermes API server bridge; generic truthy
    strings are deliberately ignored so deployments do not change execution
    ownership by accident.
    """
    source = os.environ if environ is None else environ
    cfg = config_data if isinstance(config_data, dict) else {}
    raw = str(
        source.get(_WEBUI_CHAT_BACKEND_ENV)
        or cfg.get("webui_chat_backend")
        or ""
    ).strip().lower()
    if raw in _GATEWAY_CHAT_BACKENDS:
        return "gateway"
    return WEBUI_LOCAL_CHAT_BACKEND


def webui_gateway_chat_enabled(config_data=None, environ: dict[str, str] | None = None) -> bool:
    return webui_chat_backend_mode(config_data, environ) == "gateway"


def _gateway_base_url(config_data=None, environ: dict[str, str] | None = None) -> str:
    source = os.environ if environ is None else environ
    cfg = config_data if isinstance(config_data, dict) else {}
    raw = str(
        source.get(_WEBUI_GATEWAY_BASE_URL_ENV)
        or cfg.get("webui_gateway_base_url")
        or "http://127.0.0.1:8642"
    ).strip()
    return raw.rstrip("/") or "http://127.0.0.1:8642"


def _gateway_api_key(environ: dict[str, str] | None = None) -> str:
    source = os.environ if environ is None else environ
    return str(
        source.get(_WEBUI_GATEWAY_API_KEY_ENV)
        or source.get("API_SERVER_KEY")
        or ""
    ).strip()


def _gateway_use_runs_api_enabled(config_data=None, environ: dict[str, str] | None = None) -> bool:
    """Return True only when the operator has explicitly opted into the runs API path."""
    source = os.environ if environ is None else environ
    cfg = config_data if isinstance(config_data, dict) else {}
    raw = str(
        source.get(_WEBUI_GATEWAY_USE_RUNS_API_ENV)
        or cfg.get("webui_gateway_use_runs_api")
        or ""
    ).strip().lower()
    return raw in ("1", "true", "yes", "on")


def _gateway_reasoning_effort_for_request(cfg, *, model=None, model_provider=None):
    """Read and coerce user-configured reasoning effort for a gateway request."""
    try:
        cfg_data = cfg if isinstance(cfg, dict) else {}
        effort_cfg = cfg_data.get("agent", {}) if isinstance(cfg_data, dict) else {}
        effort_raw = effort_cfg.get("reasoning_effort") if isinstance(effort_cfg, dict) else None
        coerced = coerce_reasoning_effort_for_model(
            effort_raw,
            model,
            provider_id=model_provider,
        )
        # Preserve explicit "none" while still omitting absent or invalid effort.
        return None if not coerced else str(coerced)
    except Exception:
        return None


def _gateway_session_yolo_enabled(session_id: str) -> bool:
    """Return the WebUI-owned, in-memory YOLO state for a browser session."""
    try:
        from tools.approval import is_session_yolo_enabled

        return bool(is_session_yolo_enabled(str(session_id or "")))
    except Exception:
        return False


def _settle_gateway_run_approval(
    session_id: str,
    approval_data: dict,
    base_url: str,
    api_key: str,
) -> tuple[bool, dict | None, int]:
    """Auto-approve or mirror one run approval at a session-linearized point."""
    from api.route_approvals import gateway_yolo_handoff, submit_gateway_pending_mirror

    run_id = str(approval_data.get("run_id") or "").strip()
    identity_v1 = bool(approval_data.get("_gateway_agent_identity_v1"))
    with gateway_yolo_handoff(session_id):
        if _gateway_session_yolo_enabled(session_id):
            try:
                _auto_approve_gateway_run(
                    base_url,
                    api_key,
                    run_id,
                    approval_data["approval_id"] if identity_v1 else "",
                )
                return True, None, 0
            except Exception:
                # Fail closed: if remote approval fails, surface the real card
                # before allowing a same-session toggle to pass the handoff.
                logger.warning(
                    "WebUI YOLO could not auto-approve run %s; showing approval card",
                    run_id,
                    exc_info=True,
                )
        head, total = submit_gateway_pending_mirror(session_id, approval_data)
        return False, head, total


def _auto_approve_gateway_run(
    base_url: str,
    api_key: str,
    run_id: str,
    approval_id: str,
) -> None:
    """Resolve one Runs API prompt using only the shipped approval contract.

    This is a WebUI-owned compatibility path: the Runs API does not yet expose
    session YOLO, so WebUI answers each approval request while its own session
    flag is enabled. Native Agent-side YOLO would be preferable because it can
    bypass gates before they pause and also covers Agent-owned computer-use
    policy; https://github.com/NousResearch/hermes-agent/pull/61946 tracks that
    API capability. Until then, do not send speculative fields to the Agent.
    """
    from api.runner_client import HttpRunnerClient

    HttpRunnerClient(base_url=base_url, api_key=api_key).respond_approval(
        run_id,
        approval_id,
        "once",
    )


def gateway_chat_config_status(config_data=None, environ: dict[str, str] | None = None) -> dict:
    """Return redacted Gateway-backed chat configuration status."""
    mode = webui_chat_backend_mode(config_data, environ)
    base_url = _gateway_base_url(config_data, environ)
    return {
        "enabled": mode == "gateway",
        "backend": mode,
        "base_url_configured": bool(base_url),
        "api_key_configured": bool(_gateway_api_key(environ)),
    }


def _gateway_http_error_event(exc: urllib.error.HTTPError, err_body: str, *, api_key_configured: bool) -> dict:
    safe = _redact_text(err_body or str(exc))[:500]
    if exc.code == 401:
        return {
            "label": "Gateway authentication failed",
            "type": "gateway_auth_error",
            "message": "Gateway rejected the WebUI API key (HTTP 401).",
            "hint": (
                "Set HERMES_WEBUI_GATEWAY_API_KEY to the same value as the Hermes Gateway "
                "API_SERVER_KEY, or disable HERMES_WEBUI_CHAT_BACKEND=gateway."
                if not api_key_configured
                else "Check that HERMES_WEBUI_GATEWAY_API_KEY matches the Hermes Gateway API_SERVER_KEY."
            ),
        }
    return {
        "label": "Gateway request failed",
        "type": "gateway_http_error",
        "message": f"Gateway returned HTTP {exc.code}.",
        "hint": safe or "Check the configured Gateway API server.",
    }


def _gateway_sse_delta(payload: dict) -> str:
    """Extract assistant text from an OpenAI-compatible streaming chunk."""
    try:
        choices = payload.get("choices") or []
        if not choices:
            return ""
        choice = choices[0] or {}
        delta = choice.get("delta") or {}
        content = delta.get("content")
        if isinstance(content, str):
            return content
        message = choice.get("message") or {}
        content = message.get("content")
        return content if isinstance(content, str) else ""
    except Exception:
        return ""


def _gateway_sse_reasoning_delta(payload: dict) -> str:
    """Extract reasoning text from OpenAI-compatible streaming chunks."""
    try:
        choices = payload.get("choices") or []
        if not choices:
            return ""
        choice = choices[0] or {}
        delta = choice.get("delta") or {}
        reasoning = delta.get("reasoning_content")
        if isinstance(reasoning, str) and reasoning.strip():
            return reasoning
        message = choice.get("message") or {}
        reasoning = message.get("reasoning_content")
        return reasoning if isinstance(reasoning, str) and reasoning.strip() else ""
    except Exception:
        return ""


def _gateway_stream_usage(payload: dict) -> dict:
    usage = payload.get("usage") if isinstance(payload, dict) else None
    if not isinstance(usage, dict):
        return {}
    return {
        "input_tokens": int(usage.get("prompt_tokens") or usage.get("input_tokens") or 0),
        "output_tokens": int(usage.get("completion_tokens") or usage.get("output_tokens") or 0),
        "estimated_cost": usage.get("estimated_cost") or usage.get("estimated_cost_usd") or 0,
    }


def _gateway_reasoning_delta(payload: dict) -> str:
    if not isinstance(payload, dict):
        return ""
    for key in ("text", "preview", "delta", "content"):
        value = payload.get(key)
        if isinstance(value, str) and value.strip():
            return value
    return ""


def _gateway_tool_progress_event(payload: dict) -> tuple[str, dict] | None:
    """Translate Hermes Gateway tool-progress SSE payloads to WebUI events."""
    if not isinstance(payload, dict):
        return None
    event_type = str(payload.get("event") or "").strip().lower()
    if event_type == "reasoning.available":
        reason_delta = _gateway_reasoning_delta(payload)
        if not reason_delta:
            return None
        return "reasoning", {"text": reason_delta}
    name = str(payload.get("tool") or payload.get("name") or payload.get("function_name") or "").strip()
    if not name:
        return None
    if name == "_thinking":
        reason_delta = _gateway_reasoning_delta(payload)
        if not reason_delta:
            return None
        return "reasoning", {"text": reason_delta}
    if name.startswith("_"):
        return None
    status = str(payload.get("status") or "running").strip().lower()
    tid = payload.get("toolCallId") or payload.get("tool_call_id") or payload.get("id")
    is_complete = event_type == "tool.completed" or status in {"completed", "complete", "success", "error", "failed"}
    event_payload = {
        "event_type": "tool.completed" if is_complete else "tool.started",
        "name": name,
        "preview": payload.get("label") or payload.get("preview"),
        "args": bound_run_journal_snapshot_args(payload.get("args"))
        if isinstance(payload.get("args"), dict)
        else {},
        "is_error": bool(payload.get("error")) or status in {"error", "failed"},
    }
    if tid:
        event_payload["tid"] = str(tid)
    return ("tool_complete" if is_complete else "tool"), event_payload


def _gateway_runs_approval_event(payload: dict) -> dict | None:
    """Map a runs-API approval.request payload to the WebUI approval contract."""
    if not isinstance(payload, dict):
        return None
    tool = str(payload.get("tool") or payload.get("function_name") or payload.get("pattern_key") or "").strip()
    command = str(payload.get("command") or "").strip()
    description = str(payload.get("description") or "").strip()
    pattern_keys = payload.get("pattern_keys") if isinstance(payload.get("pattern_keys"), list) else []
    pattern_key = str(payload.get("pattern_key") or "").strip()
    args = payload.get("args") if isinstance(payload.get("args"), (list, dict)) else []
    run_id = str(payload.get("run_id") or "").strip()
    raw_approval_id = str(payload.get("approval_id") or payload.get("id") or "").strip()
    approval_id = raw_approval_id
    if not approval_id:
        approval_id = uuid.uuid4().hex
    risk = str(payload.get("risk_level") or "high").strip()
    choices = payload.get("choices") if isinstance(payload.get("choices"), list) else []
    allow_permanent = payload.get("allow_permanent")
    if allow_permanent is None:
        allow_permanent = "always" in choices
    if not (tool or command or description):
        return None
    return {
        "tool": tool,
        "command": command,
        "description": description,
        "pattern_key": pattern_key,
        "pattern_keys": pattern_keys or ([pattern_key] if pattern_key else []),
        "args": args,
        "risk_level": risk,
        "run_id": run_id,
        "approval_id": approval_id,
        "_gateway_raw_approval_id_present": bool(raw_approval_id),
        "choices": choices,
        "allow_permanent": bool(allow_permanent),
    }


def _relay_gateway_run_approval(session_id, run_id, payload, base_url, api_key, *, put_gateway_event) -> None:
    """Auto-approve or surface one runs-API approval request as a WebUI approval card."""
    approval_data = _gateway_runs_approval_event(payload)
    if not approval_data:
        return
    approval_data["run_id"] = run_id
    from api.config import gateway_supports_approval_identity_v1
    identity_v1 = bool(approval_data.get("_gateway_raw_approval_id_present")) and gateway_supports_approval_identity_v1(base_url, api_key)
    approval_data["_gateway_agent_identity_v1"] = identity_v1
    auto_approved, head, total = _settle_gateway_run_approval(
        session_id,
        approval_data,
        base_url,
        api_key,
    )
    if not auto_approved:
        put_gateway_event("approval", {**(head or approval_data), "pending_count": total})


def _admit_gateway_run(url_runs, headers, run_body, stream_id) -> str:
    """POST /v1/runs; the same stream and body return the originally admitted run id."""
    req = urllib.request.Request(
        url_runs,
        data=json.dumps(run_body).encode("utf-8"),
        # Durable run record on the gateway: GET /v1/runs/{id} survives either side restarting.
        headers={**headers, "Idempotency-Key": f"webui-{stream_id}"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        run_data = json.loads(resp.read(65536))
    run_id = str(run_data.get("run_id") or run_data.get("id") or "").strip()
    if not run_id:
        raise ValueError(f"Gateway runs API returned no run_id: {run_data!r}")
    return run_id


def _gateway_run_headers(session_id, api_key) -> dict:
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
        "X-Hermes-Session-Id": session_id,
    }
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
        headers["X-Hermes-Session-Key"] = f"webui:{session_id}"
    return headers


def _run_gateway_runs_api_streaming(
    session_id, msg_text, model, workspace, stream_id,
    base_url, api_key, prefill_messages, body_extras,
    *, put_gateway_event, cancel_event,
    attachments=None, cfg=None, session=None,
    active_provider: str = "",
    on_run_id=None,
):
    """Submit via POST /v1/runs and relay SSE events including approval."""
    try:
        url_runs = f"{base_url.rstrip('/')}/v1/runs"
        headers = _gateway_run_headers(session_id, api_key)
        message_content: Any = str(msg_text or "")
        if attachments:
            try:
                from api.streaming import _build_native_multimodal_message

                message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=active_provider, active_model=(model or ""), requested_provider=active_provider, profile=getattr(session, "profile", None))
            except Exception:
                logger.debug("Failed to build runs-API multimodal attachment payload", exc_info=True)
                message_content = str(msg_text or "")
        from api.streaming import _strip_oob_blocks

        instructions_parts = []
        conversation_history = []
        for entry in getattr(session, "context_messages", None) or []:
            if not isinstance(entry, dict):
                continue
            role = str(entry.get("role") or "").strip().lower()
            if role not in {"user", "assistant"}:
                continue
            content = entry.get("content")
            if content is not None:
                content = _strip_oob_blocks(content)
                conversation_history.append({"role": role, "content": content})
        for entry in prefill_messages or []:
            if not isinstance(entry, dict):
                continue
            role = str(entry.get("role") or "").strip().lower()
            content = entry.get("content")
            if role == "system":
                if isinstance(content, str) and content.strip():
                    instructions_parts.append(content)
                elif content is not None:
                    instructions_parts.append(str(content))
                continue
            if role not in {"user", "assistant"}:
                continue
            if content is not None:
                content = _strip_oob_blocks(content)
            conversation_history.append({"role": role, "content": content})
        run_input = message_content
        if isinstance(run_input, list):
            run_input = [{"role": "user", "content": run_input}]
        run_body = {
            "model": _gateway_model_field(model) or "default",
            "input": run_input,
            **body_extras,
            "session_id": session_id,
        }
        if instructions_parts:
            run_body["instructions"] = "\n\n".join(part for part in instructions_parts if part)
        if conversation_history:
            run_body["conversation_history"] = conversation_history
        update_active_run(stream_id, phase="gateway-request")
        # Persist the exact body first: a restart before the run id is saved replays this admission.
        if on_run_id is not None:
            on_run_id("", request=run_body)
        run_id = _admit_gateway_run(url_runs, headers, run_body, stream_id)
    except Exception:
        _finish_gateway_run_starting(stream_id)
        raise

    usage: dict = {}
    _publish_gateway_run_id(stream_id, run_id)
    if on_run_id is not None:
        on_run_id(run_id)

    url_events = f"{base_url.rstrip('/')}/v1/runs/{run_id}/events"
    headers_sse = dict(headers)
    headers_sse["Accept"] = "text/event-stream"
    req_events = urllib.request.Request(url_events, headers=headers_sse, method="GET")
    final_text = ""
    sse_event = "message"
    with urllib.request.urlopen(req_events, timeout=_gateway_read_timeout_secs()) as resp:
        for raw_line in _iter_sse_lines_cancellable(resp, cancel_event):
            if cancel_event.is_set():
                put_gateway_event("cancel", {"message": "Cancelled by user"})
                return None, usage
            line = raw_line.decode("utf-8", errors="replace").strip()
            if not line:
                sse_event = "message"
                continue
            if line.startswith("event:"):
                sse_event = line[6:].strip() or "message"
                continue
            if not line.startswith("data:"):
                continue
            data = line[5:].strip()
            if data == "[DONE]":
                break
            try:
                payload = json.loads(data)
            except json.JSONDecodeError:
                continue
            payload_event = str(payload.get("event") or payload.get("type") or sse_event).strip() or "message"
            if payload_event == "approval.request":
                _relay_gateway_run_approval(
                    session_id, run_id, payload, base_url, api_key,
                    put_gateway_event=put_gateway_event,
                )
                sse_event = "message"
                continue
            if payload_event in {"tool.started", "tool.completed", "reasoning.available"}:
                translated = _gateway_tool_progress_event(payload)
                if translated:
                    event_name, event_payload = translated
                    if event_name == "reasoning":
                        reason_delta = event_payload.get("text")
                        if reason_delta and stream_id in STREAM_REASONING_TEXT:
                            STREAM_REASONING_TEXT[stream_id] += reason_delta
                    elif stream_id in STREAM_LIVE_TOOL_CALLS:
                        if event_name == "tool":
                            STREAM_LIVE_TOOL_CALLS[stream_id].append({
                                "name": event_payload.get("name"),
                                "args": event_payload.get("args") or {},
                                "done": False,
                                **({"tid": event_payload.get("tid")} if event_payload.get("tid") else {}),
                            })
                        elif event_name == "tool_complete":
                            for shared_tc in reversed(STREAM_LIVE_TOOL_CALLS[stream_id]):
                                if shared_tc.get("done"):
                                    continue
                                if (
                                    event_payload.get("tid") and shared_tc.get("tid") == event_payload.get("tid")
                                ) or shared_tc.get("name") == event_payload.get("name"):
                                    shared_tc["done"] = True
                                    shared_tc["is_error"] = bool(event_payload.get("is_error"))
                                    break
                    put_gateway_event(event_name, event_payload)
                    if event_name != "reasoning":
                        update_active_run(stream_id, phase="gateway-tool", latest_tool=event_payload.get("name"))
                sse_event = "message"
                continue
            if payload_event == "message.delta":
                delta = str(payload.get("delta") or "")
                if delta:
                    final_text += delta
                    if stream_id in STREAM_PARTIAL_TEXT:
                        STREAM_PARTIAL_TEXT[stream_id] += delta
                    put_gateway_event("token", {"text": delta})
                sse_event = "message"
                continue
            if payload_event == "run.completed":
                from api.route_approvals import settle_gateway_pending_run
                settle_gateway_pending_run(
                    session_id,
                    run_id,
                    reason="Gateway run completed before approval resolution",
                )
                if payload.get("error"):
                    raise RuntimeError(str(payload["error"]))
                output = str(payload.get("output") or "")
                if output and not final_text:
                    final_text = output
                    if stream_id in STREAM_PARTIAL_TEXT:
                        STREAM_PARTIAL_TEXT[stream_id] = output
                usage.update({k: v for k, v in _gateway_stream_usage(payload).items() if v})
                sse_event = "message"
                continue
            if payload_event == "run.failed":
                from api.route_approvals import settle_gateway_pending_run
                settle_gateway_pending_run(
                    session_id,
                    run_id,
                    reason="Gateway run failed before approval resolution",
                )
                raise RuntimeError(str(payload.get("error") or "Gateway run failed"))
            if payload_event == "run.cancelled":
                from api.route_approvals import settle_gateway_pending_run
                settle_gateway_pending_run(
                    session_id,
                    run_id,
                    reason="Gateway run was cancelled before approval resolution",
                )
                put_gateway_event("cancel", {"message": "Cancelled by gateway"})
                return None, usage
            reasoning_delta = _gateway_sse_reasoning_delta(payload)
            if reasoning_delta:
                if stream_id in STREAM_REASONING_TEXT:
                    STREAM_REASONING_TEXT[stream_id] += reasoning_delta
                put_gateway_event("reasoning", {"text": reasoning_delta})
            delta = _gateway_sse_delta(payload)
            if delta:
                final_text += delta
                if stream_id in STREAM_PARTIAL_TEXT:
                    STREAM_PARTIAL_TEXT[stream_id] += delta
                put_gateway_event("token", {"text": delta})
            usage.update({k: v for k, v in _gateway_stream_usage(payload).items() if v})
    return final_text, usage


def stop_gateway_run(run_id: str) -> bool:
    """Request gateway interruption and report whether it was acknowledged."""
    run_id = str(run_id or "").strip()
    if not run_id:
        return False
    base_url, api_key = gateway_run_endpoint(run_id)
    headers = {"Accept": "application/json", "Content-Type": "application/json"}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    req = urllib.request.Request(
        f"{base_url.rstrip('/')}/v1/runs/{urllib.parse.quote(run_id, safe='')}/stop",
        data=b"{}",
        headers=headers,
        method="POST",
    )
    class _NoRedirect(urllib.request.HTTPRedirectHandler):
        def redirect_request(self, req, fp, code, msg, headers, newurl):
            return None

    try:
        opener = urllib.request.build_opener(_NoRedirect)
        with opener.open(req, timeout=10) as response:
            final_url = str(getattr(response, "geturl", lambda: req.full_url)() or "")
            status = int(getattr(response, "status", getattr(response, "code", 0)) or 0)
            return 200 <= status < 300 and final_url == req.full_url
    except (urllib.error.HTTPError, urllib.error.URLError, OSError, ValueError):
        logger.debug("Gateway stop failed for run %s", run_id, exc_info=True)
        return False


_GATEWAY_RUN_TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled", "interrupted"})
GATEWAY_REATTACH_POLL_INTERVAL = 2.0
# Consecutive unreachable polls tolerated before the reattached turn is failed (~5 min at 2s).
GATEWAY_REATTACH_MAX_POLL_FAILURES = 150
_REATTACH_SCAN_HEAD_BYTES = 16 * 1024


def _record_gateway_run(session_id: str, stream_id: str, run_id: str, request=None, **extra) -> None:
    """Persist the run id (or, before admission, the exact request) so a restarted WebUI can reattach."""
    try:
        with _get_session_agent_lock(session_id):
            session = get_session(session_id)
            if not _stream_writeback_is_current(session, stream_id):
                return
            session.gateway_run = {"run_id": run_id, "stream_id": stream_id, **extra}
            if not run_id:
                session.gateway_run["request"] = request
            session.save(touch_updated_at=False)
    except Exception:
        logger.warning("Failed to persist gateway run %s for session %s", run_id, session_id, exc_info=True)
        if not run_id:
            raise  # never admit a run a restart could not recover


def _get_gateway_run_status(base_url: str, api_key: str, run_id: str) -> dict:
    headers = {"Accept": "application/json"}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    req = urllib.request.Request(
        f"{base_url.rstrip('/')}/v1/runs/{urllib.parse.quote(run_id, safe='')}",
        headers=headers,
        method="GET",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        payload = json.loads(resp.read(1024 * 1024) or b"{}")
    return payload if isinstance(payload, dict) else {}


def _await_gateway_run_result(
    session_id, stream_id, run_id, base_url, api_key,
    *, put_gateway_event, cancel_event,
):
    """Poll a run started by a previous WebUI process; same return contract as _run_gateway_runs_api_streaming."""
    from api.route_approvals import settle_gateway_pending_run

    _publish_gateway_run_id(stream_id, run_id)
    update_active_run(stream_id, phase="gateway-reattached")
    surfaced_approval_ids: set[str] = set()
    failures = 0
    while True:
        if cancel_event.is_set():
            put_gateway_event("cancel", {"message": "Cancelled by user"})
            return None, {}
        try:
            status = _get_gateway_run_status(base_url, api_key, run_id)
            failures = 0
        except (urllib.error.URLError, OSError, ValueError) as exc:
            code = getattr(exc, "code", None)
            if code == 404:
                raise RuntimeError("Gateway no longer has this run; its result could not be recovered after the WebUI restart.") from exc
            if code in (401, 403):
                raise RuntimeError(f"Gateway rejected the WebUI credentials (HTTP {code}) while reattaching to a run after the restart.") from exc
            failures += 1
            if failures >= GATEWAY_REATTACH_MAX_POLL_FAILURES:
                raise RuntimeError("Gateway became unreachable while waiting for a run that outlived the WebUI restart.") from exc
            cancel_event.wait(GATEWAY_REATTACH_POLL_INTERVAL)
            continue
        state = str(status.get("status") or "").strip().lower()
        approval = status.get("approval")
        if state == "waiting_for_approval" and isinstance(approval, dict):
            approval_key = str(approval.get("approval_id") or approval.get("id") or approval.get("timestamp") or "")
            if approval_key not in surfaced_approval_ids:
                surfaced_approval_ids.add(approval_key)
                _relay_gateway_run_approval(
                    session_id, run_id, approval, base_url, api_key, put_gateway_event=put_gateway_event,
                )
        if state in _GATEWAY_RUN_TERMINAL_STATUSES:
            settle_gateway_pending_run(session_id, run_id, reason=f"Gateway run {state} before approval resolution")
            if state == "cancelled":
                put_gateway_event("cancel", {"message": "Cancelled by gateway"})
                return None, {}
            if state != "completed":
                raise RuntimeError(str(status.get("error") or f"Gateway run {state}"))
            output = str(status.get("output") or "")
            if output and stream_id in STREAM_PARTIAL_TEXT:
                STREAM_PARTIAL_TEXT[stream_id] = output
            return output, {k: v for k, v in _gateway_stream_usage(status).items() if v}
        cancel_event.wait(GATEWAY_REATTACH_POLL_INTERVAL)


def resume_gateway_runs_after_restart() -> list[str]:
    """Reattach pending turns whose gateway run outlived the previous WebUI process; never raises.

    Call before serving so stale-pending repair does not mark these turns interrupted.
    """
    from api import models as _models

    try:
        candidates = _sidecars_with_active_stream(_models.SESSION_DIR)
    except Exception:
        logger.warning("gateway reattach: could not scan session sidecars", exc_info=True)
        return []
    resumed: list[str] = []
    for sid in candidates:
        try:
            if _resume_gateway_run_for_session(_models.Session.load_metadata_only(sid)):
                resumed.append(sid)
        except Exception:
            logger.warning("gateway reattach failed for session %s", sid, exc_info=True)
    if resumed:
        logger.info("Reattached %d gateway run(s) that outlived the previous WebUI process", len(resumed))
    return resumed


def _sidecars_with_active_stream(session_dir) -> list[str]:
    """Ids of sidecars (not the index, which can lag a save) whose header shows an active stream."""
    ids = []
    for path in sorted(session_dir.glob("*.json")):
        if path.name.startswith("_"):
            continue
        try:
            with path.open("rb") as fp:
                head = fp.read(_REATTACH_SCAN_HEAD_BYTES)
        except OSError:
            continue
        if b'"active_stream_id": null' not in head:
            ids.append(path.stem)
    return ids


def _gateway_endpoint_for_profile(profile_name) -> tuple[str, str]:
    """URL and key of the session's own profile, root included, never the process-active profile."""
    from api import profiles as _profiles
    from api.config import get_config_for_profile_home

    home = _profiles.get_hermes_home_for_profile(str(profile_name or "").strip())
    environ = {k: v for k, v in os.environ.items() if k not in _profiles._loaded_profile_env_keys}
    environ.update(_profiles.filter_runtime_env_for_gateway_parity(_profiles.get_profile_runtime_env(home)))
    return _gateway_base_url(get_config_for_profile_home(home), environ), _gateway_api_key(environ)


def _resume_gateway_run_for_session(session) -> bool:
    from api.config import create_stream_channel, register_session_writeback_owner, register_stream_owner

    run = (session.gateway_run if session is not None else None) or {}
    stream_id = str(run.get("stream_id") or "")
    # A persisted run is the authority whatever the process default backend is.
    if not stream_id or not (run.get("run_id") or run.get("request")) or session.active_stream_id != stream_id:
        return False
    sid = session.session_id
    endpoint = _gateway_endpoint_for_profile(session.profile)
    with STREAMS_LOCK:
        if stream_id in STREAMS:
            return False
        STREAMS[stream_id] = create_stream_channel()
    register_stream_owner(stream_id, sid)
    register_session_writeback_owner(sid, stream_id)
    _mark_gateway_run_starting(stream_id)
    threading.Thread(
        target=_run_gateway_chat_streaming,
        args=(sid, session.pending_user_message or "", session.model, session.workspace,
              stream_id, list(session.pending_attachments or [])),
        kwargs={
            "model_provider": session.model_provider,
            "goal_related": bool(run.get("goal_related")),
            "regeneration": bool(run.get("regeneration")),
            "reattach_run": run,
            "reattach_endpoint": endpoint,
        },
        name=f"gateway-reattach-{stream_id[:12]}",
        daemon=True,
    ).start()
    return True

def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, model_provider, terminal_error):
    from api.streaming import (
        _classify_provider_error,
        _materialize_pending_user_turn_before_error,
        _provider_error_payload,
        _session_payload_with_full_messages,
        _snapshot_and_append_partial_on_error,
        _terminal_turn_duration,
    )

    with _get_session_agent_lock(session_id):
        session = get_session(session_id)
        if not _stream_writeback_is_current(session, stream_id):
            return None
        error_classification = _classify_provider_error(terminal_error)
        error_payload = _provider_error_payload(
            terminal_error,
            error_classification["type"],
            error_classification.get("hint", ""),
        )
        turn_duration = _terminal_turn_duration(session)
        _materialize_pending_user_turn_before_error(session)
        session.active_stream_id = None
        session.gateway_run = None
        session.pending_user_message = None
        session.pending_attachments = []
        session.pending_started_at = None
        session.pending_user_source = None
        try:
            _snapshot_and_append_partial_on_error(session, stream_id)
        except Exception:
            logger.debug("Failed to snapshot gateway partials on terminal error", exc_info=True)
        error_message = {
            "role": "assistant",
            "content": (
                f"**{error_classification['label']}:** "
                f"{error_payload.get('message') or error_classification['label']}"
            ) + (f"\n\n*{error_payload['hint']}*" if error_payload.get("hint") else ""),
            "timestamp": int(time.time()),
            "_error": True,
        }
        if turn_duration is not None:
            error_message["_turnDuration"] = turn_duration
        if error_payload.get("details"):
            error_message["provider_details"] = error_payload["details"]
        if not isinstance(session.messages, list):
            session.messages = []
        session.messages.append(error_message)
        session.workspace = str(workspace)
        session.model = model
        session.model_provider = model_provider
        terminal_session_persisted = False
        try:
            session.save()
            terminal_session_persisted = True
        except Exception:
            logger.debug("Failed to persist gateway terminal error settlement", exc_info=True)
        error_payload["session"] = redact_session_data(
            _session_payload_with_full_messages(session, tool_calls=[])
        )
        error_payload["session_id"] = session.session_id
        error_payload["terminal_session_persisted"] = terminal_session_persisted
        if terminal_session_persisted:
            error_payload["terminal_session_persisted_session_id"] = session.session_id
        return error_payload


def _settle_gateway_cancelled_turn(session_id, stream_id) -> None:
    """Keep the prompt and a cancel marker when the run ended cancelled and Stop did not already settle it."""
    from api.streaming import _persist_cancelled_turn

    with _get_session_agent_lock(session_id):
        session = get_session(session_id)
        if not _stream_writeback_is_current(session, stream_id):
            return
        _persist_cancelled_turn(session, message="Cancelled by gateway")
        session.gateway_run = None
        session.save()


def _stream_writeback_is_current(session: Any, stream_id: str) -> bool:
    return bool(stream_id and getattr(session, "active_stream_id", None) == stream_id)


def _clear_gateway_pending_state(session: Any, stream_id: str) -> None:
    if not _stream_writeback_is_current(session, stream_id):
        # Cancel clears active_stream_id eagerly; still drop this stream's run record.
        if session is not None and (session.gateway_run or {}).get("stream_id") == stream_id:
            session.gateway_run = None
            session.save(touch_updated_at=False)
        return
    session.active_stream_id = None
    session.gateway_run = None
    session.pending_user_message = None
    session.pending_attachments = None
    session.pending_started_at = None
    session.pending_user_source = None
    session.save()


def _cleanup_gateway_pending_mirror(session_id: str) -> None:
    try:
        from api.route_approvals import (
            retire_gateway_pending_mirror,
        )

        retire_gateway_pending_mirror(session_id)
    except Exception:
        logger.debug("Failed to reconcile gateway pending mirror during teardown", exc_info=True)


def _run_gateway_chat_streaming(
    session_id,
    msg_text,
    model,
    workspace,
    stream_id,
    attachments=None,
    *,
    model_provider=None,
    goal_related=False,
    regeneration=False,
    reattach_run=None,
    reattach_endpoint=None,
):
    """Bridge a WebUI chat turn through Hermes Gateway's API server.

    This default-off path keeps the browser contract unchanged: /api/chat/start
    still returns a local stream_id and /api/chat/stream still receives WebUI SSE
    event names. The worker translates OpenAI-compatible streaming chunks from
    the configured Gateway API server into those local events and persists the
    final user/assistant turn back into the WebUI session.
    """
    q = peek_stream(stream_id)
    if q is None:
        _finish_gateway_run_starting(stream_id, result="fallback")
        _clear_gateway_run_starting(stream_id)
        # Cancelled before the worker started; release the owner entry the route
        # layer registered so STREAM_SESSION_OWNERS does not leak (no teardown finally runs).
        unregister_stream_owner(stream_id)
        # Also release the writeback-owner entry the route layer registered, so
        # SESSION_WRITEBACK_OWNERS does not leak on this pre-start cancellation
        # path (the teardown finally below never runs when we early-return here).
        clear_session_writeback_owner_if_owned(session_id, stream_id)
        return
    register_active_run(
        stream_id,
        session_id=session_id,
        started_at=time.time(),
        phase="gateway-starting",
        workspace=str(workspace),
        model=model,
        provider=model_provider,
        backend="gateway",
    )
    try:
        run_journal = RunJournalWriter(session_id, stream_id)
    except Exception:
        run_journal = None
        logger.debug("Failed to initialize gateway run journal for stream %s", stream_id, exc_info=True)
    cancel_event = threading.Event()
    with STREAMS_LOCK:
        CANCEL_FLAGS[stream_id] = cancel_event
        STREAM_PARTIAL_TEXT[stream_id] = ""
        STREAM_REASONING_TEXT[stream_id] = ""
        STREAM_LIVE_TOOL_CALLS[stream_id] = []

    success_writeback_committed = False
    runs_api_pending_marked = True

    def put_gateway_event(event, data):
        if cancel_event.is_set() and not success_writeback_committed and event not in ("cancel", "error", "apperror"):
            return
        if event == "apperror" and isinstance(data, dict):
            data = data.copy()
            data.setdefault("session_id", session_id)
        event_id = None
        if run_journal is not None:
            try:
                journaled = run_journal.append_sse_event(event, data)
                event_id = (journaled or {}).get("event_id") if isinstance(journaled, dict) else None
                if event_id:
                    STREAM_LAST_EVENT_ID[stream_id] = event_id
            except Exception:
                logger.debug("Failed to append gateway event %s for stream %s", event, stream_id, exc_info=True)
        if event_id and hasattr(q, "note_last_event_id"):
            try:
                q.note_last_event_id(event_id)
            except Exception:
                logger.debug("Failed to note gateway event_id %s for stream %s", event_id, stream_id, exc_info=True)
        try:
            queue_item = (event, data, event_id) if hasattr(q, "subscribe_with_snapshot") else (event, data)
            q.put_nowait(queue_item)
        except Exception:
            logger.debug("Failed to put gateway event to queue")

    s = None
    final_text = ""
    terminal_error = ""
    usage = {"input_tokens": 0, "output_tokens": 0, "estimated_cost": 0}
    try:
        s = get_session(session_id)
        from api.config import get_config  # imported lazily to avoid config-cycle churn

        cfg = get_config()
        reasoning_effort = _gateway_reasoning_effort_for_request(
            cfg,
            model=model,
            model_provider=model_provider,
        )
        base_url, api_key = reattach_endpoint or (_gateway_base_url(cfg), _gateway_api_key())
        with _STREAM_RUN_STARTING_CONDITION:
            _STREAM_ENDPOINTS[stream_id] = (base_url, api_key)
        try:
            from api.config import _main_model_request_overrides
            _gw_overrides = _main_model_request_overrides(
                cfg,
                effective_model=model,
                effective_provider=model_provider,
            )
        except Exception:
            _gw_overrides = {}
        _runs_api_enabled = _gateway_use_runs_api_enabled(cfg)
        _use_runs_api = bool(reattach_run) or (_runs_api_enabled and gateway_supports_approval(base_url, api_key))
        if not _use_runs_api and runs_api_pending_marked:
            _finish_gateway_run_starting(stream_id, result="fallback")
            runs_api_pending_marked = False
        try:
            from api.streaming import (
                _load_webui_prefill_context,
                _prefill_messages_with_webui_context,
                _normalize_prefill_messages_before_user_turn,
                _public_prefill_context_status,
                _webui_ephemeral_system_prompt,
            )

            prefill_context = _load_webui_prefill_context(cfg)
            # #3324: the WebUI session/delivery context (connected platforms,
            # home channels, delivery hints, session framing) is now carried in
            # the ephemeral system prompt rather than a prefill `user` message.
            # The gateway-backed path must build the SAME system prompt so that
            # context is not silently dropped on Gateway-routed WebUI chats.
            _gateway_system_prompt = _webui_ephemeral_system_prompt(
                None,
                surface_context={
                    "source": "webui",
                    "session_id": session_id,
                    "profile": getattr(s, "profile", None),
                    "workspace": s.workspace if s is not None else str(workspace),
                },
                config_data=cfg,
            )
            prefill_messages = _prefill_messages_with_webui_context(prefill_context, cfg)
            prefill_messages = _normalize_prefill_messages_before_user_turn(prefill_messages)
            prefill_messages = [
                {"role": "system", "content": _gateway_system_prompt},
                *prefill_messages,
            ]
            put_gateway_event("context_status", {
                "session_id": session_id,
                "prefill": _public_prefill_context_status(prefill_context),
            })
        except Exception:
            logger.debug("Failed to load WebUI gateway prefill context", exc_info=True)
            prefill_messages = []
        if _use_runs_api:
            body_extras = {}
            if model_provider:
                body_extras["provider"] = model_provider
            if reasoning_effort is not None:
                body_extras["reasoning_effort"] = reasoning_effort
            if _gw_overrides.get("service_tier"):
                body_extras["service_tier"] = _gw_overrides["service_tier"]
            record_run = lambda run_id, request=None: _record_gateway_run(
                session_id, stream_id, run_id, request=request,
                regeneration=bool(regeneration), goal_related=bool(goal_related),
            )
            try:
                if reattach_run:
                    run_id = str(reattach_run.get("run_id") or "").strip()
                    if not run_id:
                        # Crashed between admission and saving the id: the same key returns the original run.
                        run_id = _admit_gateway_run(
                            f"{base_url.rstrip('/')}/v1/runs", _gateway_run_headers(session_id, api_key),
                            reattach_run["request"], stream_id,
                        )
                        record_run(run_id)
                    final_text, usage = _await_gateway_run_result(
                        session_id, stream_id, run_id, base_url, api_key,
                        put_gateway_event=put_gateway_event,
                        cancel_event=cancel_event,
                    )
                else:
                    final_text, usage = _run_gateway_runs_api_streaming(
                        session_id, msg_text, model, workspace, stream_id,
                        base_url, api_key, prefill_messages, body_extras,
                        put_gateway_event=put_gateway_event,
                        cancel_event=cancel_event,
                        attachments=attachments,
                        cfg=cfg,
                        session=s,
                        active_provider=(model_provider or ""),
                        on_run_id=record_run,
                    )
            except Exception as exc:
                error_payload = _settle_gateway_terminal_error(
                    session_id,
                    stream_id,
                    workspace,
                    model,
                    model_provider,
                    str(exc),
                )
                if error_payload is None:
                    return
                put_gateway_event("apperror", error_payload)
                return
            if final_text is None:
                _settle_gateway_cancelled_turn(session_id, stream_id)
                return
        else:
            # Legacy gateway path: emit unsupported approval notice once per session,
            # but only when the gateway genuinely lacks approval capability.
            approval_reason = gateway_approval_unavailable_reason(base_url, api_key)
            if approval_reason is not None:
                if not hasattr(s, "_approval_notice_emitted"):
                    s._approval_notice_emitted = False
                if not s._approval_notice_emitted:
                    approval_message = "Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this."
                    approval_type = "approval_gateway_unsupported"
                    if approval_reason == "unreachable":
                        approval_type = "approval_gateway_offline"
                        approval_message = "Gateway connection failed. Check that the connected Hermes gateway is running and reachable."
                    put_gateway_event("warning", {
                        "type": approval_type,
                        "message": approval_message,
                    })
                    s._approval_notice_emitted = True

            url = f"{base_url}/v1/chat/completions"
            headers = {
                "Content-Type": "application/json",
                "Accept": "text/event-stream",
                "X-Hermes-Session-Id": session_id,
            }
            if api_key:
                headers["Authorization"] = f"Bearer {api_key}"
                # Scope Gateway long-term continuity to this WebUI conversation
                # without exposing the browser's auth cookie or CSRF material.
                headers["X-Hermes-Session-Key"] = f"webui:{session_id}"
            message_content: Any = str(msg_text or "")
            if attachments:
                try:
                    from api.streaming import _build_native_multimodal_message

                    message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=(model_provider or ""), active_model=(model or ""), requested_provider=(model_provider or ""), profile=getattr(s, "profile", None))
                except Exception:
                    logger.debug("Failed to build gateway multimodal attachment payload", exc_info=True)
                    message_content = str(msg_text or "")
            body = {
                "model": _gateway_model_field(model) or "default",
                "stream": True,
                "messages": [*prefill_messages, {"role": "user", "content": message_content}],
            }
            if model_provider:
                body["provider"] = model_provider
            if reasoning_effort is not None:
                body["reasoning_effort"] = reasoning_effort
            if _gw_overrides.get("service_tier"):
                body["service_tier"] = _gw_overrides["service_tier"]
            req = urllib.request.Request(
                url,
                data=json.dumps(body).encode("utf-8"),
                headers=headers,
                method="POST",
            )
            update_active_run(stream_id, phase="gateway-request")
            last_payload = {}
            sse_event = "message"
            with urllib.request.urlopen(req, timeout=_gateway_read_timeout_secs()) as resp:
                for raw_line in _iter_sse_lines_cancellable(resp, cancel_event):
                    if cancel_event.is_set():
                        put_gateway_event("cancel", {"message": "Cancelled by user"})
                        return
                    line = raw_line.decode("utf-8", errors="replace").strip()
                    if not line:
                        sse_event = "message"
                        continue
                    if line.startswith("event:"):
                        sse_event = line[6:].strip() or "message"
                        continue
                    if not line.startswith("data:"):
                        continue
                    data = line[5:].strip()
                    if data == "[DONE]":
                        break
                    try:
                        payload = json.loads(data)
                    except json.JSONDecodeError:
                        continue
                    _payload_event = str(payload.get("event") or payload.get("type") or sse_event).strip()
                    if _payload_event in {"hermes.approval.request", "approval.request"}:
                        approval_data = _gateway_runs_approval_event(payload)
                        if approval_data:
                            # Record the gateway run_id so /api/approval/respond
                            # can relay the choice back and resume the parked run
                            # (legacy path never creates a local run; without this
                            # the card renders but approve/deny returns ok:false).
                            # No-op when the payload omits run_id.
                            _approval_run_id = str(approval_data.get("run_id") or "").strip()
                            if _approval_run_id:
                                _STREAM_RUN_IDS[stream_id] = _approval_run_id
                            try:
                                from api.route_approvals import submit_gateway_pending_mirror
                                head, total = submit_gateway_pending_mirror(session_id, approval_data)
                                approval_data = {**(head or approval_data), "pending_count": total}
                            except Exception:
                                logger.debug("submit_gateway_pending_mirror failed", exc_info=True)
                            put_gateway_event("approval", approval_data)
                        else:
                            logger.debug("Ignoring malformed gateway approval payload")
                        sse_event = "message"
                        continue
                    if sse_event == "hermes.tool.progress":
                        translated = _gateway_tool_progress_event(payload)
                        if translated:
                            event_name, event_payload = translated
                            if event_name == "reasoning":
                                reason_delta = event_payload.get("text")
                                if reason_delta and stream_id in STREAM_REASONING_TEXT:
                                    STREAM_REASONING_TEXT[stream_id] += reason_delta
                            elif stream_id in STREAM_LIVE_TOOL_CALLS:
                                if event_name == "tool":
                                    STREAM_LIVE_TOOL_CALLS[stream_id].append({
                                        "name": event_payload.get("name"),
                                        "args": event_payload.get("args") or {},
                                        "done": False,
                                        **({"tid": event_payload.get("tid")} if event_payload.get("tid") else {}),
                                    })
                                else:
                                    for shared_tc in reversed(STREAM_LIVE_TOOL_CALLS[stream_id]):
                                        if shared_tc.get("done"):
                                            continue
                                        if (
                                            event_payload.get("tid") and shared_tc.get("tid") == event_payload.get("tid")
                                        ) or shared_tc.get("name") == event_payload.get("name"):
                                            shared_tc["done"] = True
                                            shared_tc["is_error"] = bool(event_payload.get("is_error"))
                                            break
                            put_gateway_event(event_name, event_payload)
                            if event_name != "reasoning":
                                update_active_run(stream_id, phase="gateway-tool", latest_tool=event_payload.get("name"))
                        sse_event = "message"
                        continue
                    if sse_event == "reasoning.available":
                        reason_delta = _gateway_reasoning_delta(payload)
                        if reason_delta:
                            if stream_id in STREAM_REASONING_TEXT:
                                STREAM_REASONING_TEXT[stream_id] += reason_delta
                            put_gateway_event("reasoning", {"text": reason_delta})
                        sse_event = "message"
                        continue
                    last_payload = payload
                    if payload.get("error"):
                        terminal_error = str(payload["error"])
                    reasoning_delta = _gateway_sse_reasoning_delta(payload)
                    if reasoning_delta:
                        if stream_id in STREAM_REASONING_TEXT:
                            STREAM_REASONING_TEXT[stream_id] += reasoning_delta
                        put_gateway_event("reasoning", {"text": reasoning_delta})
                    delta = _gateway_sse_delta(payload)
                    if delta:
                        final_text += delta
                        if stream_id in STREAM_PARTIAL_TEXT:
                            STREAM_PARTIAL_TEXT[stream_id] += delta
                        put_gateway_event("token", {"text": delta})
                    usage.update({k: v for k, v in _gateway_stream_usage(payload).items() if v})
            usage.update({k: v for k, v in _gateway_stream_usage(last_payload).items() if v})
        assistant_text = final_text.strip()
        if terminal_error:
            error_payload = _settle_gateway_terminal_error(
                session_id,
                stream_id,
                workspace,
                model,
                model_provider,
                terminal_error,
            )
            if error_payload is None:
                return
            put_gateway_event("apperror", error_payload)
            return
        if not assistant_text:
            put_gateway_event("apperror", {
                "label": "Gateway returned no response",
                "type": "gateway_empty_response",
                "message": "Gateway returned no assistant message for this turn.",
                "hint": "Check that Hermes Gateway API server is running and reachable.",
            })
            return
        with _get_session_agent_lock(session_id):
            s = get_session(session_id)
            if not _stream_writeback_is_current(s, stream_id):
                return
            # A late Stop can land after Gateway has yielded a full answer but
            # before success writeback. Treat it as cancellation so any
            # credential-exhausted process-wakeup pause stays in place.
            if cancel_event.is_set():
                put_gateway_event("cancel", {"message": "Cancelled by user"})
                return
            now = time.time()
            # Preserve subsecond ordering for gateway-backed turns. Using an
            # integer seconds timestamp gives the user and assistant rows the
            # same sort key; later transcript merges can then fall back to
            # role/content ordering instead of turn order.
            assistant_ts = now + 0.000001
            pending_source = getattr(s, "pending_user_source", None) or "webui"
            from api.streaming import (
                _active_turn_authority,
                _active_turn_token_matches,
                _materialize_active_turn_user,
            )

            active_turn_identity = _active_turn_authority(s, stream_id, msg_text)
            user_msg = _materialize_active_turn_user(
                active_turn_identity,
                str(msg_text or ""),
                pending_source,
            )
            user_msg["timestamp"] = float(
                active_turn_identity.get("timestamp") or now
            )
            assistant_msg = {"role": "assistant", "content": assistant_text, "timestamp": assistant_ts}
            saved_reasoning = STREAM_REASONING_TEXT.get(stream_id, "")
            if saved_reasoning:
                assistant_msg["reasoning"] = saved_reasoning
            previous_messages = list(getattr(s, "messages", None) or [])
            stored_context = getattr(s, "context_messages", None)
            previous_context = list(
                stored_context
                if isinstance(stored_context, list) and (regeneration or stored_context)
                else getattr(s, "messages", None) or []
            )
            previous_process_wakeup_pause = dict(getattr(s, "process_wakeup_pause", {}) or {})
            # Stamp stable ids on the two new rows (shared with the display merge
            # below) so display and model-context copies share an id for the
            # fork/truncate aligner (#context-message-stable-id).
            try:
                from api.streaming import _assign_stable_message_ids

                _assign_stable_message_ids(
                    [user_msg, assistant_msg],
                    previous_context,
                    list(getattr(s, "messages", None) or []),
                )
            except Exception:
                logger.debug("Failed to stamp stable ids on gateway turn rows", exc_info=True)
            s.context_messages = previous_context + [user_msg, assistant_msg]
            try:
                from api.streaming import _is_context_compression_marker

                display_context = [
                    msg
                    for msg in previous_context
                    if not _is_context_compression_marker(msg)
                ]
            except Exception:
                logger.debug("Failed to filter gateway display context markers", exc_info=True)
                display_context = previous_context
            display = merge_session_messages_append_only(
                previous_messages,
                display_context,
            )
            try:
                from api.streaming import _merge_display_messages_after_agent_result

                s.messages = _merge_display_messages_after_agent_result(
                    display,
                    previous_context,
                    s.context_messages,
                    str(msg_text or ""),
                    source=pending_source,
                    verification_nudge_provenance={
                        "active_turn_identity": active_turn_identity,
                    },
                )
            except Exception:
                logger.debug("Failed to merge gateway display transcript", exc_info=True)
                # Avoid duplicating the eager-save checkpointed user message.
                if display:
                    latest = display[-1]
                    if isinstance(latest, dict) and latest.get("role") == "user":
                        latest_text = " ".join(str(latest.get("content") or "").split())
                        msg_norm = " ".join(str(msg_text or "").split())
                        if latest_text == msg_norm:
                            display = display[:-1]
                s.messages = display + [user_msg, assistant_msg]
            if active_turn_identity.get("token"):
                current_display_rows = [
                    message
                    for message in s.messages
                    if _active_turn_token_matches(message, active_turn_identity)
                ]
                if len(current_display_rows) == 1:
                    current_display_rows[0]["timestamp"] = user_msg["timestamp"]
            s.active_stream_id = None
            s.gateway_run = None
            s.pending_user_message = None
            s.pending_attachments = None
            s.pending_started_at = None
            s.pending_user_source = None
            s.workspace = str(workspace)
            s.model = model
            s.model_provider = model_provider

            def _restore_cancelled_success_writeback():
                if pending_source == "process_wakeup":
                    s.context_messages = previous_context
                    s.messages = previous_messages
                    s.process_wakeup_pause = dict(previous_process_wakeup_pause)
                elif previous_process_wakeup_pause:
                    s.process_wakeup_pause = dict(previous_process_wakeup_pause)
                else:
                    clear_process_wakeup_pause(s, reason="run_completed")
                s.save()
                put_gateway_event("cancel", {"message": "Cancelled by user"})

            # Recheck immediately before clearing the pause; Stop can arrive
            # while the success transcript is being assembled.
            if cancel_event.is_set():
                _restore_cancelled_success_writeback()
                return
            clear_process_wakeup_pause(s, reason="run_completed")
            if cancel_event.is_set():
                _restore_cancelled_success_writeback()
                return
            s.save()
            if cancel_event.is_set():
                _restore_cancelled_success_writeback()
                return
            # #6366 re-gate: record the durable same-stream completion
            # event in the crash-safe turn journal. The run journal's
            # terminal state is only reached on its own ``stream_end``
            # write path, so a Gateway run whose terminal write is lost
            # would otherwise leave no completion evidence at all and
            # stale-cancel recovery would re-append a duplicate
            # recovered row after the valid final answer.
            try:
                append_turn_journal_event_for_stream(
                    session_id,
                    stream_id,
                    {"event": "completed", "created_at": time.time()},
                )
            except Exception:
                logger.debug("Failed to append completed turn journal event", exc_info=True)
            success_writeback_committed = True
        try:
            from api.goals import evaluate_goal_after_turn, has_active_goal
            from api.profiles import get_hermes_home_for_profile

            profile_home = get_hermes_home_for_profile(getattr(s, "profile", None))
            if goal_related and has_active_goal(session_id, profile_home=profile_home):
                put_gateway_event("goal", {
                    "session_id": session_id,
                    "state": "evaluating",
                    "message": "Evaluating goal progress…",
                    "message_key": "goal_evaluating_progress",
                })
                decision = evaluate_goal_after_turn(
                    session_id,
                    assistant_text,
                    user_initiated=True,
                    profile_home=profile_home,
                ) or {}
                goal_message = str(decision.get("message") or "").strip()
                if goal_message:
                    put_gateway_event("goal", {
                        "session_id": session_id,
                        "state": "continuing" if decision.get("should_continue") else "idle",
                        "message": goal_message,
                        "message_key": decision.get("message_key") or (
                            "goal_continuing" if goal_message else ""
                        ),
                        "message_args": decision.get("message_args") or [],
                        "decision": decision,
                    })
                if decision.get("should_continue"):
                    continuation_prompt = str(decision.get("continuation_prompt") or "").strip()
                    if continuation_prompt:
                        PENDING_GOAL_CONTINUATION.add(session_id)
                        put_gateway_event("goal_continue", {
                            "session_id": session_id,
                            "continuation_prompt": continuation_prompt,
                            "text": continuation_prompt,
                            "message": goal_message,
                            "message_key": decision.get("message_key") or "goal_continuing",
                            "message_args": decision.get("message_args") or [],
                            "decision": decision,
                        })
        except Exception as goal_exc:
            logger.debug(
                "Gateway goal continuation hook failed for session %s: %s",
                session_id,
                goal_exc,
            )
        from api.streaming import _session_payload_with_full_messages
        gateway_session_payload = _session_payload_with_full_messages(s, tool_calls=[])
        put_gateway_event("done", {"session": redact_session_data(gateway_session_payload), "usage": usage})
        put_gateway_event("stream_end", {"session_id": session_id})
    except urllib.error.HTTPError as exc:
        try:
            err_body = exc.read(2048).decode("utf-8", errors="replace")
        except Exception:
            err_body = ""
        put_gateway_event(
            "apperror",
            _gateway_http_error_event(exc, err_body, api_key_configured=bool(_gateway_api_key())),
        )
    except Exception as exc:
        safe = _redact_text(str(exc))[:500]
        put_gateway_event("apperror", {
            "label": "Gateway request failed",
            "type": "gateway_error",
            "message": safe or "Gateway request failed.",
            "hint": "Check HERMES_WEBUI_GATEWAY_BASE_URL and Gateway API server health.",
        })
    finally:
        mapped_run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
        if mapped_run_id:
            try:
                from api.route_approvals import settle_gateway_pending_run
                settle_gateway_pending_run(
                    session_id,
                    mapped_run_id,
                    reason="Gateway run ended during teardown before approval resolution",
                )
            except Exception:
                logger.debug("Failed to settle gateway pending approvals during teardown", exc_info=True)
        if s is not None:
            try:
                with _get_session_agent_lock(session_id):
                    _clear_gateway_pending_state(get_session(session_id), stream_id)
            except Exception:
                logger.debug("Failed to clear gateway stream state", exc_info=True)
            _cleanup_gateway_pending_mirror(session_id)
        with STREAMS_LOCK:
            AGENT_INSTANCES.pop(stream_id, None)
            CANCEL_FLAGS.pop(stream_id, None)
            STREAM_GOAL_RELATED.pop(stream_id, None)
            STREAM_PARTIAL_TEXT.pop(stream_id, None)
            STREAM_REASONING_TEXT.pop(stream_id, None)
            STREAM_LIVE_TOOL_CALLS.pop(stream_id, None)
            STREAM_LAST_EVENT_ID.pop(stream_id, None)
            STREAMS.pop(stream_id, None)
        if runs_api_pending_marked and gateway_run_id_pending(stream_id):
            _finish_gateway_run_starting(stream_id)
        _clear_gateway_run_starting(stream_id)
        with _STREAM_RUN_STARTING_CONDITION:
            _STREAM_ENDPOINTS.pop(stream_id, None)
        unregister_stream_owner(stream_id)
        unregister_active_run(stream_id)
        # Release the writeback-owner entry the route layer registered for this
        # Gateway run so SESSION_WRITEBACK_OWNERS does not grow unbounded across
        # the process lifetime (compare-and-clear: only clears if still owned by
        # this stream, mirroring the local streaming teardown).
        clear_session_writeback_owner_if_owned(session_id, stream_id)