aleksey
a2d238d92e
Make Zed Agent closer to Cursor: deterministic DevOps path index, live Traefik port probe before blind edits, stop-after-edit, and quieter Russian progress.
202 lines
6.5 KiB
Python
202 lines
6.5 KiB
Python
"""Compact, readable progress lines for Zed agent chat."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import Any
|
||
|
||
_MODEL_LABEL = {
|
||
"novita-planner": "Max",
|
||
"novita-verifier": "verify",
|
||
"a-medium-code": "Coder",
|
||
"a-simple": "Llama",
|
||
"b-complex": "DeepSeek",
|
||
"c-complex": "Max",
|
||
"approved-plan": "план",
|
||
}
|
||
|
||
_NOISE = (
|
||
"ждём LiteLLM",
|
||
"payload:",
|
||
"ctx ",
|
||
"ctx-",
|
||
"worker #",
|
||
"волна ",
|
||
"параллельно",
|
||
"bytes≈",
|
||
"upstream_stream",
|
||
"non-stream",
|
||
)
|
||
|
||
|
||
def model_label(model: str | None) -> str:
|
||
if not model:
|
||
return "модель"
|
||
m = str(model)
|
||
return _MODEL_LABEL.get(m, m.split("/")[-1][:18])
|
||
|
||
|
||
def short_path(path: str | None, *, max_parts: int = 3) -> str:
|
||
if not path:
|
||
return ""
|
||
p = str(path).replace("/", "\\").rstrip("\\")
|
||
parts = [x for x in p.split("\\") if x]
|
||
if len(parts) <= max_parts:
|
||
return "\\".join(parts)
|
||
return "\\".join(parts[-max_parts:])
|
||
|
||
|
||
def plan_paths(plan: dict[str, Any] | None) -> list[str]:
|
||
out: list[str] = []
|
||
for item in (plan or {}).get("subtasks") or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
for p in item.get("paths") or []:
|
||
if isinstance(p, str) and p.strip() and p not in out:
|
||
out.append(p.strip())
|
||
return out
|
||
|
||
|
||
def is_noise(line: str) -> bool:
|
||
s = (line or "").strip()
|
||
if not s:
|
||
return True
|
||
low = s.lower()
|
||
return any(n.lower() in low for n in _NOISE)
|
||
|
||
|
||
def humanize_line(line: str) -> str | None:
|
||
"""Map internal progress → short Russian status. None = drop."""
|
||
s = (line or "").strip()
|
||
if not s or is_noise(s):
|
||
return None
|
||
low = s.lower()
|
||
|
||
if s in ("план…", "план...", "план"):
|
||
return "Планирую задачу"
|
||
if "план через" in low or low.startswith("hierarchical:"):
|
||
return "Планирую задачу"
|
||
if "утверждён" in low or "утвержден" in low:
|
||
m = re.search(r"(\d+)\s*подзадач", s)
|
||
n = m.group(1) if m else ""
|
||
return f"План утверждён{f' · {n} шаг(а)' if n else ''}"
|
||
if "отменён" in low or "отменен" in low:
|
||
return "План отменён"
|
||
if "правки от пользователя" in low or "пересборк" in low:
|
||
return "Пересобираю план по правкам"
|
||
if "ожидает утверждения" in low or "ждём утверждения" in low or "awaiting" in low:
|
||
return "План готов — подтверди («ok»)"
|
||
if "agent: план готов" in low or "executor с tools" in low:
|
||
return "Перехожу к выполнению"
|
||
if "готово за" in low:
|
||
return "Готово"
|
||
if "json битый" in low or "repair" in low:
|
||
return "Чиню формат плана"
|
||
if low.startswith("path_index") or "path_index hit" in low:
|
||
return None # shown via dedicated path line
|
||
if "path_resolve" in low or "поиск файл" in low:
|
||
return "Ищу файлы в проекте"
|
||
if "runtime_probe" in low or "runtime:" in low:
|
||
return "Проверяю runtime"
|
||
if "force edit" in low:
|
||
return "Правлю файл"
|
||
if "kickstart" in low:
|
||
return "Читаю файл"
|
||
if "escalate" in low or "mid-loop" in low:
|
||
return None
|
||
if "408" in s or "timeout" in low:
|
||
return "Таймаут модели — продолжаю иначе"
|
||
if "400" in s and "bad" in low:
|
||
return None
|
||
# strip technical prefixes
|
||
s = re.sub(r"^план:\s*", "", s, flags=re.I)
|
||
s = re.sub(r"^agent:\s*", "", s, flags=re.I)
|
||
if len(s) > 90:
|
||
s = s[:87] + "…"
|
||
return s
|
||
|
||
|
||
def format_progress_block(lines: list[str]) -> str:
|
||
"""Final/static progress block — numbered, no junk."""
|
||
steps: list[str] = []
|
||
seen: set[str] = set()
|
||
for raw in lines or []:
|
||
h = humanize_line(str(raw))
|
||
if not h or h in seen:
|
||
continue
|
||
seen.add(h)
|
||
steps.append(h)
|
||
if not steps:
|
||
return ""
|
||
body = "\n".join(f"{i}. {t}" for i, t in enumerate(steps, 1))
|
||
return f"**Ход**\n{body}\n\n---\n\n"
|
||
|
||
|
||
def stream_header() -> str:
|
||
return "**Agent**\n"
|
||
|
||
|
||
def stream_step(text: str) -> str:
|
||
return f"· {text}\n"
|
||
|
||
|
||
def execution_banner(
|
||
*,
|
||
path_mode: bool,
|
||
plan: dict[str, Any] | None,
|
||
model: str | None,
|
||
) -> str:
|
||
"""Shown once after plan approve — what happens next."""
|
||
paths = plan_paths(plan)
|
||
lines = ["", "---", "", "**Выполнение**"]
|
||
facts = (plan or {}).get("runtime_facts") if isinstance(plan, dict) else None
|
||
step = 1
|
||
if isinstance(facts, dict) and (facts.get("host") or facts.get("hint")):
|
||
host = facts.get("host") or "?"
|
||
status = facts.get("http_status")
|
||
suggest = facts.get("suggested_backend_url")
|
||
bit = f"{host}"
|
||
if status is not None:
|
||
bit += f" → HTTP {status}"
|
||
if suggest:
|
||
bit += f" · fix `{suggest}`"
|
||
lines.append(f"{step}. Runtime: {bit}")
|
||
step += 1
|
||
if paths:
|
||
shown = ", ".join(f"`{short_path(p)}`" for p in paths[:2])
|
||
src = ""
|
||
tasks = (plan or {}).get("subtasks") or []
|
||
if tasks and isinstance(tasks[0], dict):
|
||
ps = str(tasks[0].get("path_source") or "")
|
||
if ps == "index":
|
||
src = " · индекс"
|
||
elif ps == "find":
|
||
src = " · поиск"
|
||
lines.append(f"{step}. Файл{src}: {shown}")
|
||
step += 1
|
||
if path_mode:
|
||
lines.append(f"{step}. Уточняю путь в репозитории…")
|
||
else:
|
||
lines.append(f"{step}. Читаю → правлю")
|
||
elif path_mode:
|
||
lines.append(f"{step}. Ищу нужные файлы…")
|
||
else:
|
||
lines.append(f"{step}. Запуск ({model_label(model)})")
|
||
lines.append("")
|
||
return "\n".join(lines) + "\n"
|
||
|
||
|
||
def tool_status(tool_names: list[str] | None, *, path: str | None = None) -> str:
|
||
names = [str(n).lower() for n in (tool_names or []) if n]
|
||
if any("edit" in n or "write" in n for n in names):
|
||
base = "Правлю"
|
||
elif any("read" in n for n in names):
|
||
base = "Читаю"
|
||
elif any("find" in n or "list" in n or "grep" in n for n in names):
|
||
base = "Ищу"
|
||
else:
|
||
base = "Инструмент"
|
||
if path:
|
||
return stream_step(f"{base} `{short_path(path)}`")
|
||
return stream_step(base)
|