""" 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