EventHubAiRouter/router/router.py

"""
FastAPI AI router gateway — Zed entrypoint with tier/lane orchestration.
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
import time
from typing import Any

import httpx
from fastapi import FastAPI, Header, HTTPException, Request, Response
from fastapi.responses import JSONResponse, StreamingResponse

from metrics import CLASSIFY, CLASSIFY_LLM, DURATION, ESCALATIONS, REQUESTS, metrics_payload
from hierarchical import HierarchicalRunner, should_run_hierarchical, find_pending_plan
from agent_stream import stream_agent_plan_then_act
from agent_hier import (
    completion_has_tool_calls,
    completion_preview,
    completion_to_sse_chunks,
    context_fill_for_forward,
    executor_fallback_completion,
    force_edit_after_read_completion,
    inject_plan_context,
    merge_agent_meta,
    meta_header,
    messages_have_tool_activity,
    pick_agent_executor_model,
    plan_payload_from_meta,
    prepare_agent_executor_forward,
    request_has_tools,
    rewrite_redundant_reread_completion,
    sanitize_plan_paths,
    should_force_edit_after_read,
    should_stop_after_edit,
    stop_after_edit_completion,
    synthetic_first_tool_completion,
)
from orchestrator import Orchestrator, SessionStore, Tier
from rules_loader import reload_configs

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("ai-router")

LITELLM_URL = os.environ.get("LITELLM_INTERNAL_URL", "http://litellm:4000").rstrip("/")
LITELLM_KEY = os.environ.get("LITELLM_MASTER_KEY", "")
ROUTER_API_KEY = os.environ.get("ROUTER_API_KEY", "")
DEFAULT_MODEL = os.environ.get("DEFAULT_LITELLM_MODEL", "smart-router")
REDIS_URL = os.environ.get("REDIS_URL", "redis://redis:6379/0")

session_store = SessionStore(REDIS_URL or None)
orchestrator = Orchestrator(session_store)

app = FastAPI(title="EventHub AI Router", version="2.0.0")


def _message_text(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for block in content:
            if isinstance(block, dict) and block.get("type") == "text":
                parts.append(str(block.get("text", "")))
        return "\n".join(parts)
    return ""


def _extract_text(messages: list[dict[str, Any]]) -> str:
    parts: list[str] = []
    for msg in messages:
        parts.append(_message_text(msg.get("content", "")))
    return "\n".join(parts)


def _last_user_text(messages: list[dict[str, Any]]) -> str:
    """Latest user turn only (plan approve/amend must not see prior transcript)."""
    for msg in reversed(messages or []):
        if not isinstance(msg, dict):
            continue
        if str(msg.get("role") or "") != "user":
            continue
        text = _message_text(msg.get("content", "")).strip()
        if text:
            return text
    return _extract_text(messages)


def _has_image(messages: list[dict[str, Any]]) -> bool:
    for msg in messages:
        content = msg.get("content", "")
        if not isinstance(content, list):
            continue
        for block in content:
            if isinstance(block, dict) and block.get("type") in ("image_url", "image"):
                return True
    return False


def _estimate_tokens(text: str) -> int:
    return max(1, len(text) // 4)


def _auth_or_403(authorization: str | None) -> None:
    if not ROUTER_API_KEY:
        return
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing Bearer token")
    token = authorization.removeprefix("Bearer ").strip()
    if token != ROUTER_API_KEY:
        raise HTTPException(status_code=403, detail="Invalid API key")


def _orchestrate_header(header: str | None, body: dict[str, Any]) -> str | None:
    if header:
        return header.strip().lower()
    meta = body.get("metadata") or {}
    if isinstance(meta, dict) and meta.get("orchestrate") is not None:
        return str(meta.get("orchestrate")).strip().lower()
    return None


def _completion_from_text(content: str, *, model: str, meta: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": f"hier-{int(time.time() * 1000)}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": model,
        "choices": [
            {
                "index": 0,
                "message": {"role": "assistant", "content": content},
                "finish_reason": "stop",
            }
        ],
        "x_router_meta": meta,
    }


def _meta_header(meta: dict[str, Any] | None) -> str:
    """HTTP-header-safe meta (delegates to agent_hier.meta_header)."""
    return meta_header(meta)


def _sse_completion(content: str, *, model: str, meta: dict[str, Any]) -> StreamingResponse:
    chunk = {
        "id": f"hier-{int(time.time() * 1000)}",
        "object": "chat.completion.chunk",
        "created": int(time.time()),
        "model": model,
        "choices": [
            {
                "index": 0,
                "delta": {"role": "assistant", "content": content},
                "finish_reason": "stop",
            }
        ],
    }

    async def gen():
        yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n".encode("utf-8")
        yield b"data: [DONE]\n\n"

    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={"X-Router-Meta": _meta_header(meta)},
    )


def _hier_cfg() -> dict[str, Any]:
    from rules_loader import load_orchestration

    return load_orchestration().get("hierarchical", {}) or {}


def _format_progress_block(lines: list[str]) -> str:
    from progress_ui import format_progress_block

    return format_progress_block(lines)


def _sse_chunk(
    *,
    cid: str,
    model: str,
    delta: dict[str, Any],
    finish_reason: str | None = None,
) -> bytes:
    payload = {
        "id": cid,
        "object": "chat.completion.chunk",
        "created": int(time.time()),
        "model": model,
        "choices": [
            {
                "index": 0,
                "delta": delta,
                "finish_reason": finish_reason,
            }
        ],
    }
    return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n".encode("utf-8")


def _sse_hierarchical_progress(
    runner: HierarchicalRunner,
    *,
    user_text: str,
    quality_mode: str | None,
    session_id: str,
    prompt_hash: str,
    base_meta: dict[str, Any],
    fallback_model: str,
    messages: list[dict[str, Any]] | None = None,
) -> StreamingResponse:
    """Stream live stage updates, then final answer (Zed sees progress while waiting)."""
    import asyncio

    cfg = _hier_cfg()
    stream_progress = bool(cfg.get("stream_progress", True))
    progress_in_content = bool(cfg.get("progress_in_content", True))
    cid = f"hier-{int(time.time() * 1000)}"
    model = "hierarchical"
    tier = str(base_meta.get("tier", "UNKNOWN"))
    lane = str(base_meta.get("lane", "?"))
    t0 = time.perf_counter()

    async def gen():
        queue: asyncio.Queue[tuple[str, Any]] = asyncio.Queue()

        async def on_progress(msg: str) -> None:
            if stream_progress:
                await queue.put(("progress", msg))

        async def work() -> None:
            try:
                hier = await runner.run(
                    user_text,
                    quality_mode=quality_mode,
                    session_id=session_id,
                    messages=messages,
                    on_progress=on_progress if stream_progress else None,
                )
                await queue.put(("done", hier))
            except Exception as exc:  # noqa: BLE001
                await queue.put(("error", exc))

        intro = "Планирую задачу"
        yield _sse_chunk(
            cid=cid,
            model=model,
            delta={"role": "assistant", "content": "**Agent**\n"}
            if progress_in_content
            else {"role": "assistant"},
        )
        if progress_in_content:
            yield _sse_chunk(cid=cid, model=model, delta={"content": f"· {intro}\n"})
        else:
            yield _sse_chunk(
                cid=cid,
                model=model,
                delta={"reasoning_content": f"{intro}\n"},
            )

        task = asyncio.create_task(work())
        hier_meta = dict(base_meta)
        final_text = ""
        failed: Exception | None = None

        while True:
            kind, payload = await queue.get()
            if kind == "progress":
                from progress_ui import humanize_line, stream_step

                line = humanize_line(str(payload))
                if not line:
                    continue
                if progress_in_content:
                    yield _sse_chunk(
                        cid=cid, model=model, delta={"content": stream_step(line)}
                    )
                else:
                    yield _sse_chunk(
                        cid=cid,
                        model=model,
                        delta={"reasoning_content": stream_step(line)},
                    )
            elif kind == "done":
                hier = payload
                hier_meta = {**base_meta, **hier.meta}
                final_text = hier.content or ""
                REQUESTS.labels(
                    tier=tier, lane=lane, model="hierarchical", status="200"
                ).inc()
                DURATION.labels(tier=tier, model="hierarchical").observe(
                    time.perf_counter() - t0
                )
                orchestrator.after_request(
                    session_id,
                    prompt_hash=prompt_hash,
                    success=hier.ok,
                    escalate=False,
                )
                break
            elif kind == "error":
                failed = payload
                break

        await task

        if failed is not None:
            err = (str(failed) or type(failed).__name__)[:200]
            hier_meta["hierarchical_error"] = err
            note = (
                f"- hierarchical сорвался ({err}); "
                f"отвечаю одной моделью `{fallback_model}`…\n"
            )
            if progress_in_content:
                yield _sse_chunk(cid=cid, model=model, delta={"content": note})
                yield _sse_chunk(cid=cid, model=model, delta={"content": "\n---\n\n"})
            try:
                async with httpx.AsyncClient(timeout=120.0) as client:
                    resp = await client.post(
                        f"{LITELLM_URL}/v1/chat/completions",
                        headers={
                            "Authorization": f"Bearer {LITELLM_KEY}",
                            "Content-Type": "application/json",
                        },
                        json={
                            "model": fallback_model,
                            "messages": [
                                {"role": "user", "content": user_text[-12000:]}
                            ],
                            "max_tokens": 2048,
                            "stream": False,
                        },
                    )
                if resp.status_code >= 400:
                    raise RuntimeError(f"fallback HTTP {resp.status_code}")
                data = resp.json()
                msg = ((data.get("choices") or [{}])[0].get("message") or {})
                final_text = str(
                    msg.get("content")
                    or msg.get("reasoning_content")
                    or msg.get("reasoning")
                    or ""
                )
                hier_meta["fallback_model"] = fallback_model
                REQUESTS.labels(
                    tier=tier, lane=lane, model=fallback_model, status="200"
                ).inc()
                orchestrator.after_request(
                    session_id,
                    prompt_hash=prompt_hash,
                    success=bool(final_text),
                    escalate=False,
                )
            except Exception as fb_exc:  # noqa: BLE001
                fb_err = (str(fb_exc) or type(fb_exc).__name__)[:200]
                yield _sse_chunk(
                    cid=cid,
                    model=model,
                    delta={
                        "content": f"ошибка hierarchical: {err}; "
                        f"fallback тоже: {fb_err}\n"
                    },
                )
                REQUESTS.labels(
                    tier=tier, lane=lane, model="hierarchical", status="500"
                ).inc()
                orchestrator.after_request(
                    session_id,
                    prompt_hash=prompt_hash,
                    success=False,
                    escalate=False,
                )
                yield _sse_chunk(cid=cid, model=model, delta={}, finish_reason="stop")
                yield b"data: [DONE]\n\n"
                return

        if progress_in_content and failed is None:
            yield _sse_chunk(cid=cid, model=model, delta={"content": "\n---\n\n"})
        step = 240
        for i in range(0, len(final_text), step):
            yield _sse_chunk(
                cid=cid, model=model, delta={"content": final_text[i : i + step]}
            )
        done_chunk = {
            "id": cid,
            "object": "chat.completion.chunk",
            "created": int(time.time()),
            "model": model if failed is None else fallback_model,
            "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
            "x_router_meta": hier_meta,
        }
        yield f"data: {json.dumps(done_chunk, ensure_ascii=False)}\n\n".encode("utf-8")
        yield b"data: [DONE]\n\n"

    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={"X-Router-Meta": _meta_header(base_meta)},
    )


def _quality_mode(header: str | None, body: dict[str, Any]) -> str | None:
    if header:
        mode = header.strip().lower()
        if mode in ("auto", "economy", "balanced", "max"):
            return mode
    meta = body.get("metadata") or {}
    if isinstance(meta, dict):
        mode = str(meta.get("quality_mode", "")).lower()
        if mode in ("auto", "economy", "balanced", "max"):
            return mode
    return None


def _session_id(body: dict[str, Any]) -> str:
    meta = body.get("metadata") or {}
    if isinstance(meta, dict) and meta.get("session_id"):
        return str(meta["session_id"])
    if body.get("user"):
        return str(body["user"])
    return ""


def _router_meta(decision, *, requested: str) -> dict[str, Any]:
    return {
        "tier": decision.tier.value,
        "lane": decision.lane,
        "model": decision.model,
        "escalation_level": decision.escalation_level,
        "quality_mode": decision.quality_mode,
        "confidence": round(decision.confidence, 3),
        "requested_model": requested,
        "delegated_internal": decision.delegated_internal,
        "classifier_source": decision.classifier_source,
    }


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/metrics")
async def metrics() -> Response:
    body, content_type = metrics_payload()
    return Response(content=body, media_type=content_type)


@app.post("/admin/reload-config")
async def admin_reload(authorization: str | None = Header(default=None)) -> dict[str, str]:
    _auth_or_403(authorization)
    reload_configs()
    global orchestrator  # noqa: PLW0603
    orchestrator = Orchestrator(session_store)
    return {"status": "reloaded"}


@app.get("/v1/models")
async def list_models(authorization: str | None = Header(default=None)) -> dict[str, Any]:
    _auth_or_403(authorization)
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.get(
            f"{LITELLM_URL}/v1/models",
            headers={"Authorization": f"Bearer {LITELLM_KEY}"},
        )
    if resp.status_code >= 400:
        raise HTTPException(status_code=resp.status_code, detail=resp.text)
    return resp.json()


@app.post("/classify")
async def classify_debug(
    request: Request,
    authorization: str | None = Header(default=None),
    x_ai_quality: str | None = Header(default=None, alias="X-AI-Quality"),
) -> dict[str, Any]:
    _auth_or_403(authorization)
    body = await request.json()
    messages = body.get("messages") or []
    text = _extract_text(messages)
    decision = await orchestrator.resolve(
        messages,
        quality_mode=_quality_mode(x_ai_quality, body),
        session_id=_session_id(body),
        text=text,
        has_image=_has_image(messages),
        token_estimate=_estimate_tokens(text),
    )
    CLASSIFY.labels(tier=decision.tier.value).inc()
    if decision.classifier_source == "gigachat":
        CLASSIFY_LLM.labels(tier=decision.tier.value, status="ok").inc()
    return {
        "tier": decision.tier.value,
        "lane": decision.lane,
        "model": decision.model,
        "x_router_meta": _router_meta(decision, requested=body.get("model", DEFAULT_MODEL)),
    }


def _sanitize_reasoning_fields(obj: dict[str, Any]) -> bool:
    """Normalize reasoning fields for Zed: one key, promote into content if empty."""
    changed = False
    for choice in obj.get("choices") or []:
        if not isinstance(choice, dict):
            continue
        for key in ("delta", "message"):
            block = choice.get(key)
            if not isinstance(block, dict):
                continue
            reasoning = block.get("reasoning")
            reasoning_content = block.get("reasoning_content")
            if reasoning is not None:
                if reasoning_content is None:
                    block["reasoning_content"] = reasoning
                    reasoning_content = reasoning
                    changed = True
                del block["reasoning"]
                changed = True
            if "reasoning_details" in block:
                del block["reasoning_details"]
                changed = True
            psf = block.get("provider_specific_fields")
            if isinstance(psf, dict) and "reasoning" in psf:
                del psf["reasoning"]
                changed = True
                if not psf:
                    del block["provider_specific_fields"]
            # Novita/Qwen3 often puts the visible answer only in reasoning_content
            content = block.get("content")
            rc = block.get("reasoning_content")
            if (not isinstance(content, str) or not content.strip()) and isinstance(
                rc, str
            ) and rc.strip():
                block["content"] = rc
                changed = True
    return changed


def _sanitize_sse_line(line: bytes) -> bytes:
    if not line.startswith(b"data:"):
        return line
    payload = line[5:].strip()
    if payload == b"[DONE]":
        return line
    try:
        obj = json.loads(payload)
    except json.JSONDecodeError:
        return line
    if not isinstance(obj, dict) or not _sanitize_reasoning_fields(obj):
        return line
    return b"data: " + json.dumps(obj, ensure_ascii=False).encode("utf-8")


async def _iter_sanitized_sse(resp: httpx.Response):
    buf = b""
    try:
        async for chunk in resp.aiter_bytes():
            buf += chunk
            while True:
                split_at = buf.find(b"\n")
                if split_at < 0:
                    break
                line = buf[:split_at]
                buf = buf[split_at + 1 :]
                line = line.rstrip(b"\r")
                if not line:
                    yield b"\n"
                    continue
                yield _sanitize_sse_line(line) + b"\n"
        if buf:
            yield _sanitize_sse_line(buf.rstrip(b"\r")) + b"\n"
    except httpx.HTTPError as exc:
        # Upstream drop mid-SSE must not crash ASGI → Zed "unexpected EOF"
        log.warning("upstream SSE aborted: %s", exc)
        yield b"data: [DONE]\n\n"


async def _forward_litellm(
    forward: dict[str, Any],
    *,
    stream: bool,
    decision_meta: dict[str, Any],
) -> Any:
    headers = {
        "Authorization": f"Bearer {LITELLM_KEY}",
        "Content-Type": "application/json",
    }
    start = time.perf_counter()
    tier = decision_meta.get("tier", "UNKNOWN")
    model = forward.get("model", DEFAULT_MODEL)
    lane = decision_meta.get("lane", "?")

    if stream:
        client = httpx.AsyncClient(timeout=300.0)
        try:
            req = client.build_request(
                "POST",
                f"{LITELLM_URL}/v1/chat/completions",
                headers=headers,
                content=json.dumps(forward),
            )
            resp = await client.send(req, stream=True)

            if resp.status_code >= 400:
                body = await resp.aread()
                await resp.aclose()
                await client.aclose()
                status = str(resp.status_code)
                REQUESTS.labels(tier=tier, lane=lane, model=model, status=status).inc()
                DURATION.labels(tier=tier, model=model).observe(time.perf_counter() - start)
                try:
                    content = json.loads(body)
                except json.JSONDecodeError:
                    content = {"error": body.decode("utf-8", errors="replace")}
                return JSONResponse(status_code=resp.status_code, content=content)

            async def event_stream():
                try:
                    async for chunk in _iter_sanitized_sse(resp):
                        yield chunk
                finally:
                    await resp.aclose()
                    await client.aclose()

            REQUESTS.labels(
                tier=tier,
                lane=lane,
                model=model,
                status=str(resp.status_code),
            ).inc()
            DURATION.labels(tier=tier, model=model).observe(time.perf_counter() - start)
            return StreamingResponse(
                event_stream(),
                status_code=resp.status_code,
                media_type=resp.headers.get("content-type", "text/event-stream"),
                headers={"X-Router-Meta": _meta_header(decision_meta)},
            )
        except Exception:
            await client.aclose()
            raise

    async with httpx.AsyncClient(timeout=300.0) as client:
        resp = await client.post(
            f"{LITELLM_URL}/v1/chat/completions",
            headers=headers,
            json=forward,
        )

    status = str(resp.status_code)
    REQUESTS.labels(tier=tier, lane=lane, model=model, status=status).inc()
    DURATION.labels(tier=tier, model=model).observe(time.perf_counter() - start)

    if resp.status_code >= 400:
        return JSONResponse(status_code=resp.status_code, content=resp.json())

    data = resp.json()
    if isinstance(data, dict):
        _sanitize_reasoning_fields(data)
        data["x_router_meta"] = decision_meta
    return data



async def _forward_agent_executor(
    body: dict[str, Any],
    messages: list[dict[str, Any]],
    *,
    plan: dict[str, Any] | None,
    base_meta: dict[str, Any],
    stream: bool,
    phase: str,
) -> Any:
    from path_resolve import (
        apply_deterministic_path_index,
        enrich_plan_from_discovery_tools,
        force_find_path_kickstart,
        path_resolve_needed,
        pick_path_resolve_model,
    )

    cfg = _hier_cfg()
    plan = sanitize_plan_paths(plan) if plan else plan
    # Re-apply index every turn (plan from Zed transcript loses path_source)
    if cfg.get("path_resolve_enabled", True):
        plan, unresolved = apply_deterministic_path_index(plan)
        if unresolved and not messages_have_tool_activity(messages):
            log.info("path_index miss → queries=%s", unresolved)
    if messages_have_tool_activity(messages):
        plan = enrich_plan_from_discovery_tools(plan, messages)

    # DevOps: live TCP/HTTP facts before blind Traefik edits
    if cfg.get("runtime_probe_enabled", True) and isinstance(plan, dict):
        try:
            from runtime_probe import attach_runtime_probe, runtime_facts_line

            plan = attach_runtime_probe(
                plan, user_text=_last_user_text(messages), cfg=cfg
            )
            line = runtime_facts_line(plan)
            if line:
                log.info("%s", line)
        except Exception as exc:  # noqa: BLE001
            log.warning("runtime_probe skipped: %s", exc)

    path_mode = bool(cfg.get("path_resolve_enabled", True)) and path_resolve_needed(
        plan
    )

    if path_mode:
        phase = "path_resolve"
        executor = pick_path_resolve_model(cfg)
    else:
        executor = pick_agent_executor_model(cfg)

    # After edit_file already ran — stop the Zed loop (was infinite force-edit)
    if (
        not path_mode
        and phase == "tool_loop"
        and should_stop_after_edit(messages, cfg)
    ):
        log.info("stop after edit_file (prevent edit loop)")
        data = stop_after_edit_completion(model=str(executor), messages=messages)
        meta = merge_agent_meta(
            base_meta, executor=executor, plan=plan, phase="edit_done"
        )
        meta["executor_stop_after_edit"] = True
        REQUESTS.labels(
            tier=meta.get("tier", "UNKNOWN"),
            lane=meta.get("lane", "?"),
            model=str(executor),
            status="200",
        ).inc()
        if not stream:
            data["x_router_meta"] = meta
            return JSONResponse(content=data)
        cid = f"hier-edit-done-{int(time.time() * 1000)}"

        async def edit_done_stream():
            for chunk in completion_to_sse_chunks(
                data, cid=cid, model=str(data.get("model") or executor)
            ):
                yield chunk

        return StreamingResponse(
            edit_done_stream(),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",
                "X-Router-Meta": _meta_header(meta),
            },
        )

    # After successful read_file: skip Novita DeepSeek (408→400 loop) → edit_file
    if (
        not path_mode
        and phase == "tool_loop"
        and should_force_edit_after_read(messages, cfg, plan=plan)
    ):
        log.info("force edit_file after successful read (skip DeepSeek)")
        data = force_edit_after_read_completion(
            plan=plan,
            tools=body.get("tools") if isinstance(body.get("tools"), list) else [],
            model=str(executor),
            messages=messages,
        )
        meta = merge_agent_meta(
            base_meta, executor=executor, plan=plan, phase="force_edit"
        )
        meta["executor_force_edit_after_read"] = True
        facts = (plan or {}).get("runtime_facts") if isinstance(plan, dict) else None
        if isinstance(facts, dict) and facts.get("suggested_backend_url"):
            meta["runtime_targeted_edit"] = facts.get("suggested_backend_url")
        REQUESTS.labels(
            tier=meta.get("tier", "UNKNOWN"),
            lane=meta.get("lane", "?"),
            model=str(executor),
            status="200",
        ).inc()
        if not stream:
            data["x_router_meta"] = meta
            return JSONResponse(content=data)
        cid = f"hier-force-edit-{int(time.time() * 1000)}"
        from progress_ui import short_path, stream_step
        from agent_hier import _last_successful_read_path

        path = _last_successful_read_path(messages)
        edit_label = (
            f"Правлю порт Traefik → `{facts.get('suggested_backend_url')}`"
            if isinstance(facts, dict) and facts.get("suggested_backend_url")
            else (f"Правлю `{short_path(path)}`" if path else "Правлю файл")
        )

        async def force_edit_stream():
            yield _sse_chunk(
                cid=cid,
                model=str(executor),
                delta={
                    "role": "assistant",
                    "content": stream_step(edit_label),
                },
            )
            for chunk in completion_to_sse_chunks(
                data, cid=cid, model=str(data.get("model") or executor)
            ):
                yield chunk

        return StreamingResponse(
            force_edit_stream(),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",
                "X-Router-Meta": _meta_header(meta),
            },
        )

    # Stream-to-Novita often aborts mid-SSE (Zed: unexpected EOF). Prefer non-stream.
    use_upstream_stream = bool(cfg.get("executor_use_stream", False))
    forward = prepare_agent_executor_forward(
        body,
        messages,
        plan,
        cfg,
        executor=executor,
        stream=use_upstream_stream,
        path_resolve=path_mode,
    )

    def _synth_fallback(reason: str) -> dict[str, Any]:
        if path_mode:
            return force_find_path_kickstart(
                model=str(executor),
                tools=forward.get("tools") or [],
                queries=list((plan or {}).get("path_resolve_queries") or [])
                or ["dynamic_conf.yml"],
            )
        return executor_fallback_completion(
            plan=plan,
            tools=forward.get("tools") or [],
            model=executor,
            messages=messages,
            cfg=cfg,
            reason=reason,
        )
    budget = int(
        cfg.get("executor_tool_loop_chars")
        if phase == "tool_loop"
        else cfg.get("executor_input_chars")
        or 16000
    )
    ctx_line, ctx_meta = context_fill_for_forward(
        forward, cfg, budget_chars=budget
    )
    meta = merge_agent_meta(base_meta, executor=executor, plan=plan, phase=phase)
    meta.update(
        {
            "ctx_used_chars": ctx_meta["ctx_used_chars"],
            "ctx_budget_chars": ctx_meta["ctx_budget_chars"],
            "ctx_pct": ctx_meta["ctx_pct"],
            "ctx_est_tokens": ctx_meta["ctx_est_tokens"],
        }
    )
    # prepare_agent_executor_forward may escalate model after edit failures
    actual_model = str(forward.get("model") or executor)
    if actual_model != str(executor):
        meta["executor_escalated"] = True
        meta["executor_model_requested"] = str(executor)
        meta["executor_model"] = actual_model
        meta["selected_model"] = actual_model
        meta["planner_model"] = meta.get("planner_model")
        meta["role_cost"] = {
            "planner": meta.get("planner_model"),
            "executor": actual_model,
            "escalated": True,
        }
        executor = actual_model
        log.info("hierarchical_agent escalate → %s", actual_model)
    else:
        meta["role_cost"] = {
            "planner": meta.get("planner_model"),
            "executor": str(executor),
            "escalated": False,
        }
    log.info(
        "hierarchical_agent phase=%s executor=%s tools=%s msgs=%s upstream_stream=%s %s",
        phase,
        executor,
        len(forward.get("tools") or []),
        len(forward["messages"]),
        use_upstream_stream,
        ctx_line,
    )
    try:
        if use_upstream_stream:
            return await _forward_litellm(
                forward, stream=stream, decision_meta=meta
            )

        # Gateway fails first via httpx; body timeout slightly under client so LiteLLM
        # cancels cleanly — we still convert 408 → synthetic (never expose to Zed).
        body_timeout = float(
            forward.get("timeout") or cfg.get("executor_timeout_sec") or 55
        )
        client_timeout = body_timeout + 20.0
        # Ensure LiteLLM sees the same budget (otherwise deployment 120 hangs us)
        forward = dict(forward)
        forward["timeout"] = body_timeout
        async with httpx.AsyncClient(
            timeout=httpx.Timeout(client_timeout, connect=15.0)
        ) as client:
            resp = await client.post(
                f"{LITELLM_URL}/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {LITELLM_KEY}",
                    "Content-Type": "application/json",
                },
                json=forward,
            )
        if resp.status_code >= 400:
            body_l = (resp.text or "").lower()
            is_timeout = resp.status_code in (408, 504, 502) or "timeout" in body_l
            is_bad_req = resp.status_code == 400 or "invalid request" in body_l
            # Mid tool_loop: one real model retry (no synthetic) before stop
            if (
                is_timeout
                and phase == "tool_loop"
                and messages_have_tool_activity(messages)
                and cfg.get("executor_midloop_model_retry", True)
            ):
                from agent_hier import pick_agent_escalate_model

                esc = pick_agent_escalate_model(cfg)
                already_esc = str(forward.get("model") or executor) == str(esc)
                if already_esc:
                    log.warning(
                        "mid-loop skip retry — already on escalate model %s → synth",
                        esc,
                    )
                else:
                    log.warning(
                        "litellm HTTP %s mid tool_loop → one model retry (no synthetic)",
                        resp.status_code,
                    )
                    retry_fwd = dict(forward)
                    if cfg.get("executor_midloop_escalate", True):
                        retry_fwd["model"] = esc
                        meta["executor_midloop_escalate_model"] = esc
                        log.warning("mid-loop escalate → %s", esc)
                    tools_r = list(retry_fwd.get("tools") or [])
                    prefer = []
                    for t in tools_r:
                        n = str(
                            (
                                (t.get("function") or {})
                                if isinstance(t, dict)
                                else {}
                            ).get("name")
                            or ""
                        ).lower()
                        if "read" in n or "edit" in n or "write" in n:
                            prefer.append(t)
                    if prefer:
                        retry_fwd["tools"] = prefer[:3]
                    retry_fwd["tool_choice"] = "required"
                    # Re-sanitize messages (null content → 400)
                    from agent_hier import sanitize_outbound_messages

                    retry_fwd["messages"] = sanitize_outbound_messages(
                        retry_fwd.get("messages") or []
                    )
                    async with httpx.AsyncClient(
                        timeout=httpx.Timeout(client_timeout, connect=15.0)
                    ) as client2:
                        resp2 = await client2.post(
                            f"{LITELLM_URL}/v1/chat/completions",
                            headers={
                                "Authorization": f"Bearer {LITELLM_KEY}",
                                "Content-Type": "application/json",
                            },
                            json=retry_fwd,
                        )
                    if resp2.status_code < 400:
                        data2 = resp2.json()
                        if isinstance(data2, dict):
                            _sanitize_reasoning_fields(data2)
                        if completion_has_tool_calls(data2):
                            log.info("mid-loop model retry OK → tool_calls")
                            meta["executor_midloop_retry"] = True
                            resp = resp2
                            data = rewrite_redundant_reread_completion(
                                data2,
                                messages=messages,
                                plan=plan,
                                tools=forward.get("tools") or [],
                                model=str(retry_fwd.get("model") or executor),
                                cfg=cfg,
                            )
                            choice0 = (data.get("choices") or [{}])[0]
                            msg0 = choice0.get("message") or {}
                            tcs = (
                                msg0.get("tool_calls")
                                if isinstance(msg0, dict)
                                else None
                            )
                            tc_names = []
                            if isinstance(tcs, list):
                                for tc in tcs:
                                    if isinstance(tc, dict):
                                        tc_names.append(
                                            str(
                                                (
                                                    (tc.get("function") or {}).get(
                                                        "name"
                                                    )
                                                )
                                                or "?"
                                            )
                                        )
                            log.info(
                                "executor result phase=%s finish=%s tools=%s",
                                phase,
                                choice0.get("finish_reason"),
                                tc_names,
                            )
                            REQUESTS.labels(
                                tier=meta.get("tier", "UNKNOWN"),
                                lane=meta.get("lane", "?"),
                                model=executor,
                                status=str(resp.status_code),
                            ).inc()
                            if not stream:
                                data["x_router_meta"] = meta
                                return JSONResponse(content=data)
                            cid = f"hier-agent-{int(time.time() * 1000)}"

                            async def event_stream_retry():
                                if cfg.get("show_context_fill", False):
                                    yield _sse_chunk(
                                        cid=cid,
                                        model=str(executor),
                                        delta={
                                            "role": "assistant",
                                            "content": (
                                                f"- {ctx_line}\n"
                                                "- litellm mid-loop retry OK\n"
                                            ),
                                        },
                                    )
                                for chunk in completion_to_sse_chunks(
                                    data,
                                    cid=cid,
                                    model=str(data.get("model") or executor),
                                ):
                                    yield chunk

                            return StreamingResponse(
                                event_stream_retry(),
                                media_type="text/event-stream",
                                headers={
                                    "Cache-Control": "no-cache",
                                    "X-Accel-Buffering": "no",
                                    "X-Router-Meta": _meta_header(meta),
                                },
                            )
                    log.warning(
                        "mid-loop model retry still failed status=%s",
                        getattr(resp2, "status_code", "?"),
                    )

            if cfg.get("executor_synthetic_on_timeout", True) and (
                is_timeout or is_bad_req
            ):
                log.warning(
                    "litellm HTTP %s (timeout/400) → executor fallback",
                    resp.status_code,
                )
                data = _synth_fallback(f"litellm_{resp.status_code}")
                # Prefer force-edit if we already have file body
                if (
                    phase == "tool_loop"
                    and should_force_edit_after_read(messages, cfg, plan=plan)
                    and not completion_has_tool_calls(data)
                ):
                    data = force_edit_after_read_completion(
                        plan=plan,
                        tools=forward.get("tools") or [],
                        model=str(executor),
                        messages=messages,
                    )
                    meta["executor_force_edit_after_read"] = True
                meta["executor_synthetic"] = True
                fr0 = ((data.get("choices") or [{}])[0] or {}).get("finish_reason")
                meta["executor_synthetic_reason"] = (
                    f"litellm_{resp.status_code}_abort"
                    if fr0 == "stop"
                    else f"litellm_{resp.status_code}"
                )
                if not stream:
                    data["x_router_meta"] = meta
                    return JSONResponse(content=data)
                cid = f"hier-agent-synth-{int(time.time() * 1000)}"

                async def timeout_synth_stream():
                    if cfg.get("show_context_fill", False):
                        label = (
                            "stop (no mid-loop synthetic)\n"
                            if fr0 == "stop"
                            else "kickstart/edit\n"
                        )
                        yield _sse_chunk(
                            cid=cid,
                            model=str(executor),
                            delta={
                                "role": "assistant",
                                "content": (
                                    f"- {ctx_line}\n"
                                    f"- litellm {resp.status_code} → {label}"
                                ),
                            },
                        )
                    for chunk in completion_to_sse_chunks(
                        data, cid=cid, model=executor
                    ):
                        yield chunk

                return StreamingResponse(
                    timeout_synth_stream(),
                    media_type="text/event-stream",
                    headers={
                        "Cache-Control": "no-cache",
                        "X-Accel-Buffering": "no",
                        "X-Router-Meta": _meta_header(meta),
                    },
                )
            raise RuntimeError(
                f"litellm HTTP {resp.status_code}: {resp.text[:300]}"
            )
        data = resp.json()
        if not isinstance(data, dict):
            raise RuntimeError("litellm returned non-object JSON")
        if isinstance(data, dict):
            _sanitize_reasoning_fields(data)

        choice0 = (data.get("choices") or [{}])[0]
        msg0 = choice0.get("message") or {}
        tcs = msg0.get("tool_calls") if isinstance(msg0, dict) else None
        tc_names = []
        if isinstance(tcs, list):
            for tc in tcs:
                if isinstance(tc, dict):
                    tc_names.append(
                        str(((tc.get("function") or {}).get("name")) or "?")
                    )
        if completion_has_tool_calls(data):
            data = rewrite_redundant_reread_completion(
                data,
                messages=messages,
                plan=plan,
                tools=forward.get("tools") or [],
                model=executor,
                cfg=cfg,
            )
            choice0 = (data.get("choices") or [{}])[0]
            msg0 = choice0.get("message") or {}
            tcs = msg0.get("tool_calls") if isinstance(msg0, dict) else None
            tc_names = []
            if isinstance(tcs, list):
                for tc in tcs:
                    if isinstance(tc, dict):
                        tc_names.append(
                            str(((tc.get("function") or {}).get("name")) or "?")
                        )
        log.info(
            "executor result phase=%s finish=%s tools=%s",
            phase,
            choice0.get("finish_reason"),
            tc_names,
        )

        # Model returned prose / empty tools — unblock Zed
        if (
            cfg.get("executor_synthetic_on_timeout", True)
            and not completion_has_tool_calls(data)
        ):
            log.warning(
                "executor no tool_calls phase=%s %s",
                phase,
                completion_preview(data),
            )
            # Mid-loop: escalate already tried on 408; for empty tools escalate once
            if (
                phase == "tool_loop"
                and messages_have_tool_activity(messages)
                and cfg.get("executor_no_tools_escalate", True)
            ):
                from agent_hier import pick_agent_escalate_model

                esc = pick_agent_escalate_model(cfg)
                retry_fwd = dict(forward)
                retry_fwd["model"] = esc
                retry_fwd["tool_choice"] = "required"
                try:
                    async with httpx.AsyncClient(
                        timeout=httpx.Timeout(client_timeout, connect=15.0)
                    ) as client3:
                        resp3 = await client3.post(
                            f"{LITELLM_URL}/v1/chat/completions",
                            headers={
                                "Authorization": f"Bearer {LITELLM_KEY}",
                                "Content-Type": "application/json",
                            },
                            json=retry_fwd,
                        )
                    if resp3.status_code < 400:
                        data3 = resp3.json()
                        if isinstance(data3, dict) and completion_has_tool_calls(data3):
                            data = data3
                            _sanitize_reasoning_fields(data)
                            meta["executor_no_tools_escalated"] = True
                except Exception as exc:  # noqa: BLE001
                    log.warning("no_tools escalate failed: %s", exc)
            if not completion_has_tool_calls(data):
                data = _synth_fallback(
                    f"no_tool_calls:{choice0.get('finish_reason')}"
                )
                meta["executor_synthetic"] = True
                fr_nt = ((data.get("choices") or [{}])[0] or {}).get("finish_reason")
                meta["executor_synthetic_reason"] = (
                    "no_tool_calls_abort" if fr_nt == "stop" else "no_tool_calls"
                )
                if fr_nt == "stop":
                    log.warning("tool_loop fallback stop (kickstart_only / limit)")
                else:
                    log.warning(
                        "no tool_calls (finish=%s) → kickstart/stop",
                        choice0.get("finish_reason"),
                    )

        REQUESTS.labels(
            tier=meta.get("tier", "UNKNOWN"),
            lane=meta.get("lane", "?"),
            model=executor,
            status=str(resp.status_code),
        ).inc()

        if not stream:
            data["x_router_meta"] = meta
            return JSONResponse(content=data)

        cid = f"hier-agent-{int(time.time() * 1000)}"

        async def event_stream():
            if cfg.get("show_context_fill", False):
                yield _sse_chunk(
                    cid=cid,
                    model=str(executor),
                    delta={"role": "assistant", "content": f"- {ctx_line}\n"},
                )
            for chunk in completion_to_sse_chunks(
                data, cid=cid, model=str(data.get("model") or executor)
            ):
                yield chunk

        return StreamingResponse(
            event_stream(),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",
                "X-Router-Meta": _meta_header(meta),
            },
        )
    except Exception as exc:  # noqa: BLE001
        err = (str(exc) or type(exc).__name__)[:240]
        log.exception("agent executor failed phase=%s: %s", phase, err)
        meta["executor_error"] = err
        # Always synthetic on timeout/errors when enabled — never raw 408 to Zed
        if cfg.get("executor_synthetic_on_timeout", True):
            data = _synth_fallback(f"exception:{err[:80]}")
            cid = f"hier-agent-synth-{int(time.time() * 1000)}"
            meta["executor_synthetic"] = True
            fr_ex = ((data.get("choices") or [{}])[0] or {}).get("finish_reason")
            meta["executor_synthetic_reason"] = (
                "exception_abort" if fr_ex == "stop" else "exception"
            )
            if not stream:
                data["x_router_meta"] = meta
                return JSONResponse(content=data)

            async def synth_stream():
                if cfg.get("show_context_fill", False):
                    label = "stop" if fr_ex == "stop" else "kickstart"
                    yield _sse_chunk(
                        cid=cid,
                        model=str(executor),
                        delta={
                            "role": "assistant",
                            "content": (
                                f"- executor error → {label} ({err[:80]})\n"
                            ),
                        },
                    )
                for chunk in completion_to_sse_chunks(
                    data, cid=cid, model=executor
                ):
                    yield chunk

            return StreamingResponse(
                synth_stream(),
                media_type="text/event-stream",
                headers={
                    "Cache-Control": "no-cache",
                    "X-Accel-Buffering": "no",
                    "X-Router-Meta": _meta_header(meta),
                },
            )
        msg = (
            f"Executor `{executor}` не ответил ({err}). "
            "Роутер на IFT жив — это таймаут upstream, не «серверы недоступны»."
        )
        if stream:
            return _sse_completion(msg, model="hierarchical", meta=meta)
        return _completion_from_text(msg, model="hierarchical", meta=meta)


async def _agent_hierarchical_chat(
    *,
    body: dict[str, Any],
    messages: list[dict[str, Any]],
    last_user: str,
    decision: Any,
    meta: dict[str, Any],
    session_id: str,
    prompt_hash: str,
) -> Any:
    """Plan/confirm on gateway; executor model returns tool_calls for Zed.

    Critical: when stream=True we MUST return StreamingResponse immediately and
    emit progress — never await the full plan before the first SSE byte.
    """
    stream = bool(body.get("stream", False))
    cfg = _hier_cfg()
    runner = HierarchicalRunner(litellm_url=LITELLM_URL, litellm_key=LITELLM_KEY)

    # Zed tool loop mid-flight: never text-synthesize
    if messages_have_tool_activity(messages):
        plan = find_pending_plan(messages)
        return await _forward_agent_executor(
            body,
            messages,
            plan=plan,
            base_meta=meta,
            stream=stream,
            phase="tool_loop",
        )

    if stream:
        return await stream_agent_plan_then_act(
            runner=runner,
            last_user=last_user,
            quality_mode=decision.quality_mode,
            session_id=session_id,
            messages=messages,
            body=body,
            meta=meta,
            decision=decision,
            prompt_hash=prompt_hash,
            litellm_url=LITELLM_URL,
            litellm_key=LITELLM_KEY,
            cfg=cfg,
            find_pending_plan=find_pending_plan,
            plan_payload_from_meta=plan_payload_from_meta,
            inject_plan_context=inject_plan_context,
            pick_agent_executor_model=pick_agent_executor_model,
            prepare_agent_executor_forward=prepare_agent_executor_forward,
            completion_to_sse_chunks=completion_to_sse_chunks,
            synthetic_first_tool_completion=synthetic_first_tool_completion,
            executor_fallback_completion=executor_fallback_completion,
            meta_header=meta_header,
            orchestrator=orchestrator,
            requests_metric=REQUESTS,
        )

    # Non-stream: still catch errors — never raw 500 hang for Zed
    try:
        hier = await runner.run(
            last_user,
            quality_mode=decision.quality_mode,
            session_id=session_id,
            messages=messages,
            agent_mode=True,
        )
    except Exception as exc:  # noqa: BLE001
        log.exception("agent hierarchical plan failed: %s", exc)
        meta["hierarchical_error"] = (str(exc) or type(exc).__name__)[:200]
        return await _forward_agent_executor(
            body,
            messages,
            plan=None,
            base_meta=meta,
            stream=False,
            phase="plan_error_fallback",
        )

    hier_meta = {**meta, **hier.meta}
    if hier.meta.get("agent_execute"):
        plan = plan_payload_from_meta(hier.meta) or find_pending_plan(messages)
        orchestrator.after_request(
            session_id, prompt_hash=prompt_hash, success=True, escalate=False
        )
        return await _forward_agent_executor(
            body,
            messages,
            plan=plan,
            base_meta=hier_meta,
            stream=False,
            phase="execute",
        )

    content = hier.content or ""
    if bool(cfg.get("progress_in_content", True)):
        trail = hier.meta.get("progress") or []
        if isinstance(trail, list) and trail:
            content = _format_progress_block([str(x) for x in trail]) + content
    REQUESTS.labels(
        tier=decision.tier.value,
        lane=decision.lane,
        model="hierarchical_agent",
        status="200",
    ).inc()
    orchestrator.after_request(
        session_id, prompt_hash=prompt_hash, success=True, escalate=False
    )
    return _completion_from_text(content, model="hierarchical", meta=hier_meta)


@app.post("/v1/chat/completions")
async def chat_completions(
    request: Request,
    authorization: str | None = Header(default=None),
    x_ai_quality: str | None = Header(default=None, alias="X-AI-Quality"),
    x_ai_orchestrate: str | None = Header(default=None, alias="X-AI-Orchestrate"),
) -> Any:
    _auth_or_403(authorization)

    body: dict[str, Any] = await request.json()
    messages = body.get("messages") or []
    if not isinstance(messages, list):
        raise HTTPException(status_code=400, detail="messages must be a list")

    text_all = _extract_text(messages)
    last_user = _last_user_text(messages)
    has_image = _has_image(messages)
    session_id = _session_id(body)
    prompt_hash = hashlib.sha256(text_all.encode()).hexdigest()[:16]
    quality = _quality_mode(x_ai_quality, body)
    orch_hdr = _orchestrate_header(x_ai_orchestrate, body)

    requested_model = body.get("model", DEFAULT_MODEL)
    if requested_model in ("smart-router", "auto", ""):
        decision = await orchestrator.resolve(
            messages,
            quality_mode=quality,
            session_id=session_id,
            text=text_all,
            has_image=has_image,
            token_estimate=_estimate_tokens(text_all),
        )
        if decision.classifier_source == "gigachat":
            CLASSIFY_LLM.labels(tier=decision.tier.value, status="ok").inc()
        CLASSIFY.labels(tier=decision.tier.value).inc()
        target_model = decision.model
        meta = _router_meta(decision, requested=requested_model)

        if (
            not has_image
            and should_run_hierarchical(
                tier_value=decision.tier.value,
                header=orch_hdr,
                quality_mode=decision.quality_mode,
            )
        ):
            # Zed Agent Write: keep tool loop; hierarchical only plans/routes
            if request_has_tools(body):
                return await _agent_hierarchical_chat(
                    body=body,
                    messages=messages,
                    last_user=last_user,
                    decision=decision,
                    meta=meta,
                    session_id=session_id,
                    prompt_hash=prompt_hash,
                )

            runner = HierarchicalRunner(
                litellm_url=LITELLM_URL,
                litellm_key=LITELLM_KEY,
            )
            stream = bool(body.get("stream", False))
            if stream:
                return _sse_hierarchical_progress(
                    runner,
                    user_text=last_user,
                    quality_mode=decision.quality_mode,
                    session_id=session_id,
                    prompt_hash=prompt_hash,
                    base_meta=meta,
                    fallback_model=target_model,
                    messages=messages,
                )

            start = time.perf_counter()
            try:
                hier = await runner.run(
                    last_user,
                    quality_mode=decision.quality_mode,
                    session_id=session_id,
                    messages=messages,
                )
            except Exception as exc:  # noqa: BLE001
                log.exception("hierarchical failed, falling back to single route: %s", exc)
                meta["hierarchical_error"] = (str(exc) or type(exc).__name__)[:200]
            else:
                hier_meta = {**meta, **hier.meta}
                REQUESTS.labels(
                    tier=decision.tier.value,
                    lane=decision.lane,
                    model="hierarchical",
                    status="200",
                ).inc()
                DURATION.labels(tier=decision.tier.value, model="hierarchical").observe(
                    time.perf_counter() - start
                )
                orchestrator.after_request(
                    session_id,
                    prompt_hash=prompt_hash,
                    success=hier.ok,
                    escalate=False,
                )
                content = hier.content or ""
                if bool(_hier_cfg().get("progress_in_content", True)):
                    trail = hier.meta.get("progress") or []
                    if isinstance(trail, list) and trail:
                        content = _format_progress_block([str(x) for x in trail]) + content
                return _completion_from_text(
                    content, model="hierarchical", meta=hier_meta
                )
    else:
        target_model = requested_model
        meta = {"selected_model": target_model, "requested_model": requested_model}

    forward = dict(body)
    forward["model"] = target_model
    forward.setdefault("metadata", {})
    if isinstance(forward["metadata"], dict):
        forward["metadata"]["semantic_tier"] = meta.get("tier")
        forward["metadata"]["session_id"] = session_id or forward["metadata"].get("session_id")

    log.info(
        "route tier=%s lane=%s model=%s requested=%s",
        meta.get("tier"),
        meta.get("lane"),
        target_model,
        requested_model,
    )

    stream = bool(body.get("stream", False))
    result = await _forward_litellm(forward, stream=stream, decision_meta=meta)

    escalate = isinstance(result, JSONResponse) and result.status_code >= 429
    orchestrator.after_request(
        session_id,
        prompt_hash=prompt_hash,
        success=not escalate,
        escalate=escalate,
    )
    if escalate and meta.get("lane") != "C":
        ESCALATIONS.labels(from_lane=meta.get("lane", "?"), to_lane="next", reason="http_error").inc()

    return result