From 4827278a9c8a8aa84d2fb31c8619e9d1c65f969f Mon Sep 17 00:00:00 2001 From: oliver Date: Wed, 12 Aug 2026 22:17:16 +0800 Subject: [PATCH] Enforce report-first short intents and async large execManagedNe batches. Block CLI/inventory until ume_alarm_xlsx_report (or aggregate) succeeds, and return job_id for big NE batches via get_ne_exec_job. Co-authored-by: Cursor --- .../application/gateway/ops_short_intent.py | 56 ++-- runtime/chat/exec_managed_ne_guard.py | 6 +- runtime/chat/ops_report_first_guard.py | 161 +++++++++++ runtime/chat/tool_runtime.py | 88 +++++- runtime/chat/turn_idle_guard.py | 1 + runtime/tools/ops_system_tool_allowlist.py | 2 + runtime/tools/playbook_contracts.py | 4 +- runtime/tools/public/get_ne_exec_job_tool.py | 35 +++ svc/jobs/ne_exec_jobs.py | 264 ++++++++++++++++++ tests/test_ops_short_intent_and_exec_hints.py | 2 +- tests/test_report_first_and_ne_exec_async.py | 110 ++++++++ 11 files changed, 695 insertions(+), 34 deletions(-) create mode 100644 runtime/chat/ops_report_first_guard.py create mode 100644 runtime/tools/public/get_ne_exec_job_tool.py create mode 100644 svc/jobs/ne_exec_jobs.py create mode 100644 tests/test_report_first_and_ne_exec_async.py diff --git a/runtime/application/gateway/ops_short_intent.py b/runtime/application/gateway/ops_short_intent.py index c8957a18..fb1ae244 100644 --- a/runtime/application/gateway/ops_short_intent.py +++ b/runtime/application/gateway/ops_short_intent.py @@ -9,49 +9,45 @@ _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). +# Runtime also enforces report-first: CLI/inventory blocked until report/aggregate succeeds this turn. _HINTS: dict[str, tuple[str, str]] = { "excel_export": ( - "[Ops short-intent: Excel export. Prefer ume_alarm_xlsx_report(..., deliverable=true) now. " + "[Ops short-intent: Excel export. FIRST tool call MUST be ume_alarm_xlsx_report(..., deliverable=true). " "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，勿单台循环。]", + "CLI/inventory only AFTER report ok — one hetero targets batch max for device checks.]", + "[短指令：导出 Excel。本轮第一个工具必须是 ume_alarm_xlsx_report(..., deliverable=true)。" + "已隐藏 write_xlsx/run_command。" + "报表 ok 之后才允许 CLI/清单；设备核实最多一次 hetero targets batch。]", ), "license": ( - "[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。]", + "[Ops short-intent: license/capacity. FIRST call ume_alarm_xlsx_report(mode=list, keyword=license, deliverable=true) " + "(or aggregateUmeAlarms). write_xlsx/run_command hidden; CLI only after report ok.]", + "[短指令：License/容量。先 ume_alarm_xlsx_report(mode=list, keyword=license, deliverable=true) " + "（或 aggregate）；write_xlsx/run_command 已隐藏；报表 ok 后再 CLI。]", ), "congestion": ( - "[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。]", + "[Ops short-intent: bandwidth congestion. FIRST call ume_alarm_xlsx_report(mode=list, deliverable=true) " + "(or aggregate). write_xlsx/run_command hidden; CLI only after report ok.]", + "[短指令：带宽拥塞。先 ume_alarm_xlsx_report(mode=list, deliverable=true)（或 aggregate）；" + "write_xlsx/run_command 已隐藏；报表 ok 后再 CLI。]", ), "fiber_cut": ( - "[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→单台循环。]", + "[Ops short-intent: fiber/LOS. FIRST tool MUST be ume_alarm_xlsx_report(mode=fiber_cut, deliverable=true). " + "write_xlsx/run_command hidden. CLI/inventory ONLY after report ok — at most one hetero targets batch.]", + "[短指令：断纤/LOS。本轮第一个工具必须是 ume_alarm_xlsx_report(mode=fiber_cut, deliverable=true)。" + "write_xlsx/run_command 已隐藏。报表 ok 之后才允许 CLI/清单（最多一次 hetero targets batch）。]", ), "offline": ( - "[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）。]", + "[Ops short-intent: offline NE. FIRST tool MUST be ume_alarm_xlsx_report(mode=offline, deliverable=true). " + "write_xlsx/run_command hidden; CLI only after report ok.]", + "[短指令：离线网元。本轮第一个工具必须是 ume_alarm_xlsx_report(mode=offline, deliverable=true)。" + "write_xlsx/run_command 已隐藏；报表 ok 后再 CLI。]", ), "alarm_tally": ( - "[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 仅作抽检。]", + "[Ops short-intent: alarm tally/top. FIRST call ume_alarm_xlsx_report(mode=aggregate_by_host, deliverable=true) " + "or aggregateUmeAlarms. write_xlsx/run_command hidden; CLI only after report ok.]", + "[短指令：告警统计/Top。先 ume_alarm_xlsx_report(mode=aggregate_by_host, deliverable=true) " + "或 aggregateUmeAlarms；write_xlsx/run_command 已隐藏；报表 ok 后再 CLI。]", ), "continue": ( "[Ops short-intent: continue/confirm. Resume the unfinished prior task immediately; " diff --git a/runtime/chat/exec_managed_ne_guard.py b/runtime/chat/exec_managed_ne_guard.py index fac68dc8..d04441e0 100644 --- a/runtime/chat/exec_managed_ne_guard.py +++ b/runtime/chat/exec_managed_ne_guard.py @@ -48,8 +48,12 @@ def is_batch_exec_args(args: dict[str, Any] | None) -> bool: def normalize_exec_managed_ne_args(args: dict[str, Any] | None) -> dict[str, Any]: - """Default / clamp read_timeout_sec so agents stop hitting 30s walls.""" + """Default / clamp read_timeout_sec so agents stop hitting 30s walls. + + Also strips oclaw-only ``async`` flag before the MCP HTTP call. + """ out = dict(args or {}) + out.pop("async", None) rts = out.get("read_timeout_sec") if rts is None or str(rts).strip() == "": out["read_timeout_sec"] = 60 diff --git a/runtime/chat/ops_report_first_guard.py b/runtime/chat/ops_report_first_guard.py new file mode 100644 index 00000000..4bcbf69d --- /dev/null +++ b/runtime/chat/ops_report_first_guard.py @@ -0,0 +1,161 @@ +"""Report-first gate for ops short intents: CLI/inventory only after report path.""" + +from __future__ import annotations + +import json +from typing import Any + +from runtime.application.gateway.ops_short_intent import ops_short_intent_should_filter_tools +from runtime.tools.playbook_contracts import playbook_example_for_tool, short_intent_first_step + +# Tools blocked until a report-path tool succeeds this turn. +_CLI_BEFORE_REPORT_TOOLS = frozenset( + { + "listclitargets", + "listmanagedne", + "getmanagedne", + "execmanagedne", + "queryumeneinventory", + "getumene", + "findtopologypaths", + "sqlqueryume", + "netx_list_managed_ne", + "netx_get_managed_ne", + "netx_exec_managed_ne", + "netx_sql_query_ume", + "netx_list_cli_targets", + } +) + +# Successful call to any of these unlocks CLI for the rest of the turn. +_REPORT_PATH_TOOLS = frozenset( + { + "ume_alarm_xlsx_report", + "aggregateumealarms", + "aggregateumealarmsraw", + } +) + + +def _tool_key(name: str) -> str: + raw = str(name or "").strip() + if not raw: + return "" + if "__" in raw: + raw = raw.rsplit("__", 1)[-1] + return raw.strip().lower().replace("-", "_") + + +def is_cli_before_report_tool(tool_name: str) -> bool: + return _tool_key(tool_name) in _CLI_BEFORE_REPORT_TOOLS + + +def is_report_path_tool(tool_name: str) -> bool: + return _tool_key(tool_name) in _REPORT_PATH_TOOLS + + +def _parse_json_obj(raw: Any) -> dict[str, Any]: + if isinstance(raw, dict): + return dict(raw) + if isinstance(raw, str) and raw.strip(): + try: + data = json.loads(raw) + return data if isinstance(data, dict) else {} + except Exception: + return {} + return {} + + +def turn_has_successful_report_path( + store: Any, + *, + session_id: str, + turn_uuid: str, +) -> bool: + """True if this turn already has a successful report/aggregate tool result.""" + tu = str(turn_uuid or "").strip() + sid = str(session_id or "").strip() + if not tu or not sid or store is None: + return False + try: + rows = store.get_messages(session_id=sid, limit=500) + except Exception: + return False + 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_json_obj(getattr(m, "event_payload", None)) + name = str(ep.get("tool_name") or "").strip() + if not name: + tc = _parse_json_obj(getattr(m, "tool_calls", None)) + name = str(tc.get("name") or "").strip() + if not is_report_path_tool(name): + continue + if ep.get("ok") is True: + return True + try: + payload = json.loads(str(getattr(m, "content", "") or "") or "{}") + except Exception: + payload = {} + if isinstance(payload, dict) and payload.get("ok") is True: + return True + return False + + +def report_first_block_payload(*, intent: str | None, lang: str = "en") -> dict[str, Any]: + en = str(lang or "").strip().lower().startswith("en") + step = short_intent_first_step(intent) + tool = "ume_alarm_xlsx_report" + example = playbook_example_for_tool(tool, intent=intent) or {"mode": "list", "deliverable": True} + if step: + tool, example = step[0], step[1] + hint = ( + f"Report-first gate: call {tool} successfully before CLI/inventory this turn. " + "Device confirmation via execManagedNe is allowed only after the report path returns ok." + if en + else f"报表优先：本轮先成功调用 {tool}，再允许 CLI/清单。" + "设备核实（execManagedNe）仅在报表路径 ok 之后。" + ) + return { + "ok": False, + "error_code": "report_first_required", + "failure_class": "report_first", + "error": "report_first_required", + "hint": hint, + "next_tool": tool, + "example": example, + "intent": str(intent or ""), + } + + +def maybe_block_cli_before_report( + *, + tool_name: str, + intent: str | None, + store: Any, + session_id: str, + turn_uuid: str, + lang: str = "en", + local_report_ok: bool = False, +) -> dict[str, Any] | None: + """Return a block payload when CLI/inventory is used before report on short intents.""" + if not ops_short_intent_should_filter_tools(intent): + return None + if not is_cli_before_report_tool(tool_name): + return None + if local_report_ok: + return None + if turn_has_successful_report_path(store, session_id=session_id, turn_uuid=turn_uuid): + return None + return report_first_block_payload(intent=intent, lang=lang) + + +__all__ = [ + "is_cli_before_report_tool", + "is_report_path_tool", + "maybe_block_cli_before_report", + "report_first_block_payload", + "turn_has_successful_report_path", +] diff --git a/runtime/chat/tool_runtime.py b/runtime/chat/tool_runtime.py index 8bef02b2..0a1a2bbc 100644 --- a/runtime/chat/tool_runtime.py +++ b/runtime/chat/tool_runtime.py @@ -849,6 +849,56 @@ class ToolExecutor: if timeout_s is None and "plugin" in getattr(tool, "tags", frozenset()): timeout_s = 30.0 + # Large execManagedNe batches: return job_id immediately and run in background. + try: + from runtime.chat.exec_managed_ne_guard import is_exec_managed_ne_tool + from svc.jobs.ne_exec_jobs import ( + should_run_exec_managed_ne_async, + start_ne_exec_job, + strip_async_flag, + ) + + if is_exec_managed_ne_tool(str(tc.name or "")): + decision_args = dict(tool_args if isinstance(tool_args, dict) else {}) + if isinstance(raw_args, dict) and "async" in raw_args: + decision_args["async"] = raw_args.get("async") + if should_run_exec_managed_ne_async(decision_args): + sync_args = strip_async_flag(tool_args if isinstance(tool_args, dict) else {}) + + def _bg_call() -> dict[str, Any]: + ws_ns = "" + raw_ws = str(ctx.workspace_dir or "").strip() + if raw_ws: + try: + wp = Path(raw_ws) + ws_ns = str(wp.name or wp.stem or "").strip() + except Exception: + ws_ns = "" + with workspace_path_access_scope( + ctx.store, + ctx.session_id, + owner_fallback_session_id=ctx.workspace_owner_session_id, + allowlist_tenant_id=ctx.path_policy_tenant_id, + allowlist_user_id=ctx.path_policy_user_id, + ), workspace_write_namespace_scope(ws_ns), tool_workspace_lane_scope( + workspace_owner_session_id=ctx.workspace_owner_session_id, + session_id=ctx.session_id, + workspace_lane_role=ctx.workspace_lane_role, + ): + raw = tool.handler(sync_args) + return normalize_tool_result(raw) + + ack = start_ne_exec_job( + tool_name=str(tc.name or ""), + arguments=sync_args, + runner=_bg_call, + session_id=str(ctx.session_id or ""), + timeout_s=int(float(timeout_s)) if isinstance(timeout_s, (int, float)) and float(timeout_s) > 0 else None, + ) + return normalize_tool_result(ack), int((time.perf_counter() - t0) * 1000) + except Exception: + pass + def _call() -> Any: ws_ns = "" raw_ws = str(ctx.workspace_dir or "").strip() @@ -1096,6 +1146,10 @@ class ToolExecutor: load_turn_exec_managed_ne_stats, ) from runtime.application.gateway.ops_short_intent import ops_short_intent_cli_soft_budgets + from runtime.chat.ops_report_first_guard import ( + is_report_path_tool, + maybe_block_cli_before_report, + ) prior_single_exec, prior_batch_exec, prior_exec_fails = load_turn_exec_managed_ne_stats( ctx.store, @@ -1105,9 +1159,11 @@ class ToolExecutor: single_exec_budget = exec_managed_ne_single_budget() fail_exec_budget = exec_managed_ne_fail_budget() batch_exec_budget = 0 # 0 = unlimited (global default) + short_intent = None 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) + short_intent = str(md_budget.get("ops_short_intent") or "").strip() or None + soft = ops_short_intent_cli_soft_budgets(short_intent) if soft: single_exec_budget = int(soft.get("single") or single_exec_budget) fail_exec_budget = int(soft.get("fail") or fail_exec_budget) @@ -1117,6 +1173,17 @@ class ToolExecutor: local_single_exec = 0 local_batch_exec = 0 local_exec_fails = 0 + local_report_ok = False + try: + from runtime.chat.ops_report_first_guard import turn_has_successful_report_path + + local_report_ok = turn_has_successful_report_path( + ctx.store, + session_id=str(ctx.session_id or ""), + turn_uuid=str(ctx.turn_uuid or ""), + ) + except Exception: + local_report_ok = False results_by_id: dict[str, tuple[dict[str, Any], int]] = {} runnable_tool_uses: list[LLMToolCall] = [] @@ -1246,6 +1313,22 @@ class ToolExecutor: }, ) continue + report_block = maybe_block_cli_before_report( + tool_name=tool_name, + intent=short_intent, + store=ctx.store, + session_id=str(ctx.session_id or ""), + turn_uuid=str(ctx.turn_uuid or ""), + lang=str(ctx.lang or "en"), + local_report_ok=local_report_ok, + ) + if report_block is not None: + results_by_id[tc.id] = (report_block, 0) + _trace( + "report_first_required", + {"tool_name": tc.name, "intent": short_intent or ""}, + ) + continue 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) @@ -1446,8 +1529,11 @@ class ToolExecutor: "identical_retry_blocked", "retry_forbidden_blocked", "tool_loop_guard", + "report_first_required", }: local_exec_fails += 1 + if isinstance(result, dict) and result.get("ok") is True and is_report_path_tool(str(tc.name or "")): + local_report_ok = True 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( diff --git a/runtime/chat/turn_idle_guard.py b/runtime/chat/turn_idle_guard.py index 6eaefb08..1364ef3b 100644 --- a/runtime/chat/turn_idle_guard.py +++ b/runtime/chat/turn_idle_guard.py @@ -101,6 +101,7 @@ class TurnIdleTracker: "cli_call_budget_exceeded", "cli_fail_budget_exceeded", "cli_batch_budget_exceeded", + "report_first_required", }: guard += 1 stats = RoundStats( diff --git a/runtime/tools/ops_system_tool_allowlist.py b/runtime/tools/ops_system_tool_allowlist.py index c8019d78..ce2be00c 100644 --- a/runtime/tools/ops_system_tool_allowlist.py +++ b/runtime/tools/ops_system_tool_allowlist.py @@ -51,6 +51,8 @@ _OPS_SYSTEM_TOOL_ALLOWLIST = frozenset( "get_env", # Compact tool-result refetch "fetch_tool_result", + # Background NE CLI poll + "get_ne_exec_job", } ) diff --git a/runtime/tools/playbook_contracts.py b/runtime/tools/playbook_contracts.py index 15528680..38907c40 100644 --- a/runtime/tools/playbook_contracts.py +++ b/runtime/tools/playbook_contracts.py @@ -151,10 +151,12 @@ def build_turn_checklist( lines.append(f"- goal: {goal_s[:160]}") if step: tool, example = step - lines.append(f"- step1: {tool}({_fmt_args(example)})") + lines.append(f"- step1 (REQUIRED first): {tool}({_fmt_args(example)})") if is_zh: + lines.append("- 未完成 step1 前禁止 listCliTargets/execManagedNe/清单循环") lines.append("- 完成后用 Result/Evidence 短答；勿翻页或开无关 playbook") else: + lines.append("- Do NOT call listCliTargets/execManagedNe/inventory before step1 succeeds") lines.append("- then reply with Result/Evidence; no pagination / unrelated playbooks") elif is_zh: lines.append("- 需要证据时立刻调用工具；失败时改参数或换 fallback，禁止相同参数盲重试") diff --git a/runtime/tools/public/get_ne_exec_job_tool.py b/runtime/tools/public/get_ne_exec_job_tool.py new file mode 100644 index 00000000..a613ed74 --- /dev/null +++ b/runtime/tools/public/get_ne_exec_job_tool.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +from typing import Any + +from runtime.tools.base import ToolSpec +from svc.jobs.ne_exec_jobs import get_ne_exec_job + + +def get_ne_exec_job_tool() -> ToolSpec: + def handler(args: dict[str, Any]) -> dict[str, Any]: + return get_ne_exec_job(str(args.get("job_id") or "").strip()) + + return ToolSpec( + name="get_ne_exec_job", + description=( + "Poll a background execManagedNe job started with async=true (or auto-async for large batches). " + "Pass job_id from the async ack. When status is succeeded/failed/timeout, result contains the full CLI output." + ), + parameters={ + "type": "object", + "properties": { + "job_id": {"type": "string", "description": "Job id returned by async execManagedNe."}, + }, + "required": ["job_id"], + "additionalProperties": False, + }, + handler=handler, + tags=frozenset({"netx", "ops", "jobs", "read"}), + read_only=True, + risk_level="low", + timeout_s=8.0, + ) + + +__all__ = ["get_ne_exec_job_tool"] diff --git a/svc/jobs/ne_exec_jobs.py b/svc/jobs/ne_exec_jobs.py new file mode 100644 index 00000000..10404395 --- /dev/null +++ b/svc/jobs/ne_exec_jobs.py @@ -0,0 +1,264 @@ +"""Background jobs for long execManagedNe batches (oclaw-side, disk + thread).""" + +from __future__ import annotations + +import json +import os +import threading +import time +import uuid +from pathlib import Path +from typing import Any, Callable + +from svc.config.paths import PROJECT_ROOT + +_STATUS_RUNNING = "running" +_STATUS_SUCCEEDED = "succeeded" +_STATUS_FAILED = "failed" +_STATUS_TIMEOUT = "timeout" +_TERMINAL = {_STATUS_SUCCEEDED, _STATUS_FAILED, _STATUS_TIMEOUT} +_LOCK = threading.Lock() +_INFLIGHT = 0 +_MAX_CONCURRENT = 3 +_DEFAULT_TIMEOUT_S = 900 + + +def _jobs_dir() -> Path: + override = str(os.getenv("AIA_NE_EXEC_JOB_DIR") or "").strip() + if override: + p = Path(override).expanduser().resolve() + else: + data = (Path(PROJECT_ROOT) / "data").resolve() + nested = (Path(PROJECT_ROOT) / "oclaw" / "data").resolve() + root = nested if nested.exists() else data + p = (root / "ne_exec_jobs").resolve() + p.mkdir(parents=True, exist_ok=True) + return p + + +def _job_path(job_id: str) -> Path: + return _jobs_dir() / f"{job_id}.json" + + +def _atomic_write(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(payload, ensure_ascii=False, default=str), encoding="utf-8") + tmp.replace(path) + + +def _read_job(job_id: str) -> dict[str, Any] | None: + path = _job_path(job_id) + if not path.is_file(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except Exception: + return None + return data if isinstance(data, dict) else None + + +def _write_job(job: dict[str, Any]) -> None: + jid = str(job.get("job_id") or "").strip() + if not jid: + return + _atomic_write(_job_path(jid), job) + + +def _env_int(name: str, default: int, *, min_v: int, max_v: int) -> int: + raw = str(os.getenv(name) or "").strip() + if not raw: + return default + try: + n = int(raw) + except Exception: + return default + return max(min_v, min(int(n), max_v)) + + +def async_min_ne_count() -> int: + """Auto-async when batch NE count >= this (set 0 to disable auto; explicit async=true still works).""" + return _env_int("AIA_EXEC_MANAGED_NE_ASYNC_MIN_NES", 4, min_v=0, max_v=50) + + +def count_exec_ne_targets(args: dict[str, Any] | None) -> int: + a = args if isinstance(args, dict) else {} + n = 0 + for key in ("ne_ids", "ume_ne_ids"): + val = a.get(key) + if isinstance(val, list): + n = max(n, len([x for x in val if str(x or "").strip()])) + targets = a.get("targets") + if isinstance(targets, list): + n = max(n, len([t for t in targets if isinstance(t, dict)])) + if n == 0 and (str(a.get("ne_id") or "").strip() or str(a.get("ume_ne_id") or "").strip()): + return 1 + return int(n) + + +def _truthy_async_flag(raw: Any) -> bool | None: + if raw is None: + return None + if isinstance(raw, bool): + return raw + text = str(raw).strip().lower() + if not text: + return None + if text in {"1", "true", "yes", "on"}: + return True + if text in {"0", "false", "no", "off"}: + return False + return None + + +def should_run_exec_managed_ne_async(args: dict[str, Any] | None) -> bool: + a = dict(args or {}) + flag = _truthy_async_flag(a.get("async")) + if flag is False: + return False + if flag is True: + return True + min_n = async_min_ne_count() + if min_n <= 0: + return False + return count_exec_ne_targets(a) >= int(min_n) + + +def strip_async_flag(args: dict[str, Any] | None) -> dict[str, Any]: + out = dict(args or {}) + out.pop("async", None) + return out + + +def start_ne_exec_job( + *, + tool_name: str, + arguments: dict[str, Any], + runner: Callable[[], dict[str, Any]], + session_id: str = "", + timeout_s: int | None = None, +) -> dict[str, Any]: + """Start background runner; returns immediate ack with job_id.""" + global _INFLIGHT + with _LOCK: + if _INFLIGHT >= _MAX_CONCURRENT: + return { + "ok": False, + "error_code": "ne_exec_job_busy", + "error": "ne_exec_job_busy", + "hint": ( + f"Too many concurrent NE exec jobs (max {_MAX_CONCURRENT}). " + "Poll get_ne_exec_job for running jobs or shrink the batch." + ), + } + _INFLIGHT += 1 + + job_id = uuid.uuid4().hex + timeout = int(timeout_s) if timeout_s is not None else _DEFAULT_TIMEOUT_S + timeout = max(60, min(timeout, 1800)) + now = int(time.time() * 1000) + job = { + "job_id": job_id, + "status": _STATUS_RUNNING, + "tool_name": str(tool_name or ""), + "session_id": str(session_id or ""), + "arguments": dict(arguments or {}), + "ne_count": count_exec_ne_targets(arguments), + "created_at_ms": now, + "updated_at_ms": now, + "timeout_s": timeout, + "result": None, + "error": "", + } + _write_job(job) + + def _worker() -> None: + global _INFLIGHT + started = time.time() + try: + result = runner() + if not isinstance(result, dict): + result = {"ok": False, "error": "invalid_runner_result", "payload_type": type(result).__name__} + status = _STATUS_SUCCEEDED if result.get("ok") is not False else _STATUS_FAILED + if time.time() - started > timeout: + status = _STATUS_TIMEOUT + cur = _read_job(job_id) or job + cur.update( + { + "status": status, + "updated_at_ms": int(time.time() * 1000), + "result": result, + "error": str(result.get("error") or "") if status != _STATUS_SUCCEEDED else "", + "duration_ms": int((time.time() - started) * 1000), + } + ) + _write_job(cur) + except Exception as exc: + cur = _read_job(job_id) or job + cur.update( + { + "status": _STATUS_FAILED, + "updated_at_ms": int(time.time() * 1000), + "error": f"{type(exc).__name__}: {exc}", + "result": {"ok": False, "error": f"{type(exc).__name__}: {exc}"}, + "duration_ms": int((time.time() - started) * 1000), + } + ) + _write_job(cur) + finally: + with _LOCK: + _INFLIGHT = max(0, _INFLIGHT - 1) + + threading.Thread(target=_worker, name=f"ne-exec-{job_id[:8]}", daemon=True).start() + return { + "ok": True, + "async": True, + "status": _STATUS_RUNNING, + "job_id": job_id, + "ne_count": int(job["ne_count"]), + "poll_tool": "get_ne_exec_job", + "hint": ( + f"execManagedNe started in background (job_id={job_id}, ne_count={job['ne_count']}). " + "Tell the user the job_id and end the turn; later call get_ne_exec_job(job_id=...) " + "or ask the user to continue. Do not sleep/busy-wait in this turn." + ), + "example_poll": {"job_id": job_id}, + } + + +def get_ne_exec_job(job_id: str) -> dict[str, Any]: + jid = str(job_id or "").strip() + if not jid: + return {"ok": False, "error_code": "job_id_required", "error": "job_id_required"} + job = _read_job(jid) + if not job: + return {"ok": False, "error_code": "job_not_found", "error": "job_not_found", "job_id": jid} + status = str(job.get("status") or "") + out: dict[str, Any] = { + "ok": True, + "job_id": jid, + "status": status, + "tool_name": job.get("tool_name"), + "ne_count": job.get("ne_count"), + "created_at_ms": job.get("created_at_ms"), + "updated_at_ms": job.get("updated_at_ms"), + "duration_ms": job.get("duration_ms"), + "terminal": status in _TERMINAL, + } + if status in _TERMINAL: + out["result"] = job.get("result") + if job.get("error"): + out["error"] = job.get("error") + else: + out["hint"] = "Still running; poll get_ne_exec_job again later or ask the user to continue." + return out + + +__all__ = [ + "async_min_ne_count", + "count_exec_ne_targets", + "get_ne_exec_job", + "should_run_exec_managed_ne_async", + "start_ne_exec_job", + "strip_async_flag", +] diff --git a/tests/test_ops_short_intent_and_exec_hints.py b/tests/test_ops_short_intent_and_exec_hints.py index 3068a27f..aad5ec92 100644 --- a/tests/test_ops_short_intent_and_exec_hints.py +++ b/tests/test_ops_short_intent_and_exec_hints.py @@ -73,7 +73,7 @@ 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 "FIRST" in hint or "first" in hint.lower() assert "断纤" not in hint diff --git a/tests/test_report_first_and_ne_exec_async.py b/tests/test_report_first_and_ne_exec_async.py new file mode 100644 index 00000000..651a45ad --- /dev/null +++ b/tests/test_report_first_and_ne_exec_async.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import time +from pathlib import Path + +from runtime.chat.ops_report_first_guard import ( + maybe_block_cli_before_report, + report_first_block_payload, +) +from svc.jobs.ne_exec_jobs import ( + count_exec_ne_targets, + get_ne_exec_job, + should_run_exec_managed_ne_async, + start_ne_exec_job, + strip_async_flag, +) + + +def test_report_first_blocks_cli_before_report() -> None: + blocked = maybe_block_cli_before_report( + tool_name="mcp__netx__execManagedNe", + intent="fiber_cut", + store=None, + session_id="", + turn_uuid="", + lang="en", + local_report_ok=False, + ) + assert blocked is not None + assert blocked["error_code"] == "report_first_required" + assert blocked["next_tool"] == "ume_alarm_xlsx_report" + + +def test_report_first_allows_after_local_ok() -> None: + assert ( + maybe_block_cli_before_report( + tool_name="mcp__netx__listCliTargets", + intent="offline", + store=None, + session_id="s", + turn_uuid="t", + local_report_ok=True, + ) + is None + ) + + +def test_report_first_ignores_non_report_intents() -> None: + assert ( + maybe_block_cli_before_report( + tool_name="mcp__netx__execManagedNe", + intent="continue", + store=None, + session_id="", + turn_uuid="", + local_report_ok=False, + ) + is None + ) + + +def test_report_first_payload_zh() -> None: + out = report_first_block_payload(intent="fiber_cut", lang="zh") + assert "ume_alarm_xlsx_report" in out["hint"] + assert out["example"].get("mode") == "fiber_cut" + + +def test_should_async_by_ne_count(monkeypatch) -> None: + monkeypatch.setenv("AIA_EXEC_MANAGED_NE_ASYNC_MIN_NES", "4") + assert not should_run_exec_managed_ne_async({"ume_ne_ids": ["a", "b", "c"], "commands": ["show"]}) + assert should_run_exec_managed_ne_async({"ume_ne_ids": ["a", "b", "c", "d"], "commands": ["show"]}) + assert should_run_exec_managed_ne_async({"async": True, "ume_ne_id": "x", "commands": ["show"]}) + assert not should_run_exec_managed_ne_async( + {"async": False, "ume_ne_ids": ["a", "b", "c", "d", "e"], "commands": ["show"]} + ) + + +def test_count_and_strip_async() -> None: + assert count_exec_ne_targets({"targets": [{"ume_ne_id": "1", "commands": ["a"]}, {"ne_id": "2", "commands": ["b"]}]}) == 2 + assert "async" not in strip_async_flag({"async": True, "commands": ["x"]}) + + +def test_ne_exec_job_roundtrip(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("AIA_NE_EXEC_JOB_DIR", str(tmp_path)) + + def _runner() -> dict: + time.sleep(0.05) + return {"ok": True, "data": {"results": [1]}} + + ack = start_ne_exec_job( + tool_name="mcp__netx__execManagedNe", + arguments={"ume_ne_ids": ["a", "b", "c", "d"], "commands": ["show version"]}, + runner=_runner, + session_id="sess", + ) + assert ack["ok"] is True + assert ack["async"] is True + jid = ack["job_id"] + deadline = time.time() + 2.0 + status = "" + while time.time() < deadline: + polled = get_ne_exec_job(jid) + status = str(polled.get("status") or "") + if status in {"succeeded", "failed", "timeout"}: + break + time.sleep(0.05) + assert status == "succeeded" + done = get_ne_exec_job(jid) + assert done["terminal"] is True + assert done["result"]["ok"] is True