From 54cb8af8b0ce4f78db86f5d9bbafcce4c50600a8 Mon Sep 17 00:00:00 2001 From: oliver Date: Mon, 10 Aug 2026 23:19:03 +0800 Subject: [PATCH] Block non-retryable tool re-calls and add congestion ops recipes. After insufficient_scope/retry_forbidden, refuse the same tool for the rest of the turn (even with new args), emit pre-call WA progress from ToolExecutor, TTL-cache findTopologyPaths, and add bandwidth congestion short-intent + schedule template. Co-authored-by: Cursor --- .../application/gateway/ops_short_intent.py | 11 +- runtime/chat/tool_runtime.py | 109 +++++++++++++++++- runtime/scheduler/recipe.py | 21 ++++ runtime/tools/mcp/adapter.py | 1 + tests/test_failure_class_and_retry_guard.py | 58 ++++++++++ tests/test_schedule_recipe.py | 4 + 6 files changed, 201 insertions(+), 3 deletions(-) diff --git a/runtime/application/gateway/ops_short_intent.py b/runtime/application/gateway/ops_short_intent.py index eda810bf..1b29af1d 100644 --- a/runtime/application/gateway/ops_short_intent.py +++ b/runtime/application/gateway/ops_short_intent.py @@ -34,6 +34,13 @@ _HINTS: dict[str, tuple[str, str]] = { "[短指令：License/容量。优先 aggregateUmeAlarms / queryUmeAlarmsRaw（license 关键字）" "或 ume_alarm_xlsx_report(mode=list, keyword=license)；≤3 次工具，勿刷 CLI。]", ), + "congestion": ( + "[Ops short-intent: bandwidth congestion. Prefer aggregateUmeAlarms / queryUmeAlarmsRaw " + "(bandwidth/utilization/congestion keywords) or ume_alarm_xlsx_report(mode=list); " + "≤3 tool calls — no CLI spam / no sqlQueryUme unless scoped.]", + "[短指令：带宽拥塞。优先 aggregateUmeAlarms / queryUmeAlarmsRaw（带宽/利用率/拥塞）" + "或 ume_alarm_xlsx_report(mode=list)；≤3 次工具，勿刷 CLI / 勿先 sqlQueryUme。]", + ), "continue": ( "[Ops short-intent: continue/confirm. Resume the unfinished prior task immediately; " "do not re-ask confirmation or restart the query from scratch.]", @@ -64,7 +71,9 @@ def detect_ops_short_intent(text: str) -> str | None: return "offline" if any(k in t for k in ("excel", "xlsx", "spreadsheet", "export", "send me the table", "表格", "导出")): return "excel_export" - if any(k in t for k in ("license", "licence", "capacity", "带宽", "拥塞", "license到期")): + if any(k in t for k in ("congest", "bandwidth", "utilization", "拥塞", "带宽", "利用率", "端口忙")): + return "congestion" + if any(k in t for k in ("license", "licence", "capacity", "license到期")): return "license" if any( k in t diff --git a/runtime/chat/tool_runtime.py b/runtime/chat/tool_runtime.py index 83e42677..449f2997 100644 --- a/runtime/chat/tool_runtime.py +++ b/runtime/chat/tool_runtime.py @@ -363,6 +363,59 @@ def _load_turn_failed_signatures(store: Any, *, session_id: str, turn_uuid: str) return out +def _result_is_retry_forbidden(payload: dict[str, Any] | None) -> bool: + if not isinstance(payload, dict): + return False + if payload.get("retry_forbidden") is True: + return True + code = str(payload.get("error_code") or "").strip().lower() + if code == "insufficient_scope": + return True + fc = str(payload.get("failure_class") or "").strip().lower() + if fc == "auth" and "scope" in f"{payload.get('error') or ''} {code}".lower(): + return True + return False + + +def _load_turn_retry_forbidden_tools(store: Any, *, session_id: str, turn_uuid: str) -> set[str]: + """Tool names that returned retry_forbidden / insufficient_scope earlier in this turn.""" + tu = str(turn_uuid or "").strip() + sid = str(session_id or "").strip() + out: set[str] = set() + if not tu or not sid: + return out + try: + rows = store.get_messages(session_id=sid, limit=500) + except Exception: + return out + for m in rows or []: + if str(getattr(m, "role", "") or "").strip().lower() != "tool": + continue + if str(getattr(m, "turn_uuid", "") or "").strip() != tu: + continue + ep = _parse_event_payload(getattr(m, "event_payload", None)) + name = str(ep.get("tool_name") or "").strip() + if ep.get("retry_forbidden") is True and name: + out.add(name) + continue + try: + payload = json.loads(str(getattr(m, "content", "") or "") or "{}") + except Exception: + payload = {} + if isinstance(payload, dict) and _result_is_retry_forbidden(payload): + raw_tc = getattr(m, "tool_calls", None) + if isinstance(raw_tc, str): + try: + raw_tc = json.loads(raw_tc) + except Exception: + raw_tc = None + if not name and isinstance(raw_tc, dict): + name = str(raw_tc.get("name") or "").strip() + if name: + out.add(name) + return out + + def normalize_tool_result(result: Any) -> dict[str, Any]: if isinstance(result, dict): out = dict(result) @@ -985,6 +1038,11 @@ class ToolExecutor: session_id=ctx.session_id, turn_uuid=str(ctx.turn_uuid or ""), ) + retry_forbidden_tools = _load_turn_retry_forbidden_tools( + ctx.store, + session_id=ctx.session_id, + turn_uuid=str(ctx.turn_uuid or ""), + ) results_by_id: dict[str, tuple[dict[str, Any], int]] = {} runnable_tool_uses: list[LLMToolCall] = [] @@ -993,6 +1051,27 @@ class ToolExecutor: sig_seen: dict[str, int] = {} budget = max(1, min(int(signature_budget or 2), 8)) for tc in tool_uses: + if str(tc.name or "") in retry_forbidden_tools: + results_by_id[tc.id] = ( + { + "ok": False, + "error_code": "retry_forbidden_blocked", + "failure_class": "retry_guard", + "retry_forbidden": True, + "error": f"tool blocked for remainder of turn after non-retryable failure: {tc.name}", + "hint": ( + "This tool already failed with a non-retryable error (e.g. insufficient_scope). " + "Switch to fallback tools (aggregateUmeAlarms / queryUmeAlarmsRaw / " + "ume_alarm_xlsx_report) or ask an admin — do not call it again this turn." + ), + }, + 0, + ) + _trace( + "retry_forbidden_blocked", + {"tool_name": tc.name}, + ) + continue if tc.name in _TABULAR_QUERY_TOOL_NAMES and not has_tabular_ref_in_session: results_by_id[tc.id] = ( { @@ -1099,6 +1178,13 @@ class ToolExecutor: "listclitargets", "listmanagedne", "queryumeneinventory", + "queryumealarms", + "queryumealarmsraw", + "aggregateumealarms", + "aggregateumealarmsraw", + "runumediagnostics", + "sqlqueryume", + "findtopologypaths", ) ) if count >= budget: @@ -1108,8 +1194,8 @@ class ToolExecutor: "error_code": "tool_loop_guard", "error": f"tool loop guard triggered for signature: {tc.name}", "hint": ( - "Identical list/inventory call already ran this turn; reuse prior ids/rows " - "instead of listing again." + "Identical list/query/aggregate call already ran this turn; reuse prior rows " + "or change filters — do not re-query identically." if listish else "Identical tool call already ran this turn; change arguments or continue without retry." ), @@ -1150,6 +1236,20 @@ class ToolExecutor: "tool_names": [str(getattr(x, "name", "") or "") for x in batch], }, ) + if on_tool_ui: + for tc in batch: + try: + on_tool_ui( + "tool_use_call", + { + "phase": "call", + "tool_name": str(tc.name or ""), + "tool_call_id": str(tc.id or ""), + "arguments": dict(tc.arguments or {}), + }, + ) + except Exception: + pass if len(batch) > 1: workers = min(int(self.config.max_workers), len(batch)) with ThreadPoolExecutor(max_workers=workers) as ex: @@ -1187,6 +1287,8 @@ class ToolExecutor: result = normalize_tool_result(result) if isinstance(result, dict) and result.get("ok") is False: failed_signatures.add(f"{tc.name}:{self._json_dumps_safe(dict(tc.arguments or {}))}") + if isinstance(result, dict) and _result_is_retry_forbidden(result): + retry_forbidden_tools.add(str(tc.name or "")) persisted_result, ingested_refs = ingest_embedded_image_blobs_as_refs( result, filename_prefix=f"{str(tc.name or 'tool')}-{str(tc.id or '')}", @@ -1243,6 +1345,9 @@ class ToolExecutor: "ok": bool(result.get("ok")) if isinstance(result, dict) else False, "tool_signature": tool_sig[:800], "failure_class": str((result or {}).get("failure_class") or "") if isinstance(result, dict) else "", + "retry_forbidden": bool( + isinstance(result, dict) and _result_is_retry_forbidden(result) + ), }, ) try: diff --git a/runtime/scheduler/recipe.py b/runtime/scheduler/recipe.py index 6fd0033c..7e47707e 100644 --- a/runtime/scheduler/recipe.py +++ b/runtime/scheduler/recipe.py @@ -332,6 +332,9 @@ _OPS_RECIPE_TEMPLATE_ALIASES: dict[str, str] = { "critical_xlsx_daily": "ume_critical_xlsx_daily", "license_check": "ne_license_check_weekly", "license_weekly": "ne_license_check_weekly", + "congestion": "bandwidth_congestion_daily", + "bandwidth": "bandwidth_congestion_daily", + "bandwidth_congestion": "bandwidth_congestion_daily", } OPS_RECIPE_TEMPLATES: dict[str, dict[str, Any]] = { @@ -389,6 +392,24 @@ OPS_RECIPE_TEMPLATES: dict[str, dict[str, Any]] = { ], "output": {"need_attachments": False}, }, + "bandwidth_congestion_daily": { + "version": 1, + "goal": "Daily bandwidth congestion / utilization hotspot summary for ops WhatsApp", + "steps": [ + "Call aggregateUmeAlarms or queryUmeAlarmsRaw with bandwidth/congestion/utilization keywords", + "Optionally ume_alarm_xlsx_report(mode=list) if the user wants a file (deliverable=true)", + "Summarize top congested hosts/ports in concise English — avoid sqlQueryUme unless scoped", + ], + "constraints": [ + "Prefer English for WhatsApp field ops", + "Do not spam CLI or identical alarm re-queries", + "If insufficient_scope on SQL, switch to aggregate/report tools immediately", + ], + "success_criteria": [ + "Group receives a congestion/utilization hotspot summary with freshness", + ], + "output": {"need_attachments": False}, + }, } diff --git a/runtime/tools/mcp/adapter.py b/runtime/tools/mcp/adapter.py index a0eab9d1..f47203dd 100644 --- a/runtime/tools/mcp/adapter.py +++ b/runtime/tools/mcp/adapter.py @@ -41,6 +41,7 @@ _MCP_LIST_CACHE_TTL_S: dict[str, float] = { "aggregateUmeAlarms": 45.0, "aggregateUmeAlarmsRaw": 45.0, "runUmeDiagnostics": 60.0, + "findTopologyPaths": 60.0, } _MCP_LIST_CACHE_LOCK = threading.Lock() diff --git a/tests/test_failure_class_and_retry_guard.py b/tests/test_failure_class_and_retry_guard.py index bf97cf48..685f6701 100644 --- a/tests/test_failure_class_and_retry_guard.py +++ b/tests/test_failure_class_and_retry_guard.py @@ -31,6 +31,11 @@ def test_license_short_intent() -> None: assert detect_ops_short_intent("@bot licence check") == "license" +def test_congestion_short_intent() -> None: + assert detect_ops_short_intent("bandwidth congestion top") == "congestion" + assert detect_ops_short_intent("端口忙 拥塞") == "congestion" + + def test_identical_failed_retry_blocked_across_rounds(tmp_path: Path) -> None: store = SqliteStore(str(tmp_path / "retry.sqlite")) sess = store.create_session("t") @@ -70,3 +75,56 @@ def test_identical_failed_retry_blocked_across_rounds(tmp_path: Path) -> None: assert calls["n"] == 1 assert blocked.get("error_code") == "identical_retry_blocked" assert blocked.get("failure_class") == "retry_guard" + + +def test_retry_forbidden_blocks_same_tool_different_args(tmp_path: Path) -> None: + store = SqliteStore(str(tmp_path / "scope.sqlite")) + sess = store.create_session("t") + calls = {"n": 0} + + def _handler(_args): + calls["n"] += 1 + return { + "ok": False, + "error_code": "insufficient_scope", + "error": "insufficient_scope:sql:query", + "retry_forbidden": True, + "failure_class": "auth", + } + + reg = ToolRegistry( + [ + ToolSpec( + name="mcp__netx__sqlQueryUme", + description="sql", + parameters={"type": "object", "properties": {"sql": {"type": "string"}}}, + handler=_handler, + read_only=True, + ) + ] + ) + ctx = ToolExecutionContext( + store=store, + tools=reg, + session_id=sess.id, + turn_uuid="turn-scope-1", + lang="en", + ) + ToolExecutor().execute_tool_uses( + ctx=ctx, + assistant_msg_id=1, + tool_uses=[LLMToolCall(id="c1", name="mcp__netx__sqlQueryUme", arguments={"sql": "select 1"})], + signature_budget=2, + ) + assert calls["n"] == 1 + + _, results = ToolExecutor().execute_tool_uses( + ctx=ctx, + assistant_msg_id=2, + tool_uses=[LLMToolCall(id="c2", name="mcp__netx__sqlQueryUme", arguments={"sql": "select 2"})], + signature_budget=2, + ) + blocked, _ = results["c2"] + assert calls["n"] == 1 + assert blocked.get("error_code") == "retry_forbidden_blocked" + assert blocked.get("retry_forbidden") is True diff --git a/tests/test_schedule_recipe.py b/tests/test_schedule_recipe.py index cac9c802..6b21981b 100644 --- a/tests/test_schedule_recipe.py +++ b/tests/test_schedule_recipe.py @@ -75,10 +75,14 @@ class RecipeHelpersTests(unittest.TestCase): self.assertIn("ume_alarm_tally_daily", ids) self.assertIn("ume_critical_xlsx_daily", ids) self.assertIn("ne_license_check_weekly", ids) + self.assertIn("bandwidth_congestion_daily", ids) tmpl = resolve_ops_recipe_template("alarm_tally") assert tmpl is not None self.assertTrue(recipe_has_playbook(tmpl)) self.assertEqual((tmpl.get("source") or {}).get("template_id"), "ume_alarm_tally_daily") + cong = resolve_ops_recipe_template("congestion") + assert cong is not None + self.assertEqual((cong.get("source") or {}).get("template_id"), "bandwidth_congestion_daily") self.assertIsNone(resolve_ops_recipe_template("nope")) def test_turn_instruction_modes(self) -> None: