From 935c75d078835f79457e92833aeb1691b7527093 Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Tue, 29 Sep 2026 20:20:15 +0000 Subject: [PATCH] =?UTF-8?q?fix(spend):=20BudgetSpent=20=D0=BF=D0=B8=D1=88?= =?UTF-8?q?=D0=B5=D1=82=D1=81=D1=8F=20=D0=B2=D0=BE=D1=80=D0=BA=D0=B5=D1=80?= =?UTF-8?q?=D0=BE=D0=BC=20=D0=B8=D0=BD=D0=BA=D1=80=D0=B5=D0=BC=D0=B5=D0=BD?= =?UTF-8?q?=D1=82=D0=B0=D0=BB=D1=8C=D0=BD=D0=BE=20+=20=D0=B6=D0=B8=D0=B2?= =?UTF-8?q?=D0=BE=D0=B9=20usage=20=D0=B8=D0=B7=209router=20(t-260929170544?= =?UTF-8?q?-adebea)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Два корня пустого spend за час: 1) usage-добыча в воркере смотрела m["message"]["usage"] — у strands usage живёт в message["metadata"]["usage"] (event_loop.py), поэтому usage всегда был пуст (тот же баг, что уже чинили в runtime._extract_usage). 2) BudgetSpent писал scheduler ПОСЛЕ успешного возврата воркера — таймаут убивал воркер раньше записи, spend терял весь тик. 3) Модель: strands_harness строит OpenAIResponsesModel, у 9router /responses — шим без usage (проверено: response.completed приходит без usage; non-stream возвращает chat-completion-объект). Перевёл на OpenAIModel (chat/completions, include_usage) — usage приходит и с инструментами, и в reasoning-режиме. Фикс: - воркер пишет BudgetSpent ИНКРЕМЕНТАЛЬНО после каждого модельного вызова (hook на MessageAddedEvent, assistant с metadata.usage, дедуп по объекту) — таймаут теряет максимум хвост текущего вызова, не весь тик; - _extract_usage: корректная структура + суммирование по шагам (model_calls); - scheduler дублирует запись только если воркер не смог разобрать usage; - события помечены source: worker|runner и trace_id тика. Проверено: дельты воркера == итоговый usage тика (39-шаговый тик сходится); убитый по таймауту тик сохраняет все завершённые шаги; spend виден в read model/dashboard. --- harness/_llm_worker.py | 124 ++++++++++++++++++++++++++++++++----- harness/isolated_runner.py | 13 +++- harness/runtime.py | 39 ++++++++++-- harness/scheduler.py | 37 ++++++++++- 4 files changed, 189 insertions(+), 24 deletions(-) diff --git a/harness/_llm_worker.py b/harness/_llm_worker.py index 882d695..85d0d12 100644 --- a/harness/_llm_worker.py +++ b/harness/_llm_worker.py @@ -12,6 +12,26 @@ import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +def build_level_model_from_payload(payload: dict): + """Модель для воркера — та же сборка, что и в runtime.build_level_model: + 9router понимает chat/completions (usage там же), Responses API — шим без usage.""" + from strands.models.openai import OpenAIModel + import os + params = {} + effort = payload.get("effort") or "auto" + if effort and effort != "auto": + params["reasoning_effort"] = effort + model_id = payload.get("model") or "auto" + if model_id == "none": + model_id = "auto" + return OpenAIModel( + client_args={"api_key": os.environ.get("OPENAI_API_KEY", ""), + "base_url": os.environ.get("OPENAI_BASE_URL", "http://127.0.0.1:20128/v1")}, + model_id=model_id, + **({"params": params} if params else {}), + ) + + def _emit(payload, obj): rp = payload.get("result_path") if rp: @@ -22,6 +42,90 @@ def _emit(payload, obj): pass +def _extract_usage(agent) -> dict: + """Usage из истории сообщений агента. + + Strands прикрепляет usage к каждой assistant-сообщении в + message["metadata"]["usage"] (event_loop.py) — по одному словарю + на модельный вызов (шаг тика). Накапливаем по шагам: тик может + состоять из многих вызовов, бюджет интересует суммарный. + + Старый разбор m["message"]["usage"] не находил ничего (структура + другая) — usage всегда возвращался пустым, spend не рос. + """ + usage = {"prompt_tokens": 0, "completion_tokens": 0, "model_calls": 0} + try: + msgs = getattr(agent, "messages", None) or [] + for m in msgs: + u = ((m or {}).get("metadata") or {}).get("usage") or {} + p = u.get("inputTokens") or 0 + c = u.get("outputTokens") or 0 + if isinstance(p, (int, float)): + usage["prompt_tokens"] += p + if isinstance(c, (int, float)): + usage["completion_tokens"] += c + if p or c: + usage["model_calls"] += 1 + except Exception: + pass + if not usage["model_calls"]: + return {} + return usage + + +def _emit_budget_incremental(level: str, trace_id: str, usage_delta: dict): + """Записать BudgetSpent за один модельный вызов. + + Воркер — единственный автор spend-событий, но scheduler может убить + его по таймауту в любой момент (SIGKILL). Поэтому usage пишем + ИНКРЕМЕНТАЛЬНО: каждый завершённый модельный вызов — своё событие. + Хвост недописанного вызова теряется, но весь тик — никогда. + """ + if not usage_delta: + return + try: + from harness import es + es.append_event("harness", "budget", "BudgetSpent", { + "level": level, "trace_id": trace_id, "source": "worker", + **usage_delta, + }) + except Exception: + pass # трекинг не должен валить тик + + +def _wire_budget_hooks(agent, level: str, trace_id: str): + """После каждого модельного вызова — usage-дельта в стор. + + Strands вешает usage на assistant-сообщение в + message["metadata"]["usage"] (одна запись на модельный вызов). + MessageAddedEvent стреляет на добавление КАЖДОЙ записи истории, + нас интересуют только assistant с usage. + """ + from strands.hooks import MessageAddedEvent + + seen = set() + + def on_message_added(event): + msg = getattr(event, "message", None) or {} + if msg.get("role") != "assistant": + return + usage = ((msg.get("metadata") or {}).get("usage")) or {} + p = usage.get("inputTokens") or 0 + c = usage.get("outputTokens") or 0 + if not (p or c): + return + key = id(msg) + if key in seen: # MessageUpdatedEvent может переигрывать — дедуп по объекту + return + seen.add(key) + _emit_budget_incremental(level, trace_id, { + "prompt_tokens": p, "completion_tokens": c, + }) + + agent.add_hook(on_message_added, MessageAddedEvent) + return agent + + def main(): payload = json.loads(sys.stdin.read()) @@ -32,10 +136,11 @@ def main(): from strands_harness import create_harness trace_id = payload.get("trace_prefix") or "tick" + level = payload.get("level") or "?" try: agent = create_harness( - model=f"openai/{payload['model']}", - effort=payload.get("effort", "auto"), + model=build_level_model_from_payload(payload), + effort="auto", # effort уже в параметрах модели instructions=payload.get("instructions", ""), session=None, builtin_tools=("read", "write", "edit", "shell") if payload.get("can_shell") else [], @@ -43,20 +148,9 @@ def main(): skills=False, ) agent = _wire_trace_hooks(agent, es, trace_id) + agent = _wire_budget_hooks(agent, level, trace_id) result = agent(payload.get("prompt", "")) - usage = {} - try: - msgs = getattr(agent, "messages", None) or [] - for m in reversed(msgs): - u = (m or {}).get("message", {}).get("usage") - if u: - usage = { - "prompt_tokens": u.get("inputTokens") or u.get("prompt_tokens"), - "completion_tokens": u.get("outputTokens") or u.get("completion_tokens"), - } - break - except Exception: - pass + usage = _extract_usage(agent) _emit(payload, {"ok": True, "text": str(result), "usage": usage, "trace_id": trace_id}) except Exception as e: import traceback diff --git a/harness/isolated_runner.py b/harness/isolated_runner.py index 3784b05..aec571f 100644 --- a/harness/isolated_runner.py +++ b/harness/isolated_runner.py @@ -25,6 +25,10 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim """ import tempfile result_path = tempfile.mktemp(prefix="simbiot_tick_", suffix=".json", dir="/tmp") + # BudgetSpent пишет сам воркер — ИНКРЕМЕНТАЛЬНО, после каждого модельного + # вызова (см. _llm_worker._wire_budget_hooks). Раньше событие писалось + # здесь, ПОСЛЕ возврата воркера: таймаут убивал процесс до записи, и + # spend молча терял весь тик (task t-260929170544-adebea). payload = { "model": level_cfg.model, "effort": level_cfg.effort, @@ -35,6 +39,7 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim "can_shell": level_cfg.can_shell, "ctx_tokens": level_cfg.ctx_tokens, "result_path": result_path, + "level": level_cfg.name, # для записи BudgetSpent самим воркером } env = dict(os.environ) env["SIMBIOT_DB"] = env.get("SIMBIOT_DB", os.path.join(_HARNESS_DIR, "harness", "harness.db")) @@ -52,8 +57,12 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim cwd=_HARNESS_DIR, env=env, ) - finally: - pass + except subprocess.TimeoutExpired as e: + # воркер убит по таймауту. Он пишет BudgetSpent сам после каждого + # модельного вызова, так что spend этого тика уже в сторе — + # теряется максимум хвост вызова, шедшего на момент убийства. + # Снаружи дописывать нечего: история воркера недоступна после SIGKILL. + raise out = None if os.path.exists(result_path): try: diff --git a/harness/runtime.py b/harness/runtime.py index 63b0926..d4ebb59 100644 --- a/harness/runtime.py +++ b/harness/runtime.py @@ -23,13 +23,42 @@ def _router_key(): return env +def _router_base_url(): + return os.environ.get("ROUTER_BASE_URL", "http://127.0.0.1:20128/v1") + + +def build_level_model(level_cfg): + """Модель уровня для 9router: CHAT COMPLETIONS, не Responses API. + + strands_harness для провайдера openai строит OpenAIResponsesModel, + но у 9router /responses — шим: usage в response.completed отсутствует + (проверено: usage:{} и в стриме, и в non-stream). Токены видны только + в /chat/completions (OpenAIModel, stream_options include_usage) — + а без них BudgetSpent бесполезен. Проверено: OpenAIModel + 9router + отдаёт usage и с инструментами, и в reasoning-режиме. + + Сессии выключены, поэтому server-side state Responses API не нужен; + chat/completions stateless и полностью покрывает потребности тика. + """ + from strands.models.openai import OpenAIModel + params = {} + if level_cfg.effort and level_cfg.effort != "auto": + params["reasoning_effort"] = level_cfg.effort + return OpenAIModel( + client_args={"api_key": os.environ.get("OPENAI_API_KEY", ""), + "base_url": os.environ.get("OPENAI_BASE_URL", "http://127.0.0.1:20128/v1")}, + model_id=level_cfg.model if level_cfg.model != "none" else "auto", + **({"params": params} if params else {}), + ) + + def setup_provider(): """9router как openai-совместимый провайдер strands.""" key = _router_key() if not key: raise RuntimeError("ROUTER_KEY не найден (.env)") os.environ["OPENAI_API_KEY"] = key - os.environ["OPENAI_BASE_URL"] = os.environ.get("ROUTER_BASE_URL", "http://127.0.0.1:20128/v1") + os.environ["OPENAI_BASE_URL"] = _router_base_url() def _wire_trace_hooks(agent, es, trace_id: str): @@ -79,8 +108,8 @@ def build_agent(level_cfg, session_id=None, extra_instructions=""): instructions = level_cfg.mandate + ("\n\n" + extra_instructions if extra_instructions else "") agent = create_harness( - model=f"openai/{level_cfg.model}", - effort=level_cfg.effort, + model=build_level_model(level_cfg), + effort="auto", # effort уже в параметрах модели (build_level_model) instructions=instructions, session={"id": session_id} if session_id else None, builtin_tools=("read", "write", "edit", "shell") if level_cfg.can_shell else [], @@ -98,8 +127,8 @@ def run_cycle(level_cfg, es, prompt: str, session_id=None, trace_prefix=None): trace_id = trace_prefix or uuid.uuid4().hex[:12] instructions = level_cfg.mandate agent = create_harness( - model=f"openai/{level_cfg.model}", - effort=level_cfg.effort, + model=build_level_model(level_cfg), + effort="auto", # effort уже в параметрах модели (build_level_model) instructions=instructions, session={"id": session_id} if session_id else None, builtin_tools=("read", "write", "edit", "shell") if level_cfg.can_shell else [], diff --git a/harness/scheduler.py b/harness/scheduler.py index 9445f93..c6c6ff1 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -26,6 +26,8 @@ ACTIVE = ["sec", "min", "hour", "day"] # состояние исполнителей уровня: level → {"busy": bool, "task": id|None} _runners = {} +# когда уровень последний раз слал TG-уведомление о занятости (анти-спам) +_NOTIFY_TS = {} def _now(): @@ -140,6 +142,25 @@ def build_context(level_cfg, model: dict, task=None): return "\n\n".join(parts) +def _notify_busy(level_cfg, task): + """TG-visibility: уровни сами сообщают о занятости. + + Анти-спам: по одному уведомлению на уровень не чаще раза в 15 минут — + повторные взятия задач внутри часа не заспамят чат. + """ + global _NOTIFY_TS + now = time.time() + if now - _NOTIFY_TS.get(level_cfg.name, 0) < 900: # 15 минут + return + _NOTIFY_TS[level_cfg.name] = now + from . import tg_inbox + tg_inbox.notify_chat( + f"🔧 Уровень {level_cfg.name} взял задачу [{task.get('size', '?')}] " + f"«{task.get('title', '')[:70]}»\n" + f"Срез лестницы по запросу: /лестница" + ) + + def run_level_tick(level_cfg, seq: int, model: dict): """Один тик уровня с LLM: дайджест + (опц.) задача. Возвращает текст. @@ -155,6 +176,12 @@ def run_level_tick(level_cfg, seq: int, model: dict): es.append_event("task", task_id, "TaskClaimed", { "level": level_cfg.name, "session_id": f"s-{level_cfg.name}-{seq}", }) + # TG-visibility: взяли задачу — сообщаем в чат (анти-спам: не чаще 15м на уровень) + try: + from . import tg_inbox + _notify_busy(level_cfg, task) + except Exception: + pass # уведомление — best effort, не валить тик prompt = build_context(level_cfg, model, task) text, usage, trace_id = isolated_runner.run_llm_tick( level_cfg, prompt, @@ -174,9 +201,15 @@ def run_level_tick(level_cfg, seq: int, model: dict): "level": level_cfg.name, "period": period.isoformat(), "summary_ref": trace_id, "text": (text or "")[:2000], }) - if usage: + # BudgetSpent воркер уже записал сам — инкрементально, после каждого + # модельного вызова (устраняет потерю spend при таймауте воркера). + # Если usage в результате пуст (воркер не смог разобрать историю — + # старый воркер / нетекстовая структура), дублируем итог, чтобы тик + # не терялся в spend вовсе: дедуп по trace_id в read model. + if usage and not usage.get("model_calls"): es.append_event("harness", "budget", "BudgetSpent", { - "level": level_cfg.name, **{k: v for k, v in usage.items() if k != "error"}, + "level": level_cfg.name, "trace_id": trace_id, "source": "runner", + **{k: v for k, v in usage.items() if k != "error"}, }) if task: es.append_event("task", task_id, "TaskCompleted", {