diff --git a/interfaces/admin/routes.py b/interfaces/admin/routes.py index 53aa0e60..f1d955f2 100644 --- a/interfaces/admin/routes.py +++ b/interfaces/admin/routes.py @@ -3860,6 +3860,12 @@ def build_admin_router() -> APIRouter: status="ok", detail={"synced_tools": len(norm), "compat_mode": "bailian_webparser"}, ) + try: + from runtime.direct_loop import invalidate_tool_wire_cache + + invalidate_tool_wire_cache(reason=f"mcp_admin_sync:{server_id}") + except Exception: + pass return {"ok": True, "server_id": server_id, "tools": norm, "compat_mode": "bailian_webparser"} rt = mcp_runtime_for_row(row, store=store) try: @@ -3883,6 +3889,12 @@ def build_admin_router() -> APIRouter: ) store.replace_mcp_server_tools(server_id=server_id, tools=norm) store.set_mcp_server_health(server_id=server_id, status="ok", detail={"synced_tools": len(norm)}) + try: + from runtime.direct_loop import invalidate_tool_wire_cache + + invalidate_tool_wire_cache(reason=f"mcp_admin_sync:{server_id}") + except Exception: + pass return {"ok": True, "server_id": server_id, "tools": norm} finally: rt.stop() diff --git a/runtime/application/gateway/ops_short_intent.py b/runtime/application/gateway/ops_short_intent.py index b226d22f..c8957a18 100644 --- a/runtime/application/gateway/ops_short_intent.py +++ b/runtime/application/gateway/ops_short_intent.py @@ -8,42 +8,50 @@ from typing import Any _BOT_MENTION_RE = re.compile(r"@\S+") # intent -> (en hint, zh hint) +# Prefer report path strongly; keep CLI/inventory available with soft budget (not hard-hidden). _HINTS: dict[str, tuple[str, str]] = { "excel_export": ( - "[Ops short-intent: Excel export. Call ume_alarm_xlsx_report(..., deliverable=true) now. " - "write_xlsx / inventory / CLI / run_command are hidden this turn — do not build xlsx via shell.]", - "[短指令:导出 Excel。立即 ume_alarm_xlsx_report(..., deliverable=true)。" - "本轮已隐藏 write_xlsx/清单/CLI/run_command,禁止用 shell 搓表。]", + "[Ops short-intent: Excel export. Prefer ume_alarm_xlsx_report(..., deliverable=true) now. " + "write_xlsx / run_command are hidden this turn. " + "CLI/inventory stay available with a soft budget for device checks — prefer one hetero targets batch, not single-NE loops.]", + "[短指令:导出 Excel。优先 ume_alarm_xlsx_report(..., deliverable=true)。" + "本轮已隐藏 write_xlsx/run_command。" + "CLI/清单仍可用(soft budget):核实设备优先一次 hetero targets batch,勿单台循环。]", ), "license": ( - "[Ops short-intent: license/capacity. Call ume_alarm_xlsx_report(mode=list, keyword=license, deliverable=true) " - "or aggregateUmeAlarms/queryUmeAlarmsRaw; write_xlsx/CLI/inventory are hidden this turn.]", - "[短指令:License/容量。立即 ume_alarm_xlsx_report(mode=list, keyword=license, deliverable=true) " - "或 aggregate/queryUmeAlarmsRaw;本轮已隐藏 write_xlsx/CLI/清单。]", + "[Ops short-intent: license/capacity. Prefer ume_alarm_xlsx_report(mode=list, keyword=license, deliverable=true) " + "or aggregateUmeAlarms/queryUmeAlarmsRaw. write_xlsx/run_command hidden; " + "CLI available with soft budget for confirmation (one targets batch preferred).]", + "[短指令:License/容量。优先 ume_alarm_xlsx_report(mode=list, keyword=license, deliverable=true) " + "或 aggregate/queryUmeAlarmsRaw;write_xlsx/run_command 已隐藏;" + "CLI 可用(soft budget),确认阶段优先一次 targets batch。]", ), "congestion": ( - "[Ops short-intent: bandwidth congestion. Call ume_alarm_xlsx_report(mode=list, deliverable=true) " - "or aggregateUmeAlarms/queryUmeAlarmsRaw; write_xlsx/CLI/inventory/sql are hidden this turn.]", - "[短指令:带宽拥塞。立即 ume_alarm_xlsx_report(mode=list, deliverable=true) 或 aggregate/query;" - "本轮已隐藏 write_xlsx/CLI/清单/sql。]", + "[Ops short-intent: bandwidth congestion. Prefer ume_alarm_xlsx_report(mode=list, deliverable=true) " + "or aggregateUmeAlarms/queryUmeAlarmsRaw. write_xlsx/run_command hidden; " + "CLI available with soft budget (one targets batch preferred).]", + "[短指令:带宽拥塞。优先 ume_alarm_xlsx_report(mode=list, deliverable=true) 或 aggregate/query;" + "write_xlsx/run_command 已隐藏;CLI 可用(soft budget),优先一次 targets batch。]", ), "fiber_cut": ( - "[Ops short-intent: fiber/LOS. Call ume_alarm_xlsx_report(mode=fiber_cut, deliverable=true) now. " - "write_xlsx / inventory / CLI are hidden this turn — do not try listCliTargets/execManagedNe/write_xlsx.]", - "[短指令:断纤/LOS。立即 ume_alarm_xlsx_report(mode=fiber_cut, deliverable=true)。" - "本轮已隐藏 write_xlsx/清单/CLI,勿调用 listCliTargets/execManagedNe/write_xlsx。]", + "[Ops short-intent: fiber/LOS. Prefer ume_alarm_xlsx_report(mode=fiber_cut, deliverable=true) now. " + "write_xlsx/run_command hidden. CLI/inventory available with soft budget — " + "use at most one hetero targets batch for device confirmation, not listCliTargets→single-NE loops.]", + "[短指令:断纤/LOS。优先 ume_alarm_xlsx_report(mode=fiber_cut, deliverable=true)。" + "write_xlsx/run_command 已隐藏。CLI/清单可用(soft budget):" + "设备核实最多一次 hetero targets batch,勿 listCliTargets→单台循环。]", ), "offline": ( - "[Ops short-intent: offline NE. Call ume_alarm_xlsx_report(mode=offline, deliverable=true) now. " - "write_xlsx / inventory / CLI are hidden this turn.]", - "[短指令:离线网元。立即 ume_alarm_xlsx_report(mode=offline, deliverable=true)。" - "本轮已隐藏 write_xlsx/清单/CLI。]", + "[Ops short-intent: offline NE. Prefer ume_alarm_xlsx_report(mode=offline, deliverable=true) now. " + "write_xlsx/run_command hidden; CLI/inventory with soft budget (one targets batch preferred).]", + "[短指令:离线网元。优先 ume_alarm_xlsx_report(mode=offline, deliverable=true)。" + "write_xlsx/run_command 已隐藏;CLI/清单 soft budget(优先一次 targets batch)。]", ), "alarm_tally": ( - "[Ops short-intent: alarm tally/top. Call ume_alarm_xlsx_report(mode=aggregate_by_host, deliverable=true) " - "or aggregateUmeAlarms; write_xlsx / inventory / CLI are hidden this turn.]", - "[短指令:告警统计/Top。立即 ume_alarm_xlsx_report(mode=aggregate_by_host, deliverable=true) " - "或 aggregateUmeAlarms;本轮已隐藏 write_xlsx/清单/CLI。]", + "[Ops short-intent: alarm tally/top. Prefer ume_alarm_xlsx_report(mode=aggregate_by_host, deliverable=true) " + "or aggregateUmeAlarms. write_xlsx/run_command hidden; CLI soft budget for spot checks.]", + "[短指令:告警统计/Top。优先 ume_alarm_xlsx_report(mode=aggregate_by_host, deliverable=true) " + "或 aggregateUmeAlarms;write_xlsx/run_command 已隐藏;CLI soft budget 仅作抽检。]", ), "continue": ( "[Ops short-intent: continue/confirm. Resume the unfinished prior task immediately; " @@ -52,29 +60,21 @@ _HINTS: dict[str, tuple[str, str]] = { ), } -# Report-style short intents: hide inventory/CLI loops that dominate WA tool spam. +# Report-style short intents: soft-nudge toward alarm/xlsx; hide only shell/xlsx DIY. _REPORT_TOOL_FILTER_INTENTS = frozenset( {"fiber_cut", "offline", "alarm_tally", "excel_export", "license", "congestion"} ) -# Match bare tool names and mcp__netx__* / legacy netx_* aliases. +# Soft CLI budgets during report short-intents (prefer report; allow limited device checks). +_SHORT_INTENT_CLI_SOFT_SINGLE = 2 +_SHORT_INTENT_CLI_SOFT_FAIL = 2 +_SHORT_INTENT_CLI_SOFT_BATCH = 1 + +# Only hard-hide DIY spreadsheet / shell paths — keep inventory + execManagedNe visible. _SUPPRESSED_TOOL_NAMES = frozenset( { - "listclitargets", - "listmanagedne", - "getmanagedne", - "execmanagedne", - "queryumeneinventory", - "getumene", - "findtopologypaths", - "sqlqueryume", "run_command", "write_xlsx", - "netx_list_managed_ne", - "netx_get_managed_ne", - "netx_exec_managed_ne", - "netx_sql_query_ume", - "netx_list_cli_targets", } ) @@ -93,6 +93,17 @@ def ops_short_intent_should_filter_tools(intent: str | None) -> bool: return str(intent or "").strip() in _REPORT_TOOL_FILTER_INTENTS +def ops_short_intent_cli_soft_budgets(intent: str | None) -> dict[str, int] | None: + """Tighter execManagedNe budgets for report short-intents; None = use global defaults.""" + if not ops_short_intent_should_filter_tools(intent): + return None + return { + "single": int(_SHORT_INTENT_CLI_SOFT_SINGLE), + "fail": int(_SHORT_INTENT_CLI_SOFT_FAIL), + "batch": int(_SHORT_INTENT_CLI_SOFT_BATCH), + } + + def is_ops_short_intent_suppressed_tool(tool_name: str, *, intent: str | None) -> bool: if not ops_short_intent_should_filter_tools(intent): return False @@ -101,7 +112,7 @@ def is_ops_short_intent_suppressed_tool(tool_name: str, *, intent: str | None) - def filter_tool_specs_for_ops_short_intent(tools: list[Any], *, intent: str | None) -> list[Any]: - """Drop inventory/CLI tools for report-style short intents (keep alarm/xlsx path).""" + """Drop DIY xlsx/shell tools for report-style short intents (keep alarm path + CLI).""" if not ops_short_intent_should_filter_tools(intent): return list(tools or []) out: list[Any] = [] @@ -188,5 +199,6 @@ __all__ = [ "is_ops_short_intent_suppressed_tool", "maybe_ops_short_intent_system_hint", "normalize_ops_user_text", + "ops_short_intent_cli_soft_budgets", "ops_short_intent_should_filter_tools", ] diff --git a/runtime/chat/exec_managed_ne_guard.py b/runtime/chat/exec_managed_ne_guard.py index 5b8c4258..fac68dc8 100644 --- a/runtime/chat/exec_managed_ne_guard.py +++ b/runtime/chat/exec_managed_ne_guard.py @@ -128,6 +128,8 @@ def budget_block_payload( single_budget: int, fail_used: int = 0, fail_budget: int = 0, + batch_used: int = 0, + batch_budget: int = 0, ) -> dict[str, Any]: en = str(lang or "").strip().lower().startswith("en") if reason == "fail_budget": @@ -136,14 +138,24 @@ def budget_block_payload( f"execManagedNe already failed {fail_used}/{fail_budget} times this turn. " "Stop one-NE loops; use ONE execManagedNe batch: " "ne_ids|ume_ne_ids + shared commands, or targets=[{ume_ne_id, commands},…] when commands differ — " - "or summarize reachable failures; do not keep probing." + "or summarize reachable failures / switch to ume_alarm_xlsx_report; do not keep probing." if en else f"本轮 execManagedNe 已失败 {fail_used}/{fail_budget} 次。" "停止单台循环;改用一次 batch:" "同命令用 ne_ids/ume_ne_ids,每台命令不同用 targets=[{ume_ne_id, commands},…];" - "或汇总可达性失败,勿继续盲探。" + "或汇总可达性失败 / 改走 ume_alarm_xlsx_report,勿继续盲探。" ) err = "cli_fail_budget_exceeded" + elif reason == "batch_budget": + code = "cli_batch_budget_exceeded" + hint = ( + f"execManagedNe batch soft budget exhausted ({batch_used}/{batch_budget} this turn). " + "Prefer ume_alarm_xlsx_report / alarm aggregate for the answer; do not start another CLI batch." + if en + else f"本轮 execManagedNe batch soft budget 已用尽({batch_used}/{batch_budget})。" + "优先 ume_alarm_xlsx_report / 告警聚合收口,勿再开新 CLI batch。" + ) + err = "cli_batch_budget_exceeded" else: code = "cli_call_budget_exceeded" hint = ( @@ -181,6 +193,8 @@ def budget_block_payload( "single_budget": int(single_budget), "fail_used": int(fail_used), "fail_budget": int(fail_budget), + "batch_used": int(batch_used), + "batch_budget": int(batch_budget), } diff --git a/runtime/chat/history_tool_result_compact.py b/runtime/chat/history_tool_result_compact.py index 02179314..19d7b368 100644 --- a/runtime/chat/history_tool_result_compact.py +++ b/runtime/chat/history_tool_result_compact.py @@ -46,6 +46,7 @@ def _guard_tool_result_text_for_history( cap_chars: int, image_cap_chars: int, video_cap_chars: int, + session_id: str = "", ) -> tuple[str, bool]: """Return (new_raw, changed) following the same strategy as context replay guard.""" text = str(raw or "") @@ -110,6 +111,33 @@ def _guard_tool_result_text_for_history( if len(text) <= int(cap_chars): return text, False + existing_ref = "" + if isinstance(obj, dict): + existing_ref = str(obj.get("result_ref") or "").strip() + if not existing_ref: + try: + from runtime.chat.tool_result_store import save_tool_result_blob + + sid = str(session_id or getattr(store, "_history_compact_session_id", "") or "").strip() or "_history" + parsed_full = obj if isinstance(obj, dict) else None + if parsed_full is None: + try: + p = json.loads(text) + parsed_full = p if isinstance(p, dict) else {"ok": None, "raw": text} + except Exception: + parsed_full = {"ok": None, "raw": text[: min(len(text), 200_000)]} + existing_ref = str( + save_tool_result_blob( + session_id=sid, + tool_call_id=f"history-{hash(text) & 0xFFFFFFFF:08x}", + result=parsed_full, + force=True, + ) + or "" + ) + except Exception: + existing_ref = "" + preview = text[: max(1, min(4000, int(cap_chars) - 400))] + "\n..." guarded_obj = { "ok": bool(ok) if ok is not None else None, @@ -121,13 +149,15 @@ def _guard_tool_result_text_for_history( "preview": preview, "hint": ( "Tool output was too large for safe context replay; it was truncated for history storage. " - "Use narrower queries (e.g., smaller glob/max_results) or adjust AIA_TOOL_LLM_MESSAGE_MAX_CHARS. / " - "工具输出过大,已压缩写回历史;请缩小范围或配置 AIA_TOOL_LLM_MESSAGE_MAX_CHARS。" + "Use fetch_tool_result(result_ref=...) when present, or narrower queries. / " + "工具输出过大,已压缩写回历史;有 result_ref 时用 fetch_tool_result 取全文。" ), } + if existing_ref: + guarded_obj["result_ref"] = existing_ref + guarded_obj["fetch_tool"] = "fetch_tool_result" return _json_dumps_safe(guarded_obj), True - def compact_tool_results_in_session_history( *, store: Any, @@ -150,6 +180,10 @@ def compact_tool_results_in_session_history( max_seen = 0 # We only need to scan tool messages; fetching ids+content is enough. + try: + store._history_compact_session_id = sid # noqa: SLF001 + except Exception: + pass with store._connect() as conn: # noqa: SLF001 cur = conn.execute( "select id, content from chat_message where session_id=? and role='tool' " @@ -168,6 +202,7 @@ def compact_tool_results_in_session_history( cap_chars=cap, image_cap_chars=image_cap, video_cap_chars=video_cap, + session_id=sid, ) # Defensive fallback: if content is still over cap but guard didn't report change, # force a minimal guard so polluted history can always be compacted. diff --git a/runtime/chat/tool_result_store.py b/runtime/chat/tool_result_store.py new file mode 100644 index 00000000..a85476f7 --- /dev/null +++ b/runtime/chat/tool_result_store.py @@ -0,0 +1,167 @@ +"""Persist full tool results for later fetch_tool_result while LLM sees compact payloads.""" + +from __future__ import annotations + +import hashlib +import json +import re +import time +from pathlib import Path +from typing import Any + +from svc.config.paths import attachments_dir + +_REF_RE = re.compile(r"^tr:[0-9a-f]{32}$") +_SAVE_MIN_CHARS = 4_000 +_FETCH_DEFAULT_MAX_CHARS = 120_000 + + +def _root() -> Path: + p = (attachments_dir() / "tool_results").resolve() + p.mkdir(parents=True, exist_ok=True) + return p + + +def _session_dir(session_id: str) -> Path: + sid = str(session_id or "").strip() or "_anon" + digest = hashlib.sha256(sid.encode("utf-8", errors="ignore")).hexdigest()[:24] + d = _root() / digest + d.mkdir(parents=True, exist_ok=True) + return d + + +def _json_dumps(obj: Any) -> str: + return json.dumps(obj, ensure_ascii=False, default=str) + + +def _json_size(obj: Any) -> int: + try: + return len(_json_dumps(obj)) + except Exception: + return 0 + + +def make_result_ref(*, session_id: str, tool_call_id: str, payload: Any) -> str: + raw = f"{session_id}|{tool_call_id}|{_json_size(payload)}|{time.time_ns()}" + return "tr:" + hashlib.sha256(raw.encode("utf-8", errors="ignore")).hexdigest()[:32] + + +def is_tool_result_ref(value: str) -> bool: + return bool(_REF_RE.fullmatch(str(value or "").strip().lower())) + + +def save_tool_result_blob( + *, + session_id: str, + tool_call_id: str, + result: Any, + force: bool = False, +) -> str | None: + """Store full tool JSON when large enough (or force=True). Returns result_ref or None.""" + sid = str(session_id or "").strip() + if not sid: + return None + if not isinstance(result, dict): + return None + size = _json_size(result) + if (not force) and size < int(_SAVE_MIN_CHARS): + return None + ref = make_result_ref(session_id=sid, tool_call_id=str(tool_call_id or ""), payload=result) + path = _session_dir(sid) / f"{ref[3:]}.json" + meta = { + "result_ref": ref, + "session_id": sid, + "tool_call_id": str(tool_call_id or ""), + "chars": int(size), + "saved_at_ms": int(time.time() * 1000), + } + path.write_text(_json_dumps({"meta": meta, "result": result}), encoding="utf-8") + return ref + + +def load_tool_result_blob( + result_ref: str, + *, + session_id: str, + max_chars: int | None = None, +) -> dict[str, Any]: + ref = str(result_ref or "").strip().lower() + sid = str(session_id or "").strip() + if not is_tool_result_ref(ref): + return {"ok": False, "error_code": "invalid_result_ref", "error": "invalid_result_ref"} + if not sid: + return {"ok": False, "error_code": "session_required", "error": "session_required"} + path = _session_dir(sid) / f"{ref[3:]}.json" + if not path.is_file(): + # Fallback: scan root for orphaned refs (session hash mismatch / migrate). + found = None + for child in _root().glob(f"*/{ref[3:]}.json"): + found = child + break + if found is None: + return {"ok": False, "error_code": "result_ref_not_found", "error": "result_ref_not_found", "result_ref": ref} + path = found + try: + data = json.loads(path.read_text(encoding="utf-8")) + except Exception as exc: + return {"ok": False, "error_code": "result_ref_read_failed", "error": f"{type(exc).__name__}: {exc}"} + meta = data.get("meta") if isinstance(data, dict) else None + if isinstance(meta, dict): + owner = str(meta.get("session_id") or "").strip() + if owner and owner != sid: + return {"ok": False, "error_code": "result_ref_session_mismatch", "error": "result_ref_session_mismatch"} + result = data.get("result") if isinstance(data, dict) else None + if not isinstance(result, dict): + return {"ok": False, "error_code": "result_ref_invalid_payload", "error": "result_ref_invalid_payload"} + cap = int(_FETCH_DEFAULT_MAX_CHARS if max_chars is None else max_chars) + cap = max(4_000, min(cap, 500_000)) + body = _json_dumps(result) + if len(body) <= cap: + out = dict(result) + out["result_ref"] = ref + out["_fetched_full"] = True + return {"ok": True, "result_ref": ref, "result": out, "chars": len(body), "truncated": False} + from runtime.chat.tool_runtime import truncate_tool_result_for_llm_messages + + slim = truncate_tool_result_for_llm_messages(result, max_chars=cap) + if isinstance(slim, dict): + slim = dict(slim) + slim["result_ref"] = ref + slim["_fetched_truncated"] = True + slim["hint"] = ( + str(slim.get("hint") or "") + + " Full blob still on disk; narrow the tool query or raise max_chars on fetch_tool_result." + ).strip() + return { + "ok": True, + "result_ref": ref, + "result": slim, + "chars": len(body), + "truncated": True, + "fetch_cap_chars": cap, + } + + +def attach_result_ref(payload: dict[str, Any], *, result_ref: str | None) -> dict[str, Any]: + if not result_ref or not isinstance(payload, dict): + return payload + out = dict(payload) + out["result_ref"] = str(result_ref) + if out.get("_truncated_for_llm") or out.get("_tool_result_guarded") or out.get("_history_compacted"): + out["fetch_tool"] = "fetch_tool_result" + hint = str(out.get("hint") or "").strip() + extra = ( + f"Full tool output stored as result_ref={result_ref}. " + f"Call fetch_tool_result(result_ref=\"{result_ref}\") when you need details." + ) + out["hint"] = f"{hint} {extra}".strip() if hint else extra + return out + + +__all__ = [ + "attach_result_ref", + "is_tool_result_ref", + "load_tool_result_blob", + "make_result_ref", + "save_tool_result_blob", +] diff --git a/runtime/chat/tool_runtime.py b/runtime/chat/tool_runtime.py index a638755f..8bef02b2 100644 --- a/runtime/chat/tool_runtime.py +++ b/runtime/chat/tool_runtime.py @@ -1095,15 +1095,27 @@ class ToolExecutor: is_exec_managed_ne_tool, load_turn_exec_managed_ne_stats, ) + from runtime.application.gateway.ops_short_intent import ops_short_intent_cli_soft_budgets - prior_single_exec, _prior_batch_exec, prior_exec_fails = load_turn_exec_managed_ne_stats( + prior_single_exec, prior_batch_exec, prior_exec_fails = load_turn_exec_managed_ne_stats( ctx.store, session_id=ctx.session_id, turn_uuid=str(ctx.turn_uuid or ""), ) single_exec_budget = exec_managed_ne_single_budget() fail_exec_budget = exec_managed_ne_fail_budget() + batch_exec_budget = 0 # 0 = unlimited (global default) + try: + md_budget = ctx.inbound_metadata if isinstance(ctx.inbound_metadata, dict) else {} + soft = ops_short_intent_cli_soft_budgets(str(md_budget.get("ops_short_intent") or "").strip() or None) + if soft: + single_exec_budget = int(soft.get("single") or single_exec_budget) + fail_exec_budget = int(soft.get("fail") or fail_exec_budget) + batch_exec_budget = int(soft.get("batch") or 0) + except Exception: + pass local_single_exec = 0 + local_batch_exec = 0 local_exec_fails = 0 results_by_id: dict[str, tuple[dict[str, Any], int]] = {} @@ -1237,7 +1249,31 @@ class ToolExecutor: if is_exec_managed_ne_tool(tool_name): batchish = is_batch_exec_args(dict(tc.arguments or {})) single_used = int(prior_single_exec) + int(local_single_exec) + batch_used = int(prior_batch_exec) + int(local_batch_exec) fail_used = int(prior_exec_fails) + int(local_exec_fails) + if batchish and int(batch_exec_budget) > 0 and batch_used >= int(batch_exec_budget): + results_by_id[tc.id] = ( + budget_block_payload( + reason="batch_budget", + lang=str(ctx.lang or "en"), + single_used=single_used, + single_budget=single_exec_budget, + fail_used=fail_used, + fail_budget=fail_exec_budget, + batch_used=batch_used, + batch_budget=batch_exec_budget, + ), + 0, + ) + _trace( + "cli_batch_budget_exceeded", + { + "tool_name": tc.name, + "batch_used": batch_used, + "batch_budget": batch_exec_budget, + }, + ) + continue if (not batchish) and fail_used >= int(fail_exec_budget): results_by_id[tc.id] = ( budget_block_payload( @@ -1247,6 +1283,8 @@ class ToolExecutor: single_budget=single_exec_budget, fail_used=fail_used, fail_budget=fail_exec_budget, + batch_used=batch_used, + batch_budget=batch_exec_budget, ), 0, ) @@ -1264,6 +1302,8 @@ class ToolExecutor: single_budget=single_exec_budget, fail_used=fail_used, fail_budget=fail_exec_budget, + batch_used=batch_used, + batch_budget=batch_exec_budget, ), 0, ) @@ -1330,8 +1370,11 @@ class ToolExecutor: ) continue first_tool_call_id_by_signature[sig] = str(tc.id or "") - if is_exec_managed_ne_tool(tool_name) and not is_batch_exec_args(dict(tc.arguments or {})): - local_single_exec += 1 + if is_exec_managed_ne_tool(tool_name): + if is_batch_exec_args(dict(tc.arguments or {})): + local_batch_exec += 1 + else: + local_single_exec += 1 runnable_tool_uses.append(tc) for batch in partition_tool_use_batches(runnable_tool_uses, ctx.tools): @@ -1399,6 +1442,7 @@ class ToolExecutor: if str(result.get("error_code") or "") not in { "cli_call_budget_exceeded", "cli_fail_budget_exceeded", + "cli_batch_budget_exceeded", "identical_retry_blocked", "retry_forbidden_blocked", "tool_loop_guard", @@ -1427,11 +1471,21 @@ class ToolExecutor: duration_ms=duration_ms, ) tool_log_write_ms = int((time.perf_counter() - t_db1) * 1000) - # Keep full payload during the active turn. History compaction is deferred - # until the turn finishes, so current-round model context remains lossless. + # Persist full blob when large; model context gets compact + result_ref. t_trunc = time.perf_counter() observed_rows_this_call = int(_estimate_observed_rows(result)) result_for_llm = dict(persisted_result or {}) + result_ref: str | None = None + try: + from runtime.chat.tool_result_store import attach_result_ref, save_tool_result_blob + + result_ref = save_tool_result_blob( + session_id=str(ctx.session_id or ""), + tool_call_id=str(tc.id or ""), + result=persisted_result if isinstance(persisted_result, dict) else result_for_llm, + ) + except Exception: + result_ref = None if tc.name in _SQL_REPLAY_COMPACT_TOOL_NAMES: current = int(local_turn_tool_name_counts.get(tc.name, 0)) current_rows = int(local_turn_tool_observed_rows.get(tc.name, 0)) @@ -1444,6 +1498,27 @@ class ToolExecutor: result_for_llm["_tool_observed_rows_this_call"] = int(observed_rows_this_call) result_for_llm["_tool_observed_rows_cumulative_in_turn"] = int(cumulative_rows) result_for_llm["audit_note"] = "Result compacted for history replay safety." + # Compact oversized payloads for the model wire; full body remains in blob/tool_log. + try: + capped = truncate_tool_result_for_llm_messages(result_for_llm) + if isinstance(capped, dict) and capped.get("_truncated_for_llm"): + result_for_llm = capped + if not result_ref: + from runtime.chat.tool_result_store import save_tool_result_blob + + result_ref = save_tool_result_blob( + session_id=str(ctx.session_id or ""), + tool_call_id=str(tc.id or ""), + result=persisted_result if isinstance(persisted_result, dict) else dict(result_for_llm), + force=True, + ) + if result_ref: + from runtime.chat.tool_result_store import attach_result_ref + + result_for_llm = attach_result_ref(result_for_llm, result_ref=result_ref) + except Exception: + if result_ref and isinstance(result_for_llm, dict): + result_for_llm["result_ref"] = result_ref trunc_ms = int((time.perf_counter() - t_trunc) * 1000) tool_content = self._json_dumps_safe(result_for_llm) t_db2 = time.perf_counter() @@ -1469,6 +1544,7 @@ class ToolExecutor: isinstance(result, dict) and _result_is_retry_forbidden(result) ), **({"exec_ne_mode": exec_ne_mode} if exec_ne_mode else {}), + **({"result_ref": result_ref} if result_ref else {}), }, ) try: diff --git a/runtime/chat/turn_idle_guard.py b/runtime/chat/turn_idle_guard.py index 34c07ab7..6eaefb08 100644 --- a/runtime/chat/turn_idle_guard.py +++ b/runtime/chat/turn_idle_guard.py @@ -100,6 +100,7 @@ class TurnIdleTracker: "tool_loop_guard", "cli_call_budget_exceeded", "cli_fail_budget_exceeded", + "cli_batch_budget_exceeded", }: guard += 1 stats = RoundStats( diff --git a/runtime/direct_loop.py b/runtime/direct_loop.py index caf31f59..41d3507a 100644 --- a/runtime/direct_loop.py +++ b/runtime/direct_loop.py @@ -155,6 +155,41 @@ def _tool_wire_freeze_enabled(store: Any) -> bool: return raw in {"1", "true", "yes", "on"} +def _mcp_tools_fingerprint(store: Any) -> str: + """Stable hash of enabled MCP server tool catalogs (invalidates freeze after sync).""" + try: + import hashlib + + rows = store.list_mcp_servers(enabled_only=True) if store else [] + parts: list[str] = [] + for row in rows or []: + if not isinstance(row, dict): + continue + sid = str(row.get("server_id") or "").strip() + if not sid: + continue + tools = [] + try: + tools = store.list_mcp_server_tools(server_id=sid) or [] + except Exception: + tools = [] + names: list[str] = [] + for t in tools: + if isinstance(t, dict): + n = str(t.get("tool_name") or t.get("name") or "").strip() + if n: + names.append(n) + names.sort() + parts.append(f"{sid}:{','.join(names)}") + parts.sort() + raw = "|".join(parts) + if not raw: + return "0" + return hashlib.sha256(raw.encode("utf-8", errors="ignore")).hexdigest()[:16] + except Exception: + return "x" + + def _tool_wire_settings_signature(store: Any) -> tuple[bool, str]: runtime_enabled = True try: @@ -163,6 +198,7 @@ def _tool_wire_settings_signature(store: Any) -> tuple[bool, str]: runtime_enabled = raw_flag in {"1", "true", "yes", "on"} except Exception: runtime_enabled = True + mcp_fp = _mcp_tools_fingerprint(store) sig = "|".join( [ f"rt={int(bool(runtime_enabled))}", @@ -171,11 +207,21 @@ def _tool_wire_settings_signature(store: Any) -> tuple[bool, str]: f"skill_rt={str(store.get_setting('AIA_SKILL_RUNTIME_ENABLED') or '')}", f"skill_disabled={str(store.get_setting('AIA_SKILL_DISABLED_NAMES') or '')}", f"bind_en={str(store.get_setting('AIA_SKILL_ROLE_BINDING_ENABLED') or '')}", + f"mcp_tools={mcp_fp}", ] ) return runtime_enabled, sig +def invalidate_tool_wire_cache(*, reason: str = "") -> dict[str, Any]: + """Clear frozen tool-wire cache so the next turn rebuilds from current catalogs.""" + global _TOOL_WIRE_FROZEN_SIGNATURE + with _TOOL_WIRE_CACHE_LOCK: + _TOOL_WIRE_FROZEN_SIGNATURE = None + n = len(_TOOL_WIRE_CACHE) + _TOOL_WIRE_CACHE.clear() + return {"ok": True, "cleared_entries": int(n), "reason": str(reason or "")} + def _tool_wire_cache_key( *, store: Any, @@ -416,6 +462,9 @@ def _guard_tool_results_for_llm_context( if len(raw) <= cap: out.append(_tool_message_with_content(m, raw, sid=session_id)) continue + existing_ref = "" + if isinstance(obj, dict): + existing_ref = str(obj.get("result_ref") or "").strip() preview = raw[: max(1, min(4000, cap - 400))] + "\n..." guarded_obj = { "ok": bool(ok) if ok is not None else None, @@ -427,10 +476,29 @@ def _guard_tool_results_for_llm_context( "preview": preview, "hint": ( "Tool output was too large for safe context replay; it was truncated for the model context. " - "Use narrower queries (e.g., smaller glob/max_results) or adjust AIA_TOOL_LLM_MESSAGE_MAX_CHARS. / " - "工具输出过大,已在发给模型的上下文中强制截断;请缩小范围或配置 AIA_TOOL_LLM_MESSAGE_MAX_CHARS。" + "Call fetch_tool_result(result_ref=...) when result_ref is present, or use narrower queries. / " + "工具输出过大,已在上下文中截断;有 result_ref 时用 fetch_tool_result 取全文。" ), } + if existing_ref: + guarded_obj["result_ref"] = existing_ref + guarded_obj["fetch_tool"] = "fetch_tool_result" + else: + try: + from runtime.chat.tool_result_store import save_tool_result_blob + + full_obj = obj if isinstance(obj, dict) else {"ok": ok, "raw": raw} + ref = save_tool_result_blob( + session_id=str(session_id or ""), + tool_call_id=f"msg-{int(getattr(m, 'id', 0) or 0)}", + result=full_obj if isinstance(full_obj, dict) else {"payload": full_obj}, + force=True, + ) + if ref: + guarded_obj["result_ref"] = ref + guarded_obj["fetch_tool"] = "fetch_tool_result" + except Exception: + pass guarded = _json_dumps_safe(guarded_obj) out.append(_tool_message_with_content(m, guarded, sid=session_id)) if trace_id: @@ -445,6 +513,7 @@ def _guard_tool_results_for_llm_context( "original_chars": int(len(raw)), "guarded_chars": int(len(guarded)), "guard_cap_chars": int(cap), + "result_ref": str(guarded_obj.get("result_ref") or ""), }, run_id=run_id, attempt_no=attempt_no, @@ -728,13 +797,23 @@ def _prepare_llm_tools( ) -> list[dict[str, Any]]: global _TOOL_WIRE_FROZEN_SIGNATURE now = time.time() - runtime_enabled, sig = _tool_wire_settings_signature(store) + runtime_enabled, live_sig = _tool_wire_settings_signature(store) freeze_enabled = _tool_wire_freeze_enabled(store) - frozen_sig = _TOOL_WIRE_FROZEN_SIGNATURE if freeze_enabled else None + expected_frozen = f"rt={int(bool(runtime_enabled))}|{live_sig}" + with _TOOL_WIRE_CACHE_LOCK: + frozen_sig = _TOOL_WIRE_FROZEN_SIGNATURE if freeze_enabled else None + if isinstance(frozen_sig, str) and frozen_sig.strip(): + # Reuse freeze only while settings + MCP catalog fingerprint still match. + if frozen_sig == expected_frozen: + sig = frozen_sig + else: + _TOOL_WIRE_FROZEN_SIGNATURE = None + _TOOL_WIRE_CACHE.clear() + sig = live_sig + frozen_sig = None + else: + sig = live_sig if isinstance(frozen_sig, str) and frozen_sig.strip(): - # Startup-prewarmed frozen mode: execution path reuses precomputed tool wiring - # and does not perform per-turn policy revalidation. - sig = frozen_sig try: rt_head = str(frozen_sig).split("|", 1)[0].strip().lower() runtime_enabled = rt_head == "rt=1" @@ -1670,5 +1749,11 @@ def run_direct_loop(**kwargs: Any) -> TurnRunOutcome: return run_oclaw_direct_loop(**kwargs) -__all__ = ["run_oclaw_direct_loop", "run_direct_loop", "warm_tool_wire_cache", "tool_wire_freeze_status"] +__all__ = [ + "run_oclaw_direct_loop", + "run_direct_loop", + "warm_tool_wire_cache", + "invalidate_tool_wire_cache", + "tool_wire_freeze_status", +] diff --git a/runtime/tools/mcp/sync_tools.py b/runtime/tools/mcp/sync_tools.py index 64c5a18c..dd2c3128 100644 --- a/runtime/tools/mcp/sync_tools.py +++ b/runtime/tools/mcp/sync_tools.py @@ -85,6 +85,12 @@ def sync_mcp_server_tools(store: Any, row: dict[str, Any]) -> dict[str, Any] | N store.replace_mcp_server_tools(server_id=sid, tools=tools) detail = {"synced_tools": len(tools), "compat_mode": "bailian_webparser"} store.set_mcp_server_health(server_id=sid, status="ok", detail=detail) + try: + from runtime.direct_loop import invalidate_tool_wire_cache + + invalidate_tool_wire_cache(reason=f"mcp_sync:{sid}") + except Exception: + pass return {"server_id": sid, "ok": True, "health": detail, "tools_synced": len(tools)} rt = mcp_runtime_for_row(row, store=store) @@ -128,6 +134,12 @@ def sync_mcp_server_tools(store: Any, row: dict[str, Any]) -> dict[str, Any] | N "health": health, }, ) + try: + from runtime.direct_loop import invalidate_tool_wire_cache + + invalidate_tool_wire_cache(reason=f"mcp_sync:{sid}") + except Exception: + pass return { "server_id": sid, "ok": True, diff --git a/runtime/tools/public/fetch_tool_result_tool.py b/runtime/tools/public/fetch_tool_result_tool.py new file mode 100644 index 00000000..2d514933 --- /dev/null +++ b/runtime/tools/public/fetch_tool_result_tool.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from typing import Any + +from runtime.chat.tool_invocation_context import current_tool_lane_sessions +from runtime.chat.tool_result_store import load_tool_result_blob +from runtime.tools.base import ToolSpec + + +def fetch_tool_result_tool() -> ToolSpec: + def handler(args: dict[str, Any]) -> dict[str, Any]: + ref = str(args.get("result_ref") or "").strip() + if not ref: + return {"ok": False, "error_code": "result_ref_required", "error": "result_ref_required"} + owner, sid = current_tool_lane_sessions() + session_id = str(sid or owner or "").strip() + max_chars = args.get("max_chars") + try: + max_i = int(max_chars) if max_chars is not None else None + except Exception: + max_i = None + return load_tool_result_blob(ref, session_id=session_id, max_chars=max_i) + + return ToolSpec( + name="fetch_tool_result", + description=( + "Fetch a previously truncated/guarded tool result by result_ref. " + "Use when a tool payload includes result_ref / _truncated_for_llm / _tool_result_guarded " + "and you need more detail than the compact preview." + ), + parameters={ + "type": "object", + "properties": { + "result_ref": { + "type": "string", + "description": "Opaque ref from a prior tool message (e.g. tr:…).", + }, + "max_chars": { + "type": "integer", + "minimum": 4000, + "maximum": 500000, + "description": "Optional cap for returned JSON size (default ~120k).", + }, + }, + "required": ["result_ref"], + "additionalProperties": False, + }, + handler=handler, + tags=frozenset({"system", "read", "tool_result"}), + read_only=True, + risk_level="low", + timeout_s=8.0, + ) + + +__all__ = ["fetch_tool_result_tool"] diff --git a/tests/test_ops_short_intent_and_exec_hints.py b/tests/test_ops_short_intent_and_exec_hints.py index 1e2ec5c5..3068a27f 100644 --- a/tests/test_ops_short_intent_and_exec_hints.py +++ b/tests/test_ops_short_intent_and_exec_hints.py @@ -5,9 +5,9 @@ from runtime.application.gateway.ops_short_intent import ( filter_tool_specs_for_ops_short_intent, is_ops_short_intent_suppressed_tool, maybe_ops_short_intent_system_hint, + ops_short_intent_cli_soft_budgets, ) from runtime.tools.base import ToolSpec -from runtime.tools.tool_error_hints import enrich_exec_managed_ne_error, enrich_get_managed_ne_error def test_detect_ops_short_intent_english_field() -> None: @@ -20,18 +20,19 @@ def test_detect_ops_short_intent_english_field() -> None: assert detect_ops_short_intent("hello there how are you doing today with something else") is None -def test_ops_short_intent_suppresses_inventory_cli_tools() -> None: - assert is_ops_short_intent_suppressed_tool("mcp__netx__listCliTargets", intent="fiber_cut") - assert is_ops_short_intent_suppressed_tool("mcp__netx__execManagedNe", intent="offline") +def test_ops_short_intent_suppresses_diy_xlsx_shell_only() -> None: assert is_ops_short_intent_suppressed_tool("run_command", intent="excel_export") assert is_ops_short_intent_suppressed_tool("write_xlsx", intent="fiber_cut") assert is_ops_short_intent_suppressed_tool("write_xlsx", intent="excel_export") + # Soft preference: CLI/inventory stay available for device confirmation. + assert not is_ops_short_intent_suppressed_tool("mcp__netx__listCliTargets", intent="fiber_cut") + assert not is_ops_short_intent_suppressed_tool("mcp__netx__execManagedNe", intent="offline") assert not is_ops_short_intent_suppressed_tool("ume_alarm_xlsx_report", intent="fiber_cut") assert not is_ops_short_intent_suppressed_tool("mcp__netx__queryUmeAlarmsRaw", intent="alarm_tally") assert not is_ops_short_intent_suppressed_tool("mcp__netx__execManagedNe", intent="continue") -def test_filter_tool_specs_for_ops_short_intent_keeps_report_path() -> None: +def test_filter_tool_specs_for_ops_short_intent_keeps_cli_and_report() -> None: def _spec(name: str) -> ToolSpec: return ToolSpec( name=name, @@ -52,20 +53,33 @@ def test_filter_tool_specs_for_ops_short_intent_keeps_report_path() -> None: names = {t.name for t in kept} assert "ume_alarm_xlsx_report" in names assert "mcp__netx__aggregateUmeAlarms" in names + assert "mcp__netx__listCliTargets" in names + assert "mcp__netx__execManagedNe" in names assert "write_xlsx" not in names - assert "mcp__netx__listCliTargets" not in names - assert "mcp__netx__execManagedNe" not in names assert "run_command" not in names assert len(filter_tool_specs_for_ops_short_intent(tools, intent="continue")) == len(tools) +def test_ops_short_intent_cli_soft_budgets() -> None: + soft = ops_short_intent_cli_soft_budgets("fiber_cut") + assert soft is not None + assert int(soft["single"]) <= 2 + assert int(soft["batch"]) == 1 + assert ops_short_intent_cli_soft_budgets("continue") is None + assert ops_short_intent_cli_soft_budgets(None) is None + + def test_ops_short_intent_hint_english_default() -> None: hint = maybe_ops_short_intent_system_hint(text="LOS on these sites", lang="en") assert "fiber" in hint.lower() or "LOS" in hint assert "ume_alarm_xlsx_report" in hint + assert "soft budget" in hint.lower() or "targets" in hint.lower() assert "断纤" not in hint +from runtime.tools.tool_error_hints import enrich_exec_managed_ne_error, enrich_get_managed_ne_error + + def test_enrich_exec_timeout() -> None: out = enrich_exec_managed_ne_error( {"ok": False, "error_code": "tool_timeout_or_failed", "error": "timeout"} diff --git a/tests/test_tool_result_ref.py b/tests/test_tool_result_ref.py new file mode 100644 index 00000000..bbaf3d4f --- /dev/null +++ b/tests/test_tool_result_ref.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from pathlib import Path + +from runtime.chat.tool_result_store import ( + attach_result_ref, + load_tool_result_blob, + save_tool_result_blob, +) +from runtime.tools.public.fetch_tool_result_tool import fetch_tool_result_tool + + +def test_save_and_fetch_tool_result_blob(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setattr("runtime.chat.tool_result_store.attachments_dir", lambda: tmp_path) + big = {"ok": True, "rows": ["x" * 100 for _ in range(80)]} + ref = save_tool_result_blob(session_id="sess-1", tool_call_id="tc1", result=big) + assert ref and ref.startswith("tr:") + loaded = load_tool_result_blob(ref, session_id="sess-1") + assert loaded["ok"] is True + assert loaded["result"]["ok"] is True + assert loaded["truncated"] is False + + +def test_attach_result_ref_adds_fetch_hint() -> None: + out = attach_result_ref({"ok": True, "_truncated_for_llm": True, "preview": "p"}, result_ref="tr:abcd") + assert out["result_ref"] == "tr:abcd" + assert out["fetch_tool"] == "fetch_tool_result" + assert "fetch_tool_result" in str(out.get("hint") or "") + + +def test_fetch_tool_result_public_tool(monkeypatch, tmp_path: Path) -> None: + monkeypatch.setattr("runtime.chat.tool_result_store.attachments_dir", lambda: tmp_path) + ref = save_tool_result_blob( + session_id="sess-2", + tool_call_id="tc2", + result={"ok": True, "data": "hello" * 2000}, + force=True, + ) + assert ref + monkeypatch.setattr( + "runtime.tools.public.fetch_tool_result_tool.current_tool_lane_sessions", + lambda: (None, "sess-2"), + ) + tool = fetch_tool_result_tool() + out = tool.handler({"result_ref": ref}) + assert out["ok"] is True + assert out["result_ref"] == ref diff --git a/tests/test_tool_wire_freeze_switch.py b/tests/test_tool_wire_freeze_switch.py index 9edf267d..9b6af84a 100644 --- a/tests/test_tool_wire_freeze_switch.py +++ b/tests/test_tool_wire_freeze_switch.py @@ -10,6 +10,14 @@ class _DummyStore: def get_setting(self, key: str) -> str: return str(self.settings.get(key, "")) + def list_mcp_servers(self, *, enabled_only: bool = False): + del enabled_only + return [] + + def list_mcp_server_tools(self, *, server_id: str): + del server_id + return [] + def test_tool_wire_freeze_default_on(monkeypatch) -> None: monkeypatch.setattr(dl, "_prepare_llm_tools", lambda **kwargs: []) @@ -52,3 +60,35 @@ def test_warm_tool_wire_cache_clears_frozen_stale_entries(monkeypatch) -> None: st = dl.tool_wire_freeze_status(store=store) assert st["frozen"] is True assert st["last_warm_count"] == 2 + + +def test_invalidate_tool_wire_cache_clears_freeze(monkeypatch) -> None: + monkeypatch.setattr(dl, "_prepare_llm_tools", lambda **kwargs: []) + store = _DummyStore({"AIA_TOOL_WIRE_FROZEN_ON_STARTUP": "1"}) + _ = dl.warm_tool_wire_cache(store=store, tools=object(), base_url="", roles=["generalist"]) + assert dl.tool_wire_freeze_status(store=store)["frozen"] is True + out = dl.invalidate_tool_wire_cache(reason="test") + assert out["ok"] is True + assert dl.tool_wire_freeze_status(store=store)["frozen"] is False + + +def test_mcp_tools_fingerprint_changes_with_catalog() -> None: + class _Store(_DummyStore): + def __init__(self): + super().__init__() + self.tools = [{"tool_name": "a"}] + + def list_mcp_servers(self, *, enabled_only: bool = False): + del enabled_only + return [{"server_id": "netx"}] + + def list_mcp_server_tools(self, *, server_id: str): + del server_id + return list(self.tools) + + store = _Store() + fp1 = dl._mcp_tools_fingerprint(store) + store.tools = [{"tool_name": "a"}, {"tool_name": "b"}] + fp2 = dl._mcp_tools_fingerprint(store) + assert fp1 != fp2 + assert fp1 != "x"