From efa72df3625067538058675fb64a3514ee16b3cd Mon Sep 17 00:00:00 2001 From: oliver Date: Tue, 11 Aug 2026 01:17:39 +0800 Subject: [PATCH] Remove plan mode, skill toolcall path, and inline netx tools. Wave C subtraction: keep skills as prompt-only, route netx via MCP with a shared netx_http client for xlsx/context inject, and delete plan_agent_v2 from the gateway. Co-authored-by: Cursor --- README.md | 2 +- docs/NETX_MCP_INTEGRATION.md | 16 +- docs/plan_agent_v2_shadow.md | 103 --- interfaces/admin/chat_api.py | 13 +- interfaces/admin/skills_api.py | 11 +- interfaces/admin/static/app.js | 7 +- interfaces/admin/static/chat.js | 9 +- runtime/gateway.py | 84 -- runtime/plan_agent_v2/__init__.py | 42 - runtime/plan_agent_v2/adapter.py | 281 ------- runtime/plan_agent_v2/compat.py | 46 -- runtime/plan_agent_v2/gateway_adapter.py | 54 -- runtime/plan_agent_v2/manager.py | 185 ----- runtime/plan_agent_v2/models.py | 49 -- runtime/plan_agent_v2/prompt_injector.py | 120 --- runtime/plan_agent_v2/state_store.py | 60 -- runtime/plan_agent_v2/switch.py | 35 - runtime/plan_agent_v2/tool_policy.py | 52 -- runtime/plan_agent_v2/tool_specs.py | 161 ---- runtime/plan_agent_v2/trace.py | 37 - runtime/tools/catalog.py | 29 - runtime/tools/expert_registry.py | 17 +- .../tools/experts/network_ops/netx_http.py | 104 +++ .../tools/experts/network_ops/netx_tools.py | 728 +----------------- .../network_ops/ume_alarm_xlsx_report.py | 6 +- .../skills_runtime/materialize_skill_tools.py | 63 -- runtime/workspaces/ops/ROLE_SYSTEM.en.md | 1 - runtime/workspaces/ops/ROLE_SYSTEM.md | 1 - .../ops/ops-netx-managed-ne-playbook/SKILL.md | 2 +- .../ops/ops-netx-ume-playbook/SKILL.md | 2 +- tests/test_admin_chat_stream_async_task.py | 8 +- tests/test_netx_expert_registry.py | 18 +- tests/test_netx_managed_ne_tools.py | 49 -- tests/test_netx_protocol_localize.py | 2 +- tests/test_netx_ume_ne_tools.py | 50 -- tests/test_oclaw_gateway_trace.py | 137 +--- tests/test_plan_agent_v2_gateway_dryrun.py | 142 ---- tests/test_plan_agent_v2_shadow.py | 450 ----------- .../test_skill_runtime_metadata_and_tools.py | 23 - tests/test_ume_alarm_xlsx_report.py | 3 +- 40 files changed, 169 insertions(+), 3033 deletions(-) delete mode 100644 docs/plan_agent_v2_shadow.md delete mode 100644 runtime/plan_agent_v2/__init__.py delete mode 100644 runtime/plan_agent_v2/adapter.py delete mode 100644 runtime/plan_agent_v2/compat.py delete mode 100644 runtime/plan_agent_v2/gateway_adapter.py delete mode 100644 runtime/plan_agent_v2/manager.py delete mode 100644 runtime/plan_agent_v2/models.py delete mode 100644 runtime/plan_agent_v2/prompt_injector.py delete mode 100644 runtime/plan_agent_v2/state_store.py delete mode 100644 runtime/plan_agent_v2/switch.py delete mode 100644 runtime/plan_agent_v2/tool_policy.py delete mode 100644 runtime/plan_agent_v2/tool_specs.py delete mode 100644 runtime/plan_agent_v2/trace.py create mode 100644 runtime/tools/experts/network_ops/netx_http.py delete mode 100644 runtime/tools/skills_runtime/materialize_skill_tools.py delete mode 100644 tests/test_netx_managed_ne_tools.py delete mode 100644 tests/test_netx_ume_ne_tools.py delete mode 100644 tests/test_plan_agent_v2_gateway_dryrun.py delete mode 100644 tests/test_plan_agent_v2_shadow.py diff --git a/README.md b/README.md index dbd0ecf9..8b817e8a 100644 --- a/README.md +++ b/README.md @@ -56,7 +56,7 @@ powershell -ExecutionPolicy Bypass -File .\scripts\start_all.ps1 -Background ### 可选：运维专家（network_ops）与 netx -若要在 ops 专家模式下调 **netx** 告警库，需单独启动 netx 服务，并在 Admin 安装 **netx MCP**（`server_id=netx`，env `NETX_API_URL`）。说明见 `docs/NETX_MCP_INTEGRATION.md`。迁移期可选 `OCLAW_NETX_BUILTIN_TOOLS=1` 启用旧 inline 工具。 +若要在 ops 专家模式下调 **netx** 告警库，需单独启动 netx 服务，并在 Admin 安装 **netx MCP**（`server_id=netx`，env `NETX_API_URL`）。说明见 `docs/NETX_MCP_INTEGRATION.md`。 ### 可选：股票分析专家（A股/港股，信号建议） diff --git a/docs/NETX_MCP_INTEGRATION.md b/docs/NETX_MCP_INTEGRATION.md index 9a9bc772..58b8f870 100644 --- a/docs/NETX_MCP_INTEGRATION.md +++ b/docs/NETX_MCP_INTEGRATION.md @@ -4,7 +4,7 @@ Wire **oclaw** (or any MCP host) to the standard **netx HTTP MCP** in `D:/projec netx 侧通用安装/更新说明：`D:/project/chatgpt/netx/docs/MCP.md`。 -Default path: **stdio MCP → netx REST API** (`NETX_API_URL`). Legacy inline HTTP tools in oclaw are opt-in via `OCLAW_NETX_BUILTIN_TOOLS=1`. +Default path: **stdio MCP → netx REST API** (`NETX_API_URL`). oclaw also uses the same REST base for ops context inject and `ume_alarm_xlsx_report` via shared `netx_http`. ## 1) Start netx API @@ -115,14 +115,16 @@ Then run **Health** → **Sync Tools**. In Admin **MCP specialist binding**, include server **`netx`** for the ops workspace/specialist. -## 4) Dual-track: builtin vs MCP +## 4) Shared REST helpers in oclaw -| Setting | Effect | -|---------|--------| -| `OCLAW_NETX_BUILTIN_TOOLS=0` (default) | Only MCP tools (`mcp__netx__*`); no duplicate inline `netx_*` in catalog | -| `OCLAW_NETX_BUILTIN_TOOLS=1` | Registers legacy inline HTTP tools **and** MCP if installed — avoid binding both unless testing migration | +oclaw keeps a small HTTP client (`runtime/tools/experts/network_ops/netx_http.py`) for: -Runtime anchor inject (`OCLAW_OPS_NETX_CONTEXT_INJECT=1`) still works without builtin tools; it only needs netx API reachable at `NETX_API_URL` / `OCLAW_NETX_BASE_URL`. +- ops system-context anchor inject (`OCLAW_OPS_NETX_CONTEXT_INJECT`) +- always-on `ume_alarm_xlsx_report` (Excel export shortcut) + +Interactive alarm/NE tools are **MCP only** (`mcp__netx__*`). Inline `netx_*` expert tools were removed. + +Runtime anchor inject still needs netx API reachable at `NETX_API_URL` / `OCLAW_NETX_BASE_URL`. ## 5) Cursor / Claude Desktop diff --git a/docs/plan_agent_v2_shadow.md b/docs/plan_agent_v2_shadow.md deleted file mode 100644 index 6ab799bb..00000000 --- a/docs/plan_agent_v2_shadow.md +++ /dev/null @@ -1,103 +0,0 @@ -# Plan Agent V2 (Shadow) Design - -## Goal -- Build a complete plan/agent pipeline in shadow mode first. -- Keep legacy runtime path unchanged until final one-shot cutover. -- Support instant rollback via a single feature switch. - -## Scope -- Target only `interaction_mode=expert`. -- `interaction_mode=comprehensive` remains on legacy path. -- Current implementation is dry-run/shadow ready, not wired into gateway production flow. - -## Runtime Components -- Package root: `runtime/plan_agent_v2/` - - `models.py`: state model (`PlanAgentStateV2`) - - `state_store.py`: session state persistence - - `manager.py`: plan lifecycle (`enter/confirm/exit`) - - `tool_policy.py`: plan-mode tool filtering policy - - `prompt_injector.py`: plan-mode and approved-plan prompt injection - - `tool_specs.py`: shadow plan tools (`enter_plan_mode_v2`, `exit_plan_mode_v2`) - - `switch.py`: feature switch and routing predicate - - `adapter.py`: expert-mode plan decision logic - - `gateway_adapter.py`: gateway-side shadow adapter - - `trace.py`: plan events trace helper - - `compat.py`: legacy result-shape compatibility helpers - -## Legacy Compatibility -- Flat module paths are still available and now forward to package modules: - - `runtime/plan_agent_v2_*.py` -> `runtime/plan_agent_v2/*` -- This prevents existing imports from breaking during migration. - -## Session State Contract -- Stored under key: - - `AIA_PLAN_AGENT_V2_STATE:` -- Serialized JSON fields: - - `mode`: `normal|plan` - - `owner_specialist` - - `plan_id` - - `plan_path` - - `plan_content` - - `plan_confirmed` - - `entered_at_ms` - - `updated_at_ms` - -## Feature Switches -- `AIA_EXPERT_PLAN_AGENT_V2_ENABLED` - - default: off - - effect: allow expert path to route to shadow v2 when wired -- `AIA_EXPERT_PLAN_FILE_DIR` - - optional plan file root override -- `AIA_EXPERT_PLAN_CONFIRM_STRATEGY` - - `strict` (default): confirmation in `plan` mode is blocked until user switches to `agent` - - `auto`: confirmation in `plan` mode auto-switches to execution - - `off`: disable confirmation-mode gate (same confirm behavior as `auto`) - -## Admin API Mode Fields -- `GET /admin/api/chat/sessions/{session_id}/mode` - - now returns `confirm_strategy` together with `interaction_mode/specialist/memory_mode/execution_mode`. -- `POST /admin/api/chat/sessions/{session_id}/mode` - - accepts optional `confirm_strategy` (`strict|auto|off`) - - persists per-user and per-session mode settings - - mirrors to runtime key `AIA_EXPERT_PLAN_CONFIRM_STRATEGY` for immediate effect in expert v2 turns - -## Routing Contract (Shadow) -- Predicate: - - `should_route_to_v2(store, interaction_mode, force_flag=False)` -- Rules: - - non-expert mode: always false - - expert + `force_flag=True`: true - - expert + feature on: true - - otherwise: false - -## Adapter Outputs -- `evaluate_for_expert_mode(...)` returns: - - `action`: `enter_plan|stay_plan|run_agent` - - `reply_text` - - `plan_state` - - `system_prompt_override` (set on `run_agent`) - -## Trace Events -- Emitted by `emit_plan_agent_v2_trace(...)`: - - `plan_mode_entered` - - `plan_mode_active` - - `plan_mode_confirmed` - -## Tests -- Shadow core tests: - - `tests/test_plan_agent_v2_shadow.py` -- Gateway dry-run comparison tests: - - `tests/test_plan_agent_v2_gateway_dryrun.py` - -## Cutover Plan (Later, Not Yet Applied) -- Add one gateway branch: - - if `should_route_to_v2(...)` then call `evaluate_gateway_expert_turn_shadow(...)` - - else keep legacy path -- Keep cutover in one commit for easy rollback. - -## Rollback -- Runtime rollback: - - set `AIA_EXPERT_PLAN_AGENT_V2_ENABLED=false` -- Code rollback: - - revert only gateway branch commit; shadow modules can remain dormant. - diff --git a/interfaces/admin/chat_api.py b/interfaces/admin/chat_api.py index 760f785f..36fb5226 100644 --- a/interfaces/admin/chat_api.py +++ b/interfaces/admin/chat_api.py @@ -33,7 +33,6 @@ from svc.files.session_export import export_session_json, export_session_markdow from svc.persistence.sqlite_store import SqliteStore from svc.persistence.assistant_store import get_assistant_store from runtime.gateway import OclawGateway -from runtime.plan_agent_v2.switch import v2_feature_enabled from runtime.types import StandardMessage, normalize_interaction_mode, normalize_requested_specialist from runtime.chat.history_tool_result_compact import compact_tool_results_in_session_history from runtime.chat.persist_terminal_fallback import persist_assistant_text_if_turn_missing @@ -1280,7 +1279,7 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor "execution_mode": s_em, "confirm_strategy": u_cs, "plan_agent_version": u_pav, - "plan_agent_v2_globally_enabled": bool(v2_feature_enabled(store=store)), + "plan_agent_v2_globally_enabled": False, "global_menu": { "interaction_mode": u_im, "specialist": u_sp, @@ -1325,7 +1324,7 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor "execution_mode": s_em, "confirm_strategy": u_cs, "plan_agent_version": u_pav, - "plan_agent_v2_globally_enabled": bool(v2_feature_enabled(store=store)), + "plan_agent_v2_globally_enabled": False, "global_menu": { "interaction_mode": u_im, "specialist": u_sp, @@ -1349,7 +1348,7 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor "specialist": u_sp, "confirm_strategy": u_cs, "plan_agent_version": u_pav, - "plan_agent_v2_globally_enabled": bool(v2_feature_enabled(store=store)), + "plan_agent_v2_globally_enabled": False, } @chat.post("/user-mode") @@ -1377,15 +1376,15 @@ def include_chat_routes(router: APIRouter, *, resolve_auth: Callable[[SqliteStor plan_agent_version=plan_agent_version, ) u_im, u_sp, u_cs, u_pav = _resolve_user_menu_chat_settings(store=store, tenant_id=tenant_id, user_id=user_id) - # Mirror Plan/Agent version to the gateway feature gate (⋯ menu is the control surface). - store.set_setting("AIA_EXPERT_PLAN_AGENT_V2_ENABLED", "1" if str(u_pav or "").strip().lower() == "v2" else "0") + # Plan mode removed; keep setting key cleared for leftover clients. + store.set_setting("AIA_EXPERT_PLAN_AGENT_V2_ENABLED", "0") return { "ok": True, "interaction_mode": u_im, "specialist": u_sp, "confirm_strategy": u_cs, "plan_agent_version": u_pav, - "plan_agent_v2_globally_enabled": bool(v2_feature_enabled(store=store)), + "plan_agent_v2_globally_enabled": False, } @chat.get("/admin/user-stats") diff --git a/interfaces/admin/skills_api.py b/interfaces/admin/skills_api.py index dbf8abb8..8f4ff339 100644 --- a/interfaces/admin/skills_api.py +++ b/interfaces/admin/skills_api.py @@ -98,14 +98,11 @@ def include_skill_routes( ctx = resolve_auth(store, authorization) _require_admin(ctx) raw_prompt = str(store.get_setting("AIA_SKILLS_PROMPT_IN_SYSTEM") or "").strip().lower() - raw_toolcall = str(store.get_setting("AIA_SKILL_TOOLCALL_ENABLED") or "").strip().lower() prompt_in_system = raw_prompt not in {"0", "false", "no", "off"} - toolcall_enabled = raw_toolcall in {"1", "true", "yes", "on"} market_provider = normalize_skill_market_provider_setting(str(store.get_setting(_SKILL_MARKET_PROVIDER_KEY) or "")) return { "ok": True, "prompt_in_system": bool(prompt_in_system), - "toolcall_enabled": bool(toolcall_enabled), "market_provider": market_provider, } @@ -120,14 +117,10 @@ def include_skill_routes( _require_admin(ctx) if "prompt_in_system" in payload: store.set_setting("AIA_SKILLS_PROMPT_IN_SYSTEM", "1" if bool(payload.get("prompt_in_system")) else "0") - if "toolcall_enabled" in payload: - store.set_setting("AIA_SKILL_TOOLCALL_ENABLED", "1" if bool(payload.get("toolcall_enabled")) else "0") if "market_provider" in payload: store.set_setting(_SKILL_MARKET_PROVIDER_KEY, normalize_skill_market_provider_setting(str(payload.get("market_provider") or ""))) raw_prompt = str(store.get_setting("AIA_SKILLS_PROMPT_IN_SYSTEM") or "").strip().lower() - raw_toolcall = str(store.get_setting("AIA_SKILL_TOOLCALL_ENABLED") or "").strip().lower() prompt_in_system = raw_prompt not in {"0", "false", "no", "off"} - toolcall_enabled = raw_toolcall in {"1", "true", "yes", "on"} market_provider = normalize_skill_market_provider_setting(str(store.get_setting(_SKILL_MARKET_PROVIDER_KEY) or "")) _audit( store, @@ -137,14 +130,12 @@ def include_skill_routes( status="ok", detail={ "prompt_in_system": bool(prompt_in_system), - "toolcall_enabled": bool(toolcall_enabled), "market_provider": market_provider, }, ) return { "ok": True, "prompt_in_system": bool(prompt_in_system), - "toolcall_enabled": bool(toolcall_enabled), "market_provider": market_provider, } @@ -817,7 +808,7 @@ def include_skill_routes( raise HTTPException(status_code=500, detail="tool_registry_unavailable") spec = tools.get(name) if spec is None: - # Skill prompt mode: toolcall may be disabled, fallback to direct runtime execution from manifest. + # Skill prompt mode: execute runtime entry from manifest when invoked via Admin/skills APIs. mf = next((m for m in discover_workspace_skill_manifests() if str(m.name or "").strip() == name), None) if mf is None: raise HTTPException(status_code=404, detail="tool_not_found") diff --git a/interfaces/admin/static/app.js b/interfaces/admin/static/app.js index 90afdfb9..157d21a4 100644 --- a/interfaces/admin/static/app.js +++ b/interfaces/admin/static/app.js @@ -8235,7 +8235,6 @@ async function renderSkills() { const marketStatus = el("div", { class: "muted", text: "" }); const skillModeStatus = el("div", { class: "muted", text: "" }); const skillPromptModeCb = el("input", { type: "checkbox" }); - const skillToolcallModeCb = el("input", { type: "checkbox" }); const skillMarketProviderSelect = el("select", { class: "input", style: "min-width:160px;" }, [ el("option", { value: "clawhub", text: "clawhub (ClawHub)" }), el("option", { value: "cocoloop", text: "cocoloop (CocoLoop)" }), @@ -8250,7 +8249,6 @@ async function renderSkills() { try { const r = await apiGet("/admin/api/skills/mode"); skillPromptModeCb.checked = !!r.prompt_in_system; - skillToolcallModeCb.checked = !!r.toolcall_enabled; const mp = String(r.market_provider || "clawhub").trim().toLowerCase(); skillMarketProviderSelect.value = mp === "cocoloop" ? "cocoloop" : "clawhub"; skillModeStatus.textContent = ""; @@ -8263,14 +8261,12 @@ async function renderSkills() { try { const r = await apiPost("/admin/api/skills/mode", { prompt_in_system: !!skillPromptModeCb.checked, - toolcall_enabled: !!skillToolcallModeCb.checked, market_provider: String(skillMarketProviderSelect.value || "clawhub").trim(), }); skillPromptModeCb.checked = !!r.prompt_in_system; - skillToolcallModeCb.checked = !!r.toolcall_enabled; const mp = String(r.market_provider || "clawhub").trim().toLowerCase(); skillMarketProviderSelect.value = mp === "cocoloop" ? "cocoloop" : "clawhub"; - skillModeStatus.textContent = `saved: prompt=${String(!!r.prompt_in_system)} toolcall=${String(!!r.toolcall_enabled)} market=${String(skillMarketProviderSelect.value)}`; + skillModeStatus.textContent = `saved: prompt=${String(!!r.prompt_in_system)} market=${String(skillMarketProviderSelect.value)}`; } catch (e) { skillModeStatus.textContent = `mode: ${String(e && e.message ? e.message : e)}`; } @@ -9610,7 +9606,6 @@ async function renderSkills() { el("div", { class: "card__title", text: t("title.skills") }), el("div", { class: "row", style: "gap:8px;align-items:center;flex-wrap:wrap;margin-bottom:8px;" }, [ el("label", { class: "row", style: "gap:6px;align-items:center;" }, [skillPromptModeCb, el("span", { text: "Prompt mode (inject SKILL.md)" })]), - el("label", { class: "row", style: "gap:6px;align-items:center;" }, [skillToolcallModeCb, el("span", { text: "Toolcall mode (runtime as tools)" })]), el("label", { class: "row", style: "gap:6px;align-items:center;flex-wrap:wrap;" }, [ el("span", { text: "Market (AIA_SKILL_MARKET_PROVIDER)" }), skillMarketProviderSelect, diff --git a/interfaces/admin/static/chat.js b/interfaces/admin/static/chat.js index 318bb268..3a40de06 100644 --- a/interfaces/admin/static/chat.js +++ b/interfaces/admin/static/chat.js @@ -3430,13 +3430,8 @@ async function renderChatUi() { return normalizePlanAgentVersion(currentPlanAgentVersion); }; const refreshExecUi = () => { - const modeVal = String(globalMenuModeValue || MAIN_MODE_VALUE).toLowerCase(); - const expert = modeVal !== MAIN_MODE_VALUE; - const show = - expert && - planAgentV2GloballyEnabled && - normalizePlanAgentVersion(currentPlanAgentVersion) === PLAN_AGENT_V2; - execSelectWrap.style.display = show ? "inline-flex" : "none"; + // Plan mode removed; keep agent execution only. + execSelectWrap.style.display = "none"; }; const modeOptionLabel = (v) => { const key = String(v || "").trim().toLowerCase(); diff --git a/runtime/gateway.py b/runtime/gateway.py index 2334c30b..d1e80ffa 100644 --- a/runtime/gateway.py +++ b/runtime/gateway.py @@ -1013,90 +1013,6 @@ class OclawGateway: system_prompt_override = "" tools_override = None - if interaction_mode == "expert": - from runtime.plan_agent_v2.gateway_adapter import evaluate_gateway_expert_turn_shadow - from runtime.plan_agent_v2.tool_specs import DEFAULT_SESSION_KEY, materialize_plan_mode_v2_tools - - execution_mode = str(base_metadata.get("execution_mode") or "agent").strip().lower() - if execution_mode not in {"agent", "plan"}: - execution_mode = "agent" - try: - self.store.set_setting(DEFAULT_SESSION_KEY, str(msg.session_id or "")) - except Exception: - pass - # Respect store setting AIA_EXPERT_PLAN_AGENT_V2_ENABLED (default off); do not force cutover. - shadow = evaluate_gateway_expert_turn_shadow( - store=self.store, - msg=msg, - lang=lang, - interaction_mode=interaction_mode, - requested_specialist=requested_specialist, - execution_mode=execution_mode, - base_system_prompt=str(getattr(selected_executor, "system_prompt", "") or ""), - force_flag=False, - trace_id=trace_id, - parent_span_id=None, - ) - if shadow.used_v2 and shadow.decision is not None: - action = str(shadow.decision.action or "") - if action in {"enter_plan", "stay_plan"}: - elapsed_ms = int((time.perf_counter() - t0) * 1000) - _trace_local( - event_type="response_sent", - payload={"ok": True, "elapsed_ms": elapsed_ms, "mode": "sync_direct", "plan_action": action}, - started_at=t0, - ) - _flush_trace_rows() - return OclawGatewayResult( - run_id=rid, - reply_text=str(shadow.decision.reply_text or ""), - trace_id=trace_id, - elapsed_ms=elapsed_ms, - mode="sync_direct", - selected_specialist=requested_specialist, - interaction_mode=interaction_mode, - dispatch_reason=f"plan_agent_v2:{action}", - manager_selected_specialist=requested_specialist, - requested_specialist=requested_specialist, - dynamic_agent_used=False, - dynamic_agent_name="", - relay_pointer_count=int(relay_stats.get("relay_pointer_count") or 0), - relay_envelope_present=bool(relay_stats.get("relay_envelope_present")), - relay_envelope_pointer_count=int(relay_stats.get("relay_envelope_pointer_count") or 0), - relay_ttl_turn_count=int(ttl_stats.get("turn") or 0), - relay_ttl_session_count=int(ttl_stats.get("session") or 0), - relay_ttl_keep_count=int(ttl_stats.get("keep") or 0), - turn_uuid="", - ) - if action == "run_agent": - system_prompt_override = str(shadow.decision.system_prompt_override or "") - exec_tools = getattr(selected_executor, "tools", None) - if isinstance(exec_tools, ToolRegistry): - merged = ToolRegistry(exec_tools.list() + materialize_plan_mode_v2_tools(store=self.store)) - tools_override = merged - _trace_local( - event_type="plan_mode_tools_augmented", - payload={"base_count": len(exec_tools.list()), "merged_count": len(merged.list())}, - started_at=t0, - ) - try: - plan_mode = str((shadow.decision.plan_state or {}).get("mode") or "").strip().lower() - except Exception: - plan_mode = "" - if plan_mode == "plan": - from runtime.plan_agent_v2.tool_policy import filter_tools_for_mode - - if isinstance(tools_override, ToolRegistry): - filtered = filter_tools_for_mode(registry=tools_override, mode="plan") - tools_override = ToolRegistry(filtered) - _trace_local( - event_type="plan_mode_tools_filtered", - payload={ - "before_count": len(merged.list()) if isinstance(exec_tools, ToolRegistry) else len(filtered), - "after_count": len(filtered), - }, - started_at=t0, - ) route_mode = "sync_direct" route_msg = StandardMessage( diff --git a/runtime/plan_agent_v2/__init__.py b/runtime/plan_agent_v2/__init__.py deleted file mode 100644 index af0fe6e9..00000000 --- a/runtime/plan_agent_v2/__init__.py +++ /dev/null @@ -1,42 +0,0 @@ -from .adapter import PlanAgentV2Decision, evaluate_for_expert_mode -from .compat import build_shadow_gateway_result, legacy_gateway_result_keys -from .gateway_adapter import GatewayPlanV2AdapterOutput, evaluate_gateway_expert_turn_shadow -from .manager import PlanModeManagerV2 -from .models import PLAN_MODE_NORMAL, PLAN_MODE_PLAN, PlanAgentStateV2 -from .prompt_injector import build_plan_mode_prefix, inject_plan_context -from .state_store import PlanAgentStateStoreV2 -from .switch import should_route_to_v2, v2_feature_enabled -from .tool_policy import filter_tools_for_mode, plan_mode_allowed_tool_names -from .tool_specs import ( - enter_plan_mode_v2_tool, - exit_plan_mode_v2_tool, - is_plan_mode_v2_active, - materialize_plan_mode_v2_tools, -) -from .trace import emit_plan_agent_v2_trace - -__all__ = [ - "PLAN_MODE_NORMAL", - "PLAN_MODE_PLAN", - "PlanAgentStateV2", - "PlanAgentStateStoreV2", - "PlanModeManagerV2", - "build_plan_mode_prefix", - "inject_plan_context", - "filter_tools_for_mode", - "plan_mode_allowed_tool_names", - "enter_plan_mode_v2_tool", - "exit_plan_mode_v2_tool", - "materialize_plan_mode_v2_tools", - "is_plan_mode_v2_active", - "v2_feature_enabled", - "should_route_to_v2", - "PlanAgentV2Decision", - "evaluate_for_expert_mode", - "GatewayPlanV2AdapterOutput", - "evaluate_gateway_expert_turn_shadow", - "emit_plan_agent_v2_trace", - "legacy_gateway_result_keys", - "build_shadow_gateway_result", -] - diff --git a/runtime/plan_agent_v2/adapter.py b/runtime/plan_agent_v2/adapter.py deleted file mode 100644 index 86ef7929..00000000 --- a/runtime/plan_agent_v2/adapter.py +++ /dev/null @@ -1,281 +0,0 @@ -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - -from .manager import PlanModeManagerV2 -from .models import PLAN_MODE_PLAN -from .prompt_injector import build_plan_mode_prefix, inject_plan_context -from .trace import emit_plan_agent_v2_trace - - -@dataclass(frozen=True) -class PlanAgentV2Decision: - action: str # enter_plan | stay_plan | run_agent - reply_text: str - plan_state: dict[str, Any] - system_prompt_override: str = "" - - -def _is_confirm_text(text: str) -> bool: - t = str(text or "").strip().lower() - return t in {"确认", "确认计划", "同意", "通过", "approve", "approved", "confirm", "yes"} - - -def _normalize_user_text(text: str) -> str: - return " ".join(str(text or "").strip().lower().split()) - - -def _is_low_signal_continue(text_norm: str) -> bool: - t = str(text_norm or "").strip().lower() - return t in { - "继续", - "继续啊", - "继续吧", - "可以", - "好的", - "好", - "ok", - "okay", - "go on", - "continue", - } - - -def _confirm_strategy(store: Any) -> str: - try: - raw = str(store.get_setting("AIA_EXPERT_PLAN_CONFIRM_STRATEGY") or "").strip().lower() - except Exception: - raw = "" - if raw in {"auto", "strict", "off"}: - return raw - return "strict" - - -def _last_user_text_norm_from_history(*, store: Any, session_id: str) -> str: - """Most recent persisted user message (current turn is usually not persisted yet).""" - try: - msgs = store.get_messages(session_id=session_id, limit=120) - except Exception: - return "" - for m in reversed(msgs): - if str(getattr(m, "role", "") or "").strip().lower() == "user": - return _normalize_user_text(str(getattr(m, "content", "") or "")) - return "" - - -def _agent_conversation_stall_suffix(*, lang: str) -> str: - is_en = str(lang or "").startswith("en") - if is_en: - return ( - "\n\n[Conversation stall guard — agent mode]\n" - "The user's latest message matches their previous user message in this session.\n" - "- Do not repeat your last assistant reply or restate \"I will now…\" boilerplate.\n" - "- Make substantive progress: execute the next concrete tool step, produce new actionable output, " - "or ask exactly one specific blocking question.\n" - ) - return ( - "\n\n【对话停滞防护 · agent 模式】\n" - "检测到用户本条输入与上一轮用户输入相同（会话已持久化部分）。\n" - "- 禁止复述上一轮助手回复或重复「接下来我将…」式独白。\n" - "- 必须给出实质进展：执行具体工具步骤、写出新的可执行结果，或只提一个关键追问。\n" - ) - - -def evaluate_for_expert_mode( - *, - store: Any, - session_id: str, - lang: str, - requested_specialist: str, - user_text: str, - execution_mode: str = "agent", - base_system_prompt: str, - trace_id: str | None = None, - parent_span_id: str | None = None, -) -> PlanAgentV2Decision: - mgr = PlanModeManagerV2(store=store) - st = mgr.load_state(session_id=session_id) - txt = str(user_text or "").strip() - txt_norm = _normalize_user_text(txt) - exec_mode = str(execution_mode or "").strip().lower() - if exec_mode not in {"agent", "plan"}: - exec_mode = "plan" - confirm_strategy = _confirm_strategy(store) - - if exec_mode == "agent" and st.mode != PLAN_MODE_PLAN: - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="plan_mode_bypassed", - payload={"requested_mode": "agent", "plan_mode_state": str(st.mode or "")}, - ) - last_user_norm = _last_user_text_norm_from_history(store=store, session_id=session_id) - stall = bool(txt_norm and last_user_norm and txt_norm == last_user_norm) - override = "" - if stall: - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="agent_mode_conversation_stall", - payload={"reason": "repeated_user_message"}, - ) - base = str(base_system_prompt or "").strip() - suffix = _agent_conversation_stall_suffix(lang=lang).strip() - override = f"{base}\n\n{suffix}".strip() - return PlanAgentV2Decision( - action="run_agent", - reply_text="", - plan_state=st.to_dict(), - system_prompt_override=override, - ) - - if st.mode != PLAN_MODE_PLAN: - entered = mgr.enter(session_id=session_id, owner_specialist=requested_specialist, force_new_plan=False) - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="plan_mode_entered", - payload={"owner_specialist": entered.owner_specialist, "plan_id": entered.plan_id}, - ) - prefix = build_plan_mode_prefix(state=entered, lang=lang) - return PlanAgentV2Decision( - action="run_agent", - reply_text="", - plan_state=entered.to_dict(), - system_prompt_override=f"{prefix}\n\n{str(base_system_prompt or '').strip()}".strip(), - ) - - st = mgr.refresh_plan_content(session_id=session_id) - st = mgr.update_loop_guard(session_id=session_id, user_text_norm=txt_norm) - if _is_confirm_text(txt): - if exec_mode != "agent" and confirm_strategy == "strict": - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="plan_mode_confirm_blocked", - payload={ - "reason": "execution_mode_not_agent", - "requested_mode": exec_mode, - "confirm_strategy": confirm_strategy, - }, - ) - blocked_reply = ( - "Plan is ready. Please switch to agent mode, then confirm to execute." - if str(lang or "").startswith("en") - else "计划已就绪。请先切换到 agent 模式，再回复“确认”开始执行。" - ) - return PlanAgentV2Decision( - action="stay_plan", - reply_text=blocked_reply, - plan_state=st.to_dict(), - system_prompt_override="", - ) - if exec_mode != "agent" and confirm_strategy == "auto": - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="plan_mode_confirm_auto_switched", - payload={"from_mode": exec_mode, "to_mode": "agent", "confirm_strategy": confirm_strategy}, - ) - confirmed = mgr.confirm(session_id=session_id) - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="plan_mode_confirmed", - payload={ - "plan_id": confirmed.plan_id, - "plan_confirmed": bool(confirmed.plan_confirmed), - "confirm_strategy": confirm_strategy, - }, - ) - next_system = inject_plan_context(base_system=base_system_prompt, state=confirmed, lang=lang) - reply = mgr.build_approved_execution_message(state=confirmed, lang=lang) - return PlanAgentV2Decision( - action="run_agent", - reply_text=reply, - plan_state=confirmed.to_dict(), - system_prompt_override=next_system, - ) - - if _is_low_signal_continue(txt_norm): - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="plan_mode_active", - payload={"plan_id": st.plan_id, "plan_path": st.plan_path, "loop_guard": "low_signal_continue"}, - ) - low_signal_reply = ( - "Plan mode detected a low-information continuation. " - "Please provide concrete plan adjustments, or switch to agent mode and reply 'confirm' to execute." - if str(lang or "").startswith("en") - else "检测到低信息续写（如“继续/可以”）。请给出具体计划修改点，或切换到 agent 模式后回复“确认”直接执行。" - ) - return PlanAgentV2Decision( - action="stay_plan", - reply_text=low_signal_reply, - plan_state=st.to_dict(), - system_prompt_override="", - ) - - if int(st.plan_loop_count or 0) >= 2: - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="plan_mode_active", - payload={"plan_id": st.plan_id, "plan_path": st.plan_path, "loop_guard": "hard_block"}, - ) - anti_loop_reply = ( - "I am in plan mode. I will only output a concise executable plan. " - "If you want me to execute, switch to agent mode and reply 'confirm'." - if str(lang or "").startswith("en") - else "当前为 plan 模式，我只输出可执行计划。若要开始执行，请切换到 agent 模式并回复“确认”。" - ) - return PlanAgentV2Decision( - action="stay_plan", - reply_text=anti_loop_reply, - plan_state=st.to_dict(), - system_prompt_override="", - ) - - prefix = build_plan_mode_prefix(state=st, lang=lang) - anti_loop_suffix = ( - "\n\n[Anti-loop guard]\n" - "- Do not repeat the previous response.\n" - "- If user asks similarly, refine with more concrete steps, checks, and fallback.\n" - "- Keep output as plan only; do not pretend execution is complete." - ) - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_span_id, - event_type="plan_mode_active", - payload={"plan_id": st.plan_id, "plan_path": st.plan_path, "loop_count": int(st.plan_loop_count or 0)}, - ) - return PlanAgentV2Decision( - action="run_agent", - reply_text="", - plan_state=st.to_dict(), - system_prompt_override=f"{prefix}{anti_loop_suffix}\n\n{str(base_system_prompt or '').strip()}".strip(), - ) - - -__all__ = ["PlanAgentV2Decision", "evaluate_for_expert_mode"] - diff --git a/runtime/plan_agent_v2/compat.py b/runtime/plan_agent_v2/compat.py deleted file mode 100644 index 58f1e9b6..00000000 --- a/runtime/plan_agent_v2/compat.py +++ /dev/null @@ -1,46 +0,0 @@ -from __future__ import annotations - -from typing import Any - -from .adapter import PlanAgentV2Decision -from runtime.gateway import OclawGatewayResult - - -def legacy_gateway_result_keys() -> set[str]: - return set(OclawGatewayResult.__dataclass_fields__.keys()) - - -def build_shadow_gateway_result( - *, - decision: PlanAgentV2Decision, - run_id: str, - trace_id: str, - elapsed_ms: int, - requested_specialist: str, -) -> dict[str, Any]: - return { - "run_id": str(run_id), - "reply_text": str(decision.reply_text or ""), - "trace_id": str(trace_id), - "elapsed_ms": int(elapsed_ms), - "mode": "sync_direct", - "task_id": None, - "selected_specialist": str((decision.plan_state or {}).get("owner_specialist") or requested_specialist or "generalist"), - "interaction_mode": "expert", - "dispatch_reason": f"plan_agent_v2:{decision.action}", - "manager_selected_specialist": str((decision.plan_state or {}).get("owner_specialist") or requested_specialist or "generalist"), - "requested_specialist": str(requested_specialist or "generalist"), - "dynamic_agent_used": False, - "dynamic_agent_name": "", - "relay_pointer_count": 0, - "relay_envelope_present": False, - "relay_envelope_pointer_count": 0, - "relay_ttl_turn_count": 0, - "relay_ttl_session_count": 0, - "relay_ttl_keep_count": 0, - "turn_uuid": "", - } - - -__all__ = ["build_shadow_gateway_result", "legacy_gateway_result_keys"] - diff --git a/runtime/plan_agent_v2/gateway_adapter.py b/runtime/plan_agent_v2/gateway_adapter.py deleted file mode 100644 index 231d22e6..00000000 --- a/runtime/plan_agent_v2/gateway_adapter.py +++ /dev/null @@ -1,54 +0,0 @@ -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - -from .adapter import PlanAgentV2Decision, evaluate_for_expert_mode -from .switch import should_route_to_v2 -from runtime.types import StandardMessage - - -@dataclass(frozen=True) -class GatewayPlanV2AdapterOutput: - used_v2: bool - decision: PlanAgentV2Decision | None - - -def evaluate_gateway_expert_turn_shadow( - *, - store: Any, - msg: StandardMessage, - lang: str, - interaction_mode: str, - requested_specialist: str, - execution_mode: str = "", - base_system_prompt: str, - force_flag: bool = False, - trace_id: str | None = None, - parent_span_id: str | None = None, -) -> GatewayPlanV2AdapterOutput: - if not should_route_to_v2(store=store, interaction_mode=interaction_mode, force_flag=force_flag): - return GatewayPlanV2AdapterOutput(used_v2=False, decision=None) - meta = msg.metadata if isinstance(msg.metadata, dict) else {} - if "plan_agent_version" in meta: - if str(meta.get("plan_agent_version") or "").strip().lower() != "v2": - return GatewayPlanV2AdapterOutput(used_v2=False, decision=None) - eff_mode = str(execution_mode or "").strip().lower() - if eff_mode not in {"agent", "plan"}: - eff_mode = "plan" if force_flag else "agent" - dec = evaluate_for_expert_mode( - store=store, - session_id=str(msg.session_id or ""), - lang=lang, - requested_specialist=requested_specialist, - user_text=str(msg.text or ""), - execution_mode=eff_mode, - base_system_prompt=base_system_prompt, - trace_id=trace_id, - parent_span_id=parent_span_id, - ) - return GatewayPlanV2AdapterOutput(used_v2=True, decision=dec) - - -__all__ = ["GatewayPlanV2AdapterOutput", "evaluate_gateway_expert_turn_shadow"] - diff --git a/runtime/plan_agent_v2/manager.py b/runtime/plan_agent_v2/manager.py deleted file mode 100644 index efb0d6cf..00000000 --- a/runtime/plan_agent_v2/manager.py +++ /dev/null @@ -1,185 +0,0 @@ -from __future__ import annotations - -import time -import uuid -from pathlib import Path -from typing import Any - -from .models import PLAN_MODE_NORMAL, PLAN_MODE_PLAN, PlanAgentStateV2 -from .state_store import PlanAgentStateStoreV2 - - -def _default_plan_dir() -> Path: - return Path(__file__).resolve().parents[2] / "data" / "plans" - - -def _resolve_plan_dir(store: Any) -> Path: - raw = str(store.get_setting("AIA_EXPERT_PLAN_FILE_DIR") or "").strip() - if raw: - p = Path(raw) - return p if p.is_absolute() else (Path(__file__).resolve().parents[2] / p) - return _default_plan_dir() - - -def _plan_template() -> str: - return ( - "# Plan\n\n" - "## Goal\n" - "- \n\n" - "## Scope\n" - "- \n\n" - "## Steps\n" - "1. \n" - "2. \n" - "3. \n\n" - "## Risks\n" - "- \n\n" - "## Acceptance\n" - "- \n" - ) - - -class PlanModeManagerV2: - def __init__(self, *, store: Any): - self._store = store - self._state_store = PlanAgentStateStoreV2(store) - - def load_state(self, *, session_id: str) -> PlanAgentStateV2: - return self._state_store.load(session_id=session_id) - - def enter( - self, - *, - session_id: str, - owner_specialist: str, - force_new_plan: bool = False, - ) -> PlanAgentStateV2: - prev = self._state_store.load(session_id=session_id) - if prev.mode == PLAN_MODE_PLAN and not force_new_plan: - return prev - sid = str(session_id or "").strip() - if not sid: - return prev - plan_id = uuid.uuid4().hex - plan_root = _resolve_plan_dir(self._store) / sid - plan_root.mkdir(parents=True, exist_ok=True) - plan_path = plan_root / f"{plan_id}.md" - plan_content = _plan_template() - plan_path.write_text(plan_content, encoding="utf-8") - now_ms = int(time.time() * 1000) - next_state = PlanAgentStateV2( - mode=PLAN_MODE_PLAN, - owner_specialist=str(owner_specialist or "generalist").strip().lower() or "generalist", - plan_id=plan_id, - plan_path=str(plan_path), - plan_content=plan_content, - plan_confirmed=False, - entered_at_ms=now_ms, - updated_at_ms=now_ms, - last_user_text_norm="", - plan_loop_count=0, - ) - return self._state_store.save(session_id=sid, state=next_state) - - def refresh_plan_content(self, *, session_id: str) -> PlanAgentStateV2: - st = self._state_store.load(session_id=session_id) - p = Path(str(st.plan_path or "").strip()) - if not p.exists() or not p.is_file(): - return st - content = p.read_text(encoding="utf-8", errors="replace") - return self._state_store.save( - session_id=session_id, - state=PlanAgentStateV2( - mode=st.mode, - owner_specialist=st.owner_specialist, - plan_id=st.plan_id, - plan_path=st.plan_path, - plan_content=content, - plan_confirmed=st.plan_confirmed, - entered_at_ms=st.entered_at_ms, - updated_at_ms=st.updated_at_ms, - last_user_text_norm=st.last_user_text_norm, - plan_loop_count=st.plan_loop_count, - ), - ) - - def update_loop_guard(self, *, session_id: str, user_text_norm: str) -> PlanAgentStateV2: - st = self._state_store.load(session_id=session_id) - nxt_count = int(st.plan_loop_count or 0) + 1 if user_text_norm and user_text_norm == st.last_user_text_norm else 0 - return self._state_store.save( - session_id=session_id, - state=PlanAgentStateV2( - mode=st.mode, - owner_specialist=st.owner_specialist, - plan_id=st.plan_id, - plan_path=st.plan_path, - plan_content=st.plan_content, - plan_confirmed=st.plan_confirmed, - entered_at_ms=st.entered_at_ms, - updated_at_ms=st.updated_at_ms, - last_user_text_norm=user_text_norm, - plan_loop_count=nxt_count, - ), - ) - - def confirm(self, *, session_id: str) -> PlanAgentStateV2: - st = self.refresh_plan_content(session_id=session_id) - return self._state_store.save( - session_id=session_id, - state=PlanAgentStateV2( - mode=PLAN_MODE_NORMAL, - owner_specialist=st.owner_specialist, - plan_id=st.plan_id, - plan_path=st.plan_path, - plan_content=st.plan_content, - plan_confirmed=True, - entered_at_ms=st.entered_at_ms, - updated_at_ms=st.updated_at_ms, - last_user_text_norm="", - plan_loop_count=0, - ), - ) - - def build_approved_execution_message(self, *, state: PlanAgentStateV2, lang: str) -> str: - is_en = str(lang or "").startswith("en") - plan_path = str(state.plan_path or "").strip() or "unknown" - plan_content = str(state.plan_content or "").strip() - if plan_content: - if is_en: - return ( - "User has approved your plan. You can now start implementation.\n\n" - f"Plan file: {plan_path}\n\n" - f"## Approved Plan\n{plan_content}" - ) - return ( - "用户已确认计划，你可以开始执行实现。\n\n" - f"计划文件：{plan_path}\n\n" - f"## 已确认计划\n{plan_content}" - ) - return ( - f"Plan approved. You can now start implementation. Plan file: {plan_path}" - if is_en - else f"计划已确认，你可以开始执行实现。计划文件：{plan_path}" - ) - - def exit_without_confirm(self, *, session_id: str) -> PlanAgentStateV2: - st = self._state_store.load(session_id=session_id) - return self._state_store.save( - session_id=session_id, - state=PlanAgentStateV2( - mode=PLAN_MODE_NORMAL, - owner_specialist=st.owner_specialist, - plan_id=st.plan_id, - plan_path=st.plan_path, - plan_content=st.plan_content, - plan_confirmed=False, - entered_at_ms=st.entered_at_ms, - updated_at_ms=st.updated_at_ms, - last_user_text_norm="", - plan_loop_count=0, - ), - ) - - -__all__ = ["PlanModeManagerV2"] - diff --git a/runtime/plan_agent_v2/models.py b/runtime/plan_agent_v2/models.py deleted file mode 100644 index 3c17545e..00000000 --- a/runtime/plan_agent_v2/models.py +++ /dev/null @@ -1,49 +0,0 @@ -from __future__ import annotations - -from dataclasses import asdict, dataclass -from typing import Any - - -PLAN_MODE_NORMAL = "normal" -PLAN_MODE_PLAN = "plan" -_VALID_MODES = {PLAN_MODE_NORMAL, PLAN_MODE_PLAN} - - -@dataclass(frozen=True) -class PlanAgentStateV2: - mode: str = PLAN_MODE_NORMAL - owner_specialist: str = "generalist" - plan_id: str = "" - plan_path: str = "" - plan_content: str = "" - plan_confirmed: bool = False - entered_at_ms: int = 0 - updated_at_ms: int = 0 - last_user_text_norm: str = "" - plan_loop_count: int = 0 - - def to_dict(self) -> dict[str, Any]: - return asdict(self) - - @staticmethod - def from_dict(raw: dict[str, Any] | None) -> "PlanAgentStateV2": - obj = raw if isinstance(raw, dict) else {} - mode = str(obj.get("mode") or PLAN_MODE_NORMAL).strip().lower() - if mode not in _VALID_MODES: - mode = PLAN_MODE_NORMAL - return PlanAgentStateV2( - mode=mode, - owner_specialist=str(obj.get("owner_specialist") or "generalist").strip().lower() or "generalist", - plan_id=str(obj.get("plan_id") or "").strip(), - plan_path=str(obj.get("plan_path") or "").strip(), - plan_content=str(obj.get("plan_content") or ""), - plan_confirmed=bool(obj.get("plan_confirmed")), - entered_at_ms=int(obj.get("entered_at_ms") or 0), - updated_at_ms=int(obj.get("updated_at_ms") or 0), - last_user_text_norm=str(obj.get("last_user_text_norm") or "").strip().lower(), - plan_loop_count=int(obj.get("plan_loop_count") or 0), - ) - - -__all__ = ["PLAN_MODE_NORMAL", "PLAN_MODE_PLAN", "PlanAgentStateV2"] - diff --git a/runtime/plan_agent_v2/prompt_injector.py b/runtime/plan_agent_v2/prompt_injector.py deleted file mode 100644 index 62c965ce..00000000 --- a/runtime/plan_agent_v2/prompt_injector.py +++ /dev/null @@ -1,120 +0,0 @@ -from __future__ import annotations - -from pathlib import Path - -from .models import PLAN_MODE_PLAN, PlanAgentStateV2 - - -def _plan_file_info(state: PlanAgentStateV2) -> str: - plan_path = str(state.plan_path or "").strip() - if not plan_path: - return "No plan file path is available yet." - p = Path(plan_path) - if p.exists(): - return ( - f"A plan file already exists at {plan_path}. " - "You can read it and make incremental edits." - ) - return ( - f"No plan file exists yet. You should create your plan at {plan_path}." - ) - - -def build_plan_mode_prefix(*, state: PlanAgentStateV2, lang: str) -> str: - if state.mode != PLAN_MODE_PLAN: - return "" - is_en = str(lang or "").startswith("en") - file_info = _plan_file_info(state) - if is_en: - return ( - "Plan mode is active. The user does not want execution yet.\n" - "You MUST NOT make real project edits, run non-readonly tools, or claim implementation is done.\n" - "## Execution discipline (critical)\n" - "- Do NOT narrate as if you will run scripts, migrate files, or touch disk *in this turn*. " - "Phrases like \"I'll write the script and run it\", \"starting migration now\", or " - "\"let me execute\" mislead the user—refuse that pattern.\n" - "- If the user needs real execution, say explicitly: switch to **agent mode** in the UI, " - "then confirm; you cannot perform execution while plan mode is active.\n" - "- Do NOT repeat the same \"next I will…\" monologue across turns. On vague follow-ups, " - "edit the plan file or ask **one** concrete question—do not restate boilerplate.\n" - "- Do not ask the user to \"approve the plan\" in chat when the product expects mode switch + " - "confirm; instead tell them the handoff: agent mode → confirm to execute.\n\n" - "## Plan File Info\n" - f"{file_info}\n" - "Only the plan file is allowed to be edited while in plan mode.\n\n" - "## Plan Workflow\n" - "### Phase 1: Initial Understanding\n" - "- Understand the request and inspect relevant codepaths.\n" - "- Reuse existing functions/utilities/patterns when possible.\n\n" - "### Phase 2: Design\n" - "- Propose a concrete implementation strategy with trade-offs.\n\n" - "### Phase 3: Review\n" - "- Validate alignment with user intent and constraints.\n" - "- Clarify unresolved requirements only when necessary.\n\n" - "### Phase 4: Final Plan\n" - "- Output sections: Context, Changes, Critical files, Verification.\n" - "- Prefer one recommended approach over listing many alternatives.\n\n" - "### Phase 5: Execution Handoff\n" - "- Ask user to switch to agent mode and confirm before execution.\n" - "- Do not execute while still in plan mode.\n\n" - "## Plan mode tools (lifecycle)\n" - "- Built-in tools `enter_plan_mode_v2` and `exit_plan_mode_v2` mirror cc-mini-style " - "Enter/Exit plan mode: they bind or release plan-mode state for this session.\n" - "- Prefer updating the plan file in place while staying in plan mode; only call " - "`enter_plan_mode_v2` with `force_new_plan: true` when the user explicitly wants a new plan document.\n" - "- When the written plan is ready for review, either keep plan mode and summarize next steps for the user, " - "or call `exit_plan_mode_v2`. Use `confirm: true` only when the user has explicitly approved executing " - "this plan; use `confirm: false` to leave plan mode without marking the plan approved for execution.\n" - "- If the user sends low-content prompts such as 'continue' or 'ok', do not repeat long boilerplate; " - "revise the plan file or ask one concrete clarification." - ) - return ( - "当前处于 plan 模式，用户暂不要求执行。\n" - "你必须不做真实项目改动、不调用非只读工具，也不要声称已经实现完成。\n" - "## 执行纪律（必须遵守）\n" - "- 禁止用「我现在写脚本并执行」「开始迁移/复制」「让我跑一下」等表述，假装本回合会动磁盘或执行命令。\n" - "- 若用户需要真实执行，必须明确说明：请在界面切换到 **agent 模式**，再按产品流程确认；" - "在 plan 模式下你无法代为执行。\n" - "- 禁止多轮重复同一套「接下来我将……」的独白；用户只说「继续/好的」时，应小幅改计划文件或只提一个具体问题，" - "不要复读长模板。\n" - "- 不要用闲聊式「你同意这个计划吗？」代替产品要求的 **切 agent + 确认**；应提示用户按界面切换到 agent 模式后再确认执行。\n\n" - "## 计划文件信息\n" - f"{file_info}\n" - "在 plan 模式下，只允许围绕计划文件进行编辑。\n\n" - "## 计划工作流\n" - "### 阶段1：理解问题\n" - "- 先理解需求并检查相关代码路径。\n" - "- 优先复用现有函数、工具和既有模式。\n\n" - "### 阶段2：方案设计\n" - "- 给出可落地的实现方案，并说明关键取舍。\n\n" - "### 阶段3：对齐复核\n" - "- 核对是否满足用户目标与约束。\n" - "- 仅在必要时提出澄清问题。\n\n" - "### 阶段4：最终计划\n" - "- 输出结构：背景、改动点、关键文件、验证方式。\n" - "- 推荐一个主方案，不要只堆备选项。\n\n" - "### 阶段5：执行切换\n" - "- 明确提示用户先切换到 agent 模式并确认后再执行。\n" - "- 在 plan 模式下不要执行实现。\n\n" - "## 计划模式工具（生命周期）\n" - "- 内置工具 `enter_plan_mode_v2` 与 `exit_plan_mode_v2` 对应 cc-mini 风格的进入/退出计划模式，用于绑定或释放本会话的 plan 状态。\n" - "- 优先在 plan 模式下就地更新计划文件；仅在用户明确要求新开计划文档时，才对 `enter_plan_mode_v2` 使用 `force_new_plan: true`。\n" - "- 计划文档写完后，可继续保持 plan 模式并给用户摘要；也可调用 `exit_plan_mode_v2`。仅在用户已明确同意按该计划执行时使用 " - "`confirm: true`；若只是结束规划、尚未批准执行，使用 `confirm: false`。\n" - "- 若用户输入信息量低的续写（如「继续」「好的」），不要重复大段套话，应小幅修订计划文件或提出一个具体问题。" - ) - - -def inject_plan_context(*, base_system: str, state: PlanAgentStateV2, lang: str, max_chars: int = 3000) -> str: - plan_text = str(state.plan_content or "").strip() - if not plan_text: - return base_system - if len(plan_text) > max_chars: - plan_text = plan_text[:max_chars] + "\n..." - is_en = str(lang or "").startswith("en") - header = "Approved plan context:\n" if is_en else "已确认计划上下文：\n" - return f"{header}{plan_text}\n\n{str(base_system or '').strip()}".strip() - - -__all__ = ["build_plan_mode_prefix", "inject_plan_context"] - diff --git a/runtime/plan_agent_v2/state_store.py b/runtime/plan_agent_v2/state_store.py deleted file mode 100644 index 37eab080..00000000 --- a/runtime/plan_agent_v2/state_store.py +++ /dev/null @@ -1,60 +0,0 @@ -from __future__ import annotations - -import json -import time -from typing import Any - -from .models import PLAN_MODE_NORMAL, PlanAgentStateV2 - - -def _state_key(session_id: str) -> str: - return f"AIA_PLAN_AGENT_V2_STATE:{str(session_id or '').strip()}" - - -class PlanAgentStateStoreV2: - def __init__(self, store: Any): - self._store = store - - def load(self, *, session_id: str) -> PlanAgentStateV2: - sid = str(session_id or "").strip() - if not sid: - return PlanAgentStateV2() - raw = str(self._store.get_setting(_state_key(sid)) or "").strip() - if not raw: - return PlanAgentStateV2() - try: - obj = json.loads(raw) - except Exception: - return PlanAgentStateV2() - return PlanAgentStateV2.from_dict(obj if isinstance(obj, dict) else None) - - def save(self, *, session_id: str, state: PlanAgentStateV2) -> PlanAgentStateV2: - sid = str(session_id or "").strip() - if not sid: - return state - now_ms = int(time.time() * 1000) - next_state = PlanAgentStateV2( - mode=state.mode, - owner_specialist=state.owner_specialist, - plan_id=state.plan_id, - plan_path=state.plan_path, - plan_content=state.plan_content, - plan_confirmed=bool(state.plan_confirmed), - entered_at_ms=int(state.entered_at_ms or 0), - updated_at_ms=now_ms, - last_user_text_norm=str(state.last_user_text_norm or "").strip().lower(), - plan_loop_count=int(state.plan_loop_count or 0), - ) - self._store.set_setting(_state_key(sid), json.dumps(next_state.to_dict(), ensure_ascii=False)) - return next_state - - def reset(self, *, session_id: str) -> PlanAgentStateV2: - sid = str(session_id or "").strip() - if not sid: - return PlanAgentStateV2() - self._store.delete_setting(_state_key(sid)) - return PlanAgentStateV2(mode=PLAN_MODE_NORMAL) - - -__all__ = ["PlanAgentStateStoreV2"] - diff --git a/runtime/plan_agent_v2/switch.py b/runtime/plan_agent_v2/switch.py deleted file mode 100644 index 8070064f..00000000 --- a/runtime/plan_agent_v2/switch.py +++ /dev/null @@ -1,35 +0,0 @@ -from __future__ import annotations - -import os -from typing import Any - - -def _is_truthy(raw: str | None) -> bool: - return str(raw or "").strip().lower() in {"1", "true", "yes", "on"} - - -def v2_feature_enabled(*, store: Any | None = None) -> bool: - # Default off for shadow path safety. - raw = "" - try: - if store is not None: - raw = str(store.get_setting("AIA_EXPERT_PLAN_AGENT_V2_ENABLED") or "").strip() - except Exception: - raw = "" - if not raw: - raw = str(os.getenv("AIA_EXPERT_PLAN_AGENT_V2_ENABLED") or "").strip() - if not raw: - return False - return _is_truthy(raw) - - -def should_route_to_v2(*, store: Any | None, interaction_mode: str, force_flag: bool = False) -> bool: - if str(interaction_mode or "").strip().lower() != "expert": - return False - if force_flag: - return True - return v2_feature_enabled(store=store) - - -__all__ = ["should_route_to_v2", "v2_feature_enabled"] - diff --git a/runtime/plan_agent_v2/tool_policy.py b/runtime/plan_agent_v2/tool_policy.py deleted file mode 100644 index 5f6384ff..00000000 --- a/runtime/plan_agent_v2/tool_policy.py +++ /dev/null @@ -1,52 +0,0 @@ -from __future__ import annotations - -from typing import Iterable - -from .models import PLAN_MODE_PLAN -from runtime.tools.base import ToolRegistry, ToolSpec - - -_DEFAULT_PLAN_ALLOWLIST = frozenset( - { - "read_file", - "search_files", - "glob", - "list_directory", - "list_workspace_tree", - "search_files_context", - "system_time", - # Plan-mode control tools are non-read-only by design, but must stay callable. - "enter_plan_mode_v2", - "exit_plan_mode_v2", - } -) - - -def plan_mode_allowed_tool_names(extra_allowed: Iterable[str] | None = None) -> set[str]: - out = set(_DEFAULT_PLAN_ALLOWLIST) - for x in (extra_allowed or []): - n = str(x or "").strip() - if n: - out.add(n) - return out - - -def filter_tools_for_mode( - *, - registry: ToolRegistry, - mode: str, - extra_allowed: Iterable[str] | None = None, -) -> list[ToolSpec]: - tools = list(registry.list()) - if str(mode or "").strip().lower() != PLAN_MODE_PLAN: - return tools - allow = plan_mode_allowed_tool_names(extra_allowed=extra_allowed) - out: list[ToolSpec] = [] - for t in tools: - if t.name in allow or bool(t.is_read_only()): - out.append(t) - return out - - -__all__ = ["filter_tools_for_mode", "plan_mode_allowed_tool_names"] - diff --git a/runtime/plan_agent_v2/tool_specs.py b/runtime/plan_agent_v2/tool_specs.py deleted file mode 100644 index ec6ee6e5..00000000 --- a/runtime/plan_agent_v2/tool_specs.py +++ /dev/null @@ -1,161 +0,0 @@ -from __future__ import annotations - -from typing import Any - -from .manager import PlanModeManagerV2 -from .models import PLAN_MODE_PLAN -from .trace import emit_plan_agent_v2_trace -from runtime.tools.base import ToolSpec - -DEFAULT_SESSION_KEY = "AIA_PLAN_AGENT_V2_DEFAULT_SESSION_ID" - - -def _emit_tool_trace( - *, - store: Any, - session_id: str, - args: dict[str, Any], - event_type: str, - payload: dict[str, Any] | None = None, -) -> None: - trace_id = str(args.get("trace_id") or "").strip() - if not trace_id: - return - parent_raw = str(args.get("parent_span_id") or "").strip() - emit_plan_agent_v2_trace( - store=store, - session_id=session_id, - trace_id=trace_id, - parent_span_id=parent_raw or None, - event_type=event_type, - payload=payload, - ) - - -def _resolve_session_id(*, store: Any, args: dict[str, Any]) -> str: - sid = str(args.get("session_id") or "").strip() - if sid: - return sid - try: - return str(store.get_setting(DEFAULT_SESSION_KEY) or "").strip() - except Exception: - return "" - - -def enter_plan_mode_v2_tool(*, store: Any) -> ToolSpec: - mgr = PlanModeManagerV2(store=store) - - def _handler(args: dict[str, Any]) -> dict[str, Any]: - session_id = _resolve_session_id(store=store, args=args) - if not session_id: - return {"ok": False, "error_code": "session_id_required", "error": "session_id_required"} - specialist = str(args.get("owner_specialist") or "generalist").strip().lower() or "generalist" - force_new = bool(args.get("force_new_plan")) - st = mgr.enter(session_id=session_id, owner_specialist=specialist, force_new_plan=force_new) - _emit_tool_trace( - store=store, - session_id=session_id, - args=args, - event_type="plan_mode_tool_enter", - payload={ - "tool": "enter_plan_mode_v2", - "owner_specialist": specialist, - "force_new_plan": force_new, - "plan_id": st.plan_id, - }, - ) - return {"ok": True, "state": st.to_dict()} - - return ToolSpec( - name="enter_plan_mode_v2", - description=( - "Enter plan mode for this session (shadow v2), cc-mini-style: binds a dedicated plan file path. " - "Use when the user wants structured planning. Set force_new_plan=true only when starting a brand-new " - "plan document; otherwise reuse the existing plan when possible. Optional trace_id / parent_span_id " - "attach observability to the current trace." - ), - parameters={ - "type": "object", - "properties": { - "session_id": {"type": "string"}, - "owner_specialist": {"type": "string"}, - "force_new_plan": {"type": "boolean", "default": False}, - "trace_id": {"type": "string"}, - "parent_span_id": {"type": "string"}, - }, - "required": [], - "additionalProperties": False, - }, - handler=_handler, - tags=frozenset({"plan_mode", "shadow_v2", "read"}), - read_only=True, - risk_level="low", - ) - - -def exit_plan_mode_v2_tool(*, store: Any) -> ToolSpec: - mgr = PlanModeManagerV2(store=store) - - def _handler(args: dict[str, Any]) -> dict[str, Any]: - session_id = _resolve_session_id(store=store, args=args) - if not session_id: - return {"ok": False, "error_code": "session_id_required", "error": "session_id_required"} - confirm = bool(args.get("confirm")) - st = mgr.confirm(session_id=session_id) if confirm else mgr.exit_without_confirm(session_id=session_id) - _emit_tool_trace( - store=store, - session_id=session_id, - args=args, - event_type="plan_mode_tool_exit", - payload={ - "tool": "exit_plan_mode_v2", - "confirmed": bool(confirm), - "plan_id": st.plan_id, - "plan_confirmed": bool(st.plan_confirmed), - }, - ) - return {"ok": True, "confirmed": bool(confirm), "state": st.to_dict()} - - return ToolSpec( - name="exit_plan_mode_v2", - description=( - "Exit plan mode for this session (shadow v2). confirm=true marks the plan as approved for execution " - "(same intent as the user confirming in agent mode). confirm=false leaves plan mode without approving " - "execution—use when ending planning without a run approval yet. Optional trace_id / parent_span_id for " - "trace correlation." - ), - parameters={ - "type": "object", - "properties": { - "session_id": {"type": "string"}, - "confirm": {"type": "boolean", "default": False}, - "trace_id": {"type": "string"}, - "parent_span_id": {"type": "string"}, - }, - "required": [], - "additionalProperties": False, - }, - handler=_handler, - tags=frozenset({"plan_mode", "shadow_v2", "write"}), - read_only=False, - risk_level="high", - ) - - -def materialize_plan_mode_v2_tools(*, store: Any) -> list[ToolSpec]: - return [enter_plan_mode_v2_tool(store=store), exit_plan_mode_v2_tool(store=store)] - - -def is_plan_mode_v2_active(*, store: Any, session_id: str) -> bool: - mgr = PlanModeManagerV2(store=store) - return mgr.load_state(session_id=session_id).mode == PLAN_MODE_PLAN - - -__all__ = [ - "enter_plan_mode_v2_tool", - "exit_plan_mode_v2_tool", - "materialize_plan_mode_v2_tools", - "is_plan_mode_v2_active", - "DEFAULT_SESSION_KEY", -] - diff --git a/runtime/plan_agent_v2/trace.py b/runtime/plan_agent_v2/trace.py deleted file mode 100644 index 7df2e5e8..00000000 --- a/runtime/plan_agent_v2/trace.py +++ /dev/null @@ -1,37 +0,0 @@ -from __future__ import annotations - -import time -from typing import Any - - -def emit_plan_agent_v2_trace( - *, - store: Any, - session_id: str, - trace_id: str | None, - parent_span_id: str | None, - event_type: str, - payload: dict[str, Any] | None = None, -) -> None: - if not str(trace_id or "").strip(): - return - merged = dict(payload or {}) - merged.setdefault("pipeline", "plan_agent_v2") - merged.setdefault("ts_ms", int(time.time() * 1000)) - try: - from runtime.orchestration.trace import new_span_id - - store.add_trace_event( - session_id=str(session_id or ""), - trace_id=str(trace_id), - span_id=new_span_id(), - parent_span_id=parent_span_id, - event_type=str(event_type or "plan_agent_v2"), - payload=merged, - ) - except Exception: - pass - - -__all__ = ["emit_plan_agent_v2_trace"] - diff --git a/runtime/tools/catalog.py b/runtime/tools/catalog.py index 5238ae37..3b0618cd 100644 --- a/runtime/tools/catalog.py +++ b/runtime/tools/catalog.py @@ -13,7 +13,6 @@ from runtime.tools.base import ToolRegistry, ToolSpec from runtime.tools.expert_registry import materialize_tools_for_expert from runtime.tools.mcp.adapter import materialize_mcp_tools_for_specialist from runtime.tools.public_registry import materialize_public_tools -from runtime.tools.skills_runtime.materialize_skill_tools import materialize_executable_skill_tools from runtime.skills import SkillSpec, materialize_skills_from_tool_specs logger = logging.getLogger(__name__) @@ -48,21 +47,6 @@ def _plugin_tools_enabled(store: Any | None = None) -> bool: return True -def _skill_toolcall_enabled(store: Any | None) -> bool: - try: - raw_env = str(os.getenv("AIA_SKILL_TOOLCALL_ENABLED") or "").strip() - if raw_env: - return _is_truthy(raw_env) - if store is not None: - raw = str(store.get_setting("AIA_SKILL_TOOLCALL_ENABLED") or "").strip() - if raw: - return _is_truthy(raw) - except Exception: - pass - # Default off: skill uses prompt-injection path, not toolcall path. - return False - - def _apply_declared_tool_policy( tools: list[ToolSpec], *, @@ -206,19 +190,6 @@ def materialize_tool_specs( except Exception as exc: logger.warning("expert tool load skipped: %s", exc) - # collect: skill runtime - if _skill_toolcall_enabled(store): - try: - for spec in materialize_executable_skill_tools(store=store): - if not isinstance(spec, ToolSpec): - continue - if _hidden_from_model(str(spec.name or "")): - logger.info("skill runtime tool hidden from model registry: %s", str(spec.name or "")) - continue - collected.append(("skill_runtime", spec)) - except Exception as exc: - logger.warning("skill runtime tool load skipped: %s", exc) - # MCP tools are role-bound and should be materialized before model injection. # Fine-grained penalty/visibility is still applied by wire policy in direct_loop. mcp_enabled = True diff --git a/runtime/tools/expert_registry.py b/runtime/tools/expert_registry.py index 7402a7ca..19aa2732 100644 --- a/runtime/tools/expert_registry.py +++ b/runtime/tools/expert_registry.py @@ -3,7 +3,6 @@ from __future__ import annotations import importlib.util import inspect import logging -import os from collections.abc import Callable from pathlib import Path from typing import Any @@ -20,22 +19,12 @@ _DEPRECATED_TOOL_NAMES: set[str] = { # Deprecated internal tool from legacy src/tools chain. "get_weather", } - - -def netx_builtin_tools_enabled() -> bool: - """When false, skip ``network_ops/netx_tools.py`` factories (use MCP ``mcp__netx__*`` instead).""" - return str(os.getenv("OCLAW_NETX_BUILTIN_TOOLS") or "0").strip().lower() not in { - "0", - "false", - "no", - "off", - } +# Helpers / non-tool modules under experts/ (no *_tool factories expected). +_SKIP_EXPERT_MODULES: frozenset[str] = frozenset({"netx_tools", "netx_http"}) def _skip_expert_module(module_path: Path) -> bool: - if module_path.stem == "netx_tools" and not netx_builtin_tools_enabled(): - return True - return False + return module_path.stem in _SKIP_EXPERT_MODULES def _load_module_from_path(module_path: Path, module_name: str) -> Any | None: diff --git a/runtime/tools/experts/network_ops/netx_http.py b/runtime/tools/experts/network_ops/netx_http.py new file mode 100644 index 00000000..9a2f03e3 --- /dev/null +++ b/runtime/tools/experts/network_ops/netx_http.py @@ -0,0 +1,104 @@ +"""Shared netx REST client for oclaw (context inject + ume_alarm_xlsx_report). + +Aligns base URL with netx-mcp: ``NETX_API_URL`` then ``OCLAW_NETX_BASE_URL``. +""" + +from __future__ import annotations + +import contextvars +import os +from typing import Any + +import httpx + +# Set by ToolExecutor so responses match session language. +NETX_TOOL_LANG: contextvars.ContextVar[str] = contextvars.ContextVar("netx_tool_lang", default="zh") + +_PROTOCOL_KEY_ZH_TO_EN: dict[str, str] = { + "其他": "Other", + "时钟": "Clock", + "OTN/光": "OTN/Optical", + "电源": "Power", +} + + +def _netx_base_url() -> str: + return ( + os.getenv("NETX_API_URL") or os.getenv("OCLAW_NETX_BASE_URL") or "http://127.0.0.1:8890" + ).strip().rstrip("/") + + +def _netx_headers() -> dict[str, str]: + h = {"accept": "application/json"} + tok = (os.getenv("OCLAW_NETX_API_TOKEN") or os.getenv("NETX_API_TOKEN") or "").strip() + if tok: + h["authorization"] = f"Bearer {tok}" + return h + + +def _netx_lang_query_params() -> dict[str, str]: + lang = str(NETX_TOOL_LANG.get() or "zh").strip().lower() + if lang.startswith("en"): + return {"lang": "en"} + return {} + + +def _localize_netx_payload(data: dict[str, Any], *, lang: str) -> dict[str, Any]: + """Map legacy Chinese protocol bucket labels to English for en sessions.""" + if not str(lang or "").strip().lower().startswith("en"): + return data + proto = data.get("protocol_summary") + if isinstance(proto, list): + for row in proto: + if isinstance(row, dict): + k = str(row.get("key") or "") + if k in _PROTOCOL_KEY_ZH_TO_EN: + row["key"] = _PROTOCOL_KEY_ZH_TO_EN[k] + return data + + +def _http_post_json(path: str, body: dict[str, Any], *, timeout: float = 180.0) -> dict[str, Any]: + base = _netx_base_url() + url = f"{base}{path}" + try: + with httpx.Client(timeout=timeout, trust_env=False) as client: + resp = client.post(url, json=body, headers=_netx_headers()) + text = resp.text + if not resp.is_success: + return {"ok": False, "error": f"netx_http_{resp.status_code}", "detail": text[:800]} + data = resp.json() if text else {} + if isinstance(data, dict): + data = _localize_netx_payload(data, lang=str(NETX_TOOL_LANG.get() or "zh")) + return {"ok": True, "data": data if isinstance(data, dict) else {"raw": data}} + except Exception as exc: + return {"ok": False, "error": "netx_request_failed", "detail": str(exc)[:800]} + + +def _http_json(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: + base = _netx_base_url() + url = f"{base}{path}" + merged: dict[str, Any] = dict(_netx_lang_query_params()) + if params: + merged.update(params) + try: + with httpx.Client(timeout=45.0, trust_env=False) as client: + resp = client.request(method, url, params=merged or None, headers=_netx_headers()) + text = resp.text + if not resp.is_success: + return {"ok": False, "error": f"netx_http_{resp.status_code}", "detail": text[:800]} + data = resp.json() if text else {} + if isinstance(data, dict): + data = _localize_netx_payload(data, lang=str(NETX_TOOL_LANG.get() or "zh")) + return {"ok": True, "data": data if isinstance(data, dict) else {"raw": data}} + except Exception as exc: + return {"ok": False, "error": "netx_request_failed", "detail": str(exc)[:800]} + + +__all__ = [ + "NETX_TOOL_LANG", + "_http_json", + "_http_post_json", + "_localize_netx_payload", + "_netx_base_url", + "_netx_headers", +] diff --git a/runtime/tools/experts/network_ops/netx_tools.py b/runtime/tools/experts/network_ops/netx_tools.py index 641cce00..c32442d9 100644 --- a/runtime/tools/experts/network_ops/netx_tools.py +++ b/runtime/tools/experts/network_ops/netx_tools.py @@ -1,145 +1,40 @@ -"""netx ops helpers: runtime context inject + optional legacy inline HTTP tools. +"""netx ops helpers: runtime context inject for ops specialist. -Default: tools are exposed via stdio MCP (``pip install -e packages/netx-mcp`` then ``python -m netx_mcp``). -Set ``OCLAW_NETX_BUILTIN_TOOLS=1`` to re-register inline ``netx_*`` expert tools. +Alarm/NE tools are exposed only via stdio MCP (``mcp__netx__*``). +HTTP helpers live in ``netx_http`` (shared with ``ume_alarm_xlsx_report``). Configure via environment: -- ``OCLAW_NETX_BASE_URL`` (default ``http://127.0.0.1:8890``) — runtime anchor HTTP probe -- ``OCLAW_NETX_API_TOKEN`` (optional) → sent as ``Authorization: Bearer …`` if set. +- ``NETX_API_URL`` or ``OCLAW_NETX_BASE_URL`` (default ``http://127.0.0.1:8890``) +- ``OCLAW_NETX_API_TOKEN`` / ``NETX_API_TOKEN`` (optional Bearer) """ from __future__ import annotations -import contextvars import os import threading import time from typing import Any -import httpx +from runtime.tools.experts.network_ops.netx_http import ( + NETX_TOOL_LANG, + _http_json, + _http_post_json, + _localize_netx_payload, + _netx_base_url, + _netx_headers, +) -from runtime.tools.base import ToolSpec - -# Set by ToolExecutor for netx_* tools so responses match session language. -NETX_TOOL_LANG: contextvars.ContextVar[str] = contextvars.ContextVar("netx_tool_lang", default="zh") - -_PROTOCOL_KEY_ZH_TO_EN: dict[str, str] = { - "其他": "Other", - "时钟": "Clock", - "OTN/光": "OTN/Optical", - "电源": "Power", -} - - -_UME_RAW_GROUP_FIELDS = [ - "alarm_alarm_key", - "alarm_host_name", - "alarm_ne_id", - "alarm_object_name", - "alarm_event_type", - "alarm_native_probable_cause", - "alarm_perceived_severity", - "alarm_is_cleared", - "alarm_time_created", - "alarm_root_cause_alarm_indication", - "ne_ne_id", - "ne_ne_name", - "ne_user_label", - "ne_ip_address", - "ne_ipv6_address", - "ne_ne_type", - "ne_device_level", - "ne_host_name", - "ne_location", - "ne_hardware_version", - "ne_loopback", - "ne_consistent_state", - "ne_interface_version", - "ne_mac", - "ne_admin_status", - "ne_address_type", - "ne_connection_status", - "ne_maintain_status", - "ne_net_mask", - "ne_create_time", - "ne_creator", - "ne_vendor", - "ne_source_type", - "ne_exists", +__all__ = [ + "NETX_TOOL_LANG", + "ops_netx_system_context_extension", + "_http_json", + "_http_post_json", + "_localize_netx_payload", + "_netx_base_url", + "_netx_headers", ] -def _netx_base_url() -> str: - return (os.getenv("OCLAW_NETX_BASE_URL") or "http://127.0.0.1:8890").strip().rstrip("/") - - -def _netx_headers() -> dict[str, str]: - h = {"accept": "application/json"} - tok = (os.getenv("OCLAW_NETX_API_TOKEN") or "").strip() - if tok: - h["authorization"] = f"Bearer {tok}" - return h - - -def _netx_lang_query_params() -> dict[str, str]: - lang = str(NETX_TOOL_LANG.get() or "zh").strip().lower() - if lang.startswith("en"): - return {"lang": "en"} - return {} - - -def _localize_netx_payload(data: dict[str, Any], *, lang: str) -> dict[str, Any]: - """Map legacy Chinese protocol bucket labels to English for en sessions.""" - if not str(lang or "").strip().lower().startswith("en"): - return data - proto = data.get("protocol_summary") - if isinstance(proto, list): - for row in proto: - if isinstance(row, dict): - k = str(row.get("key") or "") - if k in _PROTOCOL_KEY_ZH_TO_EN: - row["key"] = _PROTOCOL_KEY_ZH_TO_EN[k] - return data - - -def _http_post_json(path: str, body: dict[str, Any], *, timeout: float = 180.0) -> dict[str, Any]: - base = _netx_base_url() - url = f"{base}{path}" - try: - with httpx.Client(timeout=timeout, trust_env=False) as client: - resp = client.post(url, json=body, headers=_netx_headers()) - text = resp.text - if not resp.is_success: - return {"ok": False, "error": f"netx_http_{resp.status_code}", "detail": text[:800]} - data = resp.json() if text else {} - if isinstance(data, dict): - data = _localize_netx_payload(data, lang=str(NETX_TOOL_LANG.get() or "zh")) - return {"ok": True, "data": data if isinstance(data, dict) else {"raw": data}} - except Exception as exc: - return {"ok": False, "error": "netx_request_failed", "detail": str(exc)[:800]} - - -def _http_json(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: - base = _netx_base_url() - url = f"{base}{path}" - merged: dict[str, Any] = dict(_netx_lang_query_params()) - if params: - merged.update(params) - try: - # Do not inherit system proxy settings for local netx calls. - with httpx.Client(timeout=45.0, trust_env=False) as client: - resp = client.request(method, url, params=merged or None, headers=_netx_headers()) - text = resp.text - if not resp.is_success: - return {"ok": False, "error": f"netx_http_{resp.status_code}", "detail": text[:800]} - data = resp.json() if text else {} - if isinstance(data, dict): - data = _localize_netx_payload(data, lang=str(NETX_TOOL_LANG.get() or "zh")) - return {"ok": True, "data": data if isinstance(data, dict) else {"raw": data}} - except Exception as exc: - return {"ok": False, "error": "netx_request_failed", "detail": str(exc)[:800]} - - def _resolve_ume_anchor() -> dict[str, Any]: """Resolve current UME alarm anchor from netx sync status.""" r = _http_json("GET", "/v1/ume/sync/status", params={"page": 1, "page_size": 20}) @@ -259,584 +154,3 @@ def ops_netx_system_context_extension(*, lang: str = "zh") -> str: with _OPS_NETX_SYS_CTX_LOCK: _OPS_NETX_SYS_CTX_CACHE[lk] = (store_ts, text) return text - - -def netx_query_ume_alarms_tool() -> ToolSpec: - """Paginated UME current alarms from netx.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - # Guardrail: avoid accidental full scans by endless paging. - page = max(1, int(args.get("page") or 1)) - if page > 2: - page = 2 - page_size = min(500, max(1, int(args.get("page_size") or 50))) - params: dict[str, Any] = {"page": page, "page_size": page_size} - if str(args.get("severity") or "").strip(): - params["severity"] = str(args.get("severity")).strip() - ne_name = str(args.get("ne_name") or "").strip() - keyword = str(args.get("keyword") or "").strip() - if keyword: - params["keyword"] = keyword - elif ne_name: - params["keyword"] = ne_name - if str(args.get("ne_id") or "").strip(): - params["ne_id"] = str(args.get("ne_id")).strip() - return _http_json("GET", "/v1/ume/alarms", params=params) - - return ToolSpec( - name="netx_query_ume_alarms", - description=( - "读取 netx UME 当前告警明细（实时表）；每条含 host_name（网元主展示键，同步时已写入告警表）。" - "支持 severity/ne_id/keyword 与分页。" - ), - parameters={ - "type": "object", - "properties": { - "severity": {"type": "string"}, - "ne_id": {"type": "string"}, - "ne_name": {"type": "string", "description": "兼容参数，会映射到 keyword"}, - "keyword": {"type": "string", "description": "按网元名/标签/IP/对象名等关键字检索"}, - "page": {"type": "integer", "minimum": 1, "default": 1}, - "page_size": {"type": "integer", "minimum": 1, "maximum": 500, "default": 50}, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "alarms", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_aggregate_ume_alarms_tool() -> ToolSpec: - """Aggregate UME current alarms from netx.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - _ = args - return _http_json("GET", "/v1/ume/alarms/aggregate", params=None) - - return ToolSpec( - name="netx_aggregate_ume_alarms", - description="读取 netx UME 当前告警聚合（by_severity/by_ne）。", - parameters={"type": "object", "properties": {}, "required": [], "additionalProperties": False}, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "alarms", "aggregate", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_run_ume_diagnostics_tool() -> ToolSpec: - """Diagnostics summary for UME current alarms.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - _ = args - return _http_json("GET", "/v1/ume/diagnostics", params=None) - - return ToolSpec( - name="netx_run_ume_diagnostics", - description="读取 netx UME 告警诊断摘要（级别分布、Top 告警码、Top 网元、协议归类）。", - parameters={"type": "object", "properties": {}, "required": [], "additionalProperties": False}, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "diagnostics", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_query_ume_ne_inventory_tool() -> ToolSpec: - """Paged UME NE inventory synced in netx (PostgreSQL-backed).""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - page = max(1, int(args.get("page") or 1)) - page_size = min(500, max(1, int(args.get("page_size") or 50))) - params: dict[str, Any] = {"page": page, "page_size": page_size} - if str(args.get("keyword") or "").strip(): - params["keyword"] = str(args.get("keyword")).strip() - return _http_json("GET", "/v1/ume/inventory/ne", params=params) - - return ToolSpec( - name="netx_query_ume_ne_inventory", - description=( - "查询 netx 已同步的 UME 网元清单（读 /v1/ume/inventory/ne，与 netx Web「网元清单」同源）。" - "keyword 可选：匹配 ne_id / ne_name / user_label / ip_address / host_name（主机名）包含。" - "返回 total、page、page_size、items（含 host_name、在线状态、地址、类型等）。" - ), - parameters={ - "type": "object", - "properties": { - "keyword": {"type": "string", "description": "关键字过滤（可选）"}, - "page": {"type": "integer", "minimum": 1, "default": 1}, - "page_size": {"type": "integer", "minimum": 1, "maximum": 500, "default": 50}, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "inventory", "ne", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_get_ume_ne_tool() -> ToolSpec: - """Single UME NE detail by ne_id from netx.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - from urllib.parse import quote - - ne_id = str(args.get("ne_id") or "").strip() - if not ne_id: - return {"ok": False, "error": "ne_id_required", "error_code": "ne_id_required"} - safe = quote(ne_id, safe="") - return _http_json("GET", f"/v1/ume/inventory/ne/{safe}", params=None) - - return ToolSpec( - name="netx_get_ume_ne", - description=( - "按网元 UUID（ne_id）读取 netx 中单条 UME 网元详情（GET /v1/ume/inventory/ne/{ne_id}）。" - "含 vendor、source_type、raw_json 等；404 时上游返回 ume_ne_not_found。" - ), - parameters={ - "type": "object", - "properties": { - "ne_id": {"type": "string", "description": "网元 UUID（与清单中 ne_id 一致）"}, - }, - "required": ["ne_id"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "inventory", "ne", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_query_ume_alarms_raw_tool() -> ToolSpec: - """Power query UME current alarms with full alarm+NE fields.""" - - presets: dict[str, list[str]] = { - "brief": [ - "alarm_alarm_key", - "alarm_host_name", - "alarm_perceived_severity", - "alarm_event_type", - "alarm_last_seen_at", - "ne_host_name", - "ne_user_label", - "ne_ne_name", - "ne_ip_address", - "ne_exists", - ], - "evidence": [ - "alarm_alarm_key", - "alarm_host_name", - "alarm_object_name", - "alarm_event_type", - "alarm_native_probable_cause", - "alarm_perceived_severity", - "alarm_is_cleared", - "alarm_time_created", - "alarm_last_seen_at", - "ne_host_name", - "ne_user_label", - "ne_ne_name", - "ne_ip_address", - "ne_connection_status", - "ne_exists", - ], - "ne_debug": [ - "alarm_alarm_key", - "alarm_ne_id", - "alarm_perceived_severity", - "alarm_last_seen_at", - "ne_user_label", - "ne_ne_name", - "ne_ip_address", - "ne_ipv6_address", - "ne_device_level", - "ne_host_name", - "ne_connection_status", - "ne_admin_status", - "ne_address_type", - "ne_maintain_status", - "ne_exists", - ], - } - - def handler(args: dict[str, Any]) -> dict[str, Any]: - page = max(1, int(args.get("page") or 1)) - page_size = min(500, max(1, int(args.get("page_size") or 50))) - params: dict[str, Any] = {"page": page, "page_size": page_size} - for k in ("severity", "is_cleared", "ne_id", "event_type", "keyword", "time_from", "time_to", "order_by", "order"): - v = str(args.get(k) or "").strip() - if v: - params[k] = v - sf = args.get("select_fields") - fields: list[str] = [] - if isinstance(sf, list): - fields = [str(x).strip() for x in sf if str(x).strip()] - if not fields: - preset = str(args.get("field_preset") or "").strip().lower() - fields = list(presets.get(preset) or []) - if fields: - params["select_fields"] = ",".join(fields) - return _http_json("GET", "/v1/ume/alarms/raw", params=params) - - return ToolSpec( - name="netx_query_ume_alarms_raw", - description=( - "自由查询 netx UME 当前告警原始视图，返回 alarm_* + ne_* 全字段。" - "可按 severity/is_cleared/ne_id/event_type/keyword/time_from/time_to 过滤，支持排序分页。" - "select_fields 可按需指定返回字段，降低输出体积。" - "field_preset 可快速选用默认字段集（brief/evidence/ne_debug）。" - "建议先调用 netx_list_ume_alarm_fields 查看可用字段。" - ), - parameters={ - "type": "object", - "properties": { - "severity": {"type": "string"}, - "is_cleared": {"type": "string"}, - "ne_id": {"type": "string"}, - "event_type": {"type": "string"}, - "keyword": {"type": "string"}, - "time_from": {"type": "string", "description": "ISO8601 时间下界（按 last_seen_at）"}, - "time_to": {"type": "string", "description": "ISO8601 时间上界（按 last_seen_at）"}, - "order_by": {"type": "string", "enum": ["last_seen_at", "time_created", "perceived_severity", "event_type", "ne_id"]}, - "order": {"type": "string", "enum": ["asc", "desc"]}, - "select_fields": { - "type": "array", - "items": {"type": "string"}, - "description": "可选返回字段，如 alarm_alarm_key/ne_user_label/ne_exists", - }, - "field_preset": { - "type": "string", - "enum": ["brief", "evidence", "ne_debug"], - "description": "字段集预设；当未传 select_fields 时生效", - }, - "page": {"type": "integer", "minimum": 1, "default": 1}, - "page_size": {"type": "integer", "minimum": 1, "maximum": 500, "default": 50}, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "alarms", "power_query", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_list_ume_alarm_fields_tool() -> ToolSpec: - """List field names for UME raw alarm query.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - _ = args - return _http_json("GET", "/v1/ume/alarms/fields", params=None) - - return ToolSpec( - name="netx_list_ume_alarm_fields", - description="列出 UME 当前告警 raw 查询可用字段（alarm_fields/ne_fields/order_by_allowed）。", - parameters={"type": "object", "properties": {}, "required": [], "additionalProperties": False}, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "alarms", "schema", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_sql_query_ume_tool() -> ToolSpec: - """Execute read-only SQL on UME tables in netx.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - sql = str(args.get("sql") or "").strip() - limit = max(1, min(2000, int(args.get("limit") or 200))) - statement_timeout_ms = max(0, min(30000, int(args.get("statement_timeout_ms") or 0))) - if not sql: - return {"ok": False, "error": "sql_required"} - base = _netx_base_url() - url = f"{base}/v1/sql/ume_query" - try: - # Do not inherit system proxy settings for local netx calls. - with httpx.Client(timeout=60.0, trust_env=False) as client: - resp = client.post( - url, - json={"sql": sql, "limit": limit, "statement_timeout_ms": statement_timeout_ms}, - headers=_netx_headers(), - ) - text = resp.text - if not resp.is_success: - return {"ok": False, "error": f"netx_http_{resp.status_code}", "detail": text[:800]} - data = resp.json() if text else {} - return {"ok": True, "data": data if isinstance(data, dict) else {"raw": data}} - except Exception as exc: - return {"ok": False, "error": "netx_request_failed", "detail": str(exc)[:800]} - - return ToolSpec( - name="netx_sql_query_ume", - description=( - "在 netx 上执行 UME 只读 SQL（服务端强制 SELECT-only、单语句、限制表为 " - "ume_alarms_current/ume_inventory_ne，并强制 limit）。" - "推荐默认模板：设置 statement_timeout_ms=8000，且 SQL 带时间窗过滤（last_seen_at >= now() - interval '30 minutes'）。" - ), - parameters={ - "type": "object", - "properties": { - "sql": {"type": "string", "description": "只读 SELECT SQL；仅允许 UME 当前告警与网元表"}, - "limit": {"type": "integer", "minimum": 1, "maximum": 2000, "default": 200}, - "statement_timeout_ms": { - "type": "integer", - "minimum": 0, - "maximum": 30000, - "default": 0, - "description": "可选查询超时(ms)；0 表示使用数据库默认超时", - }, - }, - "required": ["sql"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "sql", "power_query", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_aggregate_ume_alarms_raw_tool() -> ToolSpec: - """Dynamic aggregation on UME raw fields.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - params: dict[str, Any] = {} - for k in ( - "group_by", - "group_by2", - "severity", - "is_cleared", - "ne_id", - "event_type", - "keyword", - "time_from", - "time_to", - "limit", - ): - v = args.get(k) - if v is None: - continue - sv = str(v).strip() - if sv: - params[k] = sv - return _http_json("GET", "/v1/ume/alarms/aggregate/raw", params=params) - - return ToolSpec( - name="netx_aggregate_ume_alarms_raw", - description=( - "按 UME raw 字段做动态聚合（group_by/group_by2），支持与 raw 同口径过滤条件。" - "group_by 需使用 alarm_*/ne_* 字段，建议先 netx_list_ume_alarm_fields。" - ), - parameters={ - "type": "object", - "properties": { - "group_by": { - "type": "string", - "enum": _UME_RAW_GROUP_FIELDS, - "description": "主分组字段（网元主键优先 alarm_host_name 或 ne_host_name；勿用 alarm_ne_id/ne_ne_id）", - }, - "group_by2": {"type": "string", "enum": _UME_RAW_GROUP_FIELDS, "description": "可选第二分组字段"}, - "severity": {"type": "string"}, - "is_cleared": {"type": "string"}, - "ne_id": {"type": "string"}, - "event_type": {"type": "string"}, - "keyword": {"type": "string"}, - "time_from": {"type": "string"}, - "time_to": {"type": "string"}, - "limit": {"type": "integer", "minimum": 1, "maximum": 2000, "default": 200}, - }, - "required": ["group_by"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "ume", "alarms", "aggregate", "power_query", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_list_managed_ne_tool() -> ToolSpec: - """List netx managed NEs (inventory for CLI login targets).""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - page = max(1, int(args.get("page") or 1)) - page_size = min(500, max(1, int(args.get("page_size") or 50))) - params: dict[str, Any] = {"page": page, "page_size": page_size} - if str(args.get("keyword") or "").strip(): - params["keyword"] = str(args.get("keyword")).strip() - if str(args.get("vendor") or "").strip(): - params["vendor"] = str(args.get("vendor")).strip() - if str(args.get("connect_status") or "").strip(): - params["connect_status"] = str(args.get("connect_status")).strip() - return _http_json("GET", "/v1/managed-ne", params=params) - - return ToolSpec( - name="netx_list_managed_ne", - description=( - "列出 netx「网元管理」中已纳管的设备（GET /v1/managed-ne）。" - "返回 id、name、ip、vendor、device_type、connect_status 等（不含密码）。" - "必须至少提供 keyword / vendor / connect_status 之一；keyword 可匹配名称/IP/用户名/标签，且至少 2 个字符。" - "登录查配置前先用本工具定位 ne_id，再 netx_get_managed_ne / netx_exec_managed_ne。" - ), - parameters={ - "type": "object", - "properties": { - "keyword": {"type": "string", "description": "名称/IP/用户名/标签包含（可选，若提供至少 2 个字符）"}, - "vendor": {"type": "string", "description": "厂商过滤（可选）"}, - "connect_status": { - "type": "string", - "enum": ["unknown", "testing", "pass", "fail"], - "description": "连通性状态过滤（可选）", - }, - "page": {"type": "integer", "minimum": 1, "default": 1}, - "page_size": {"type": "integer", "minimum": 1, "maximum": 100, "default": 20}, - }, - "required": [], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "managed_ne", "inventory", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def netx_get_managed_ne_tool() -> ToolSpec: - """Single managed NE metadata from netx.""" - - def handler(args: dict[str, Any]) -> dict[str, Any]: - ne_id = str(args.get("ne_id") or "").strip() - if not ne_id: - return {"ok": False, "error": "ne_id_required", "error_code": "ne_id_required"} - return _http_json("GET", f"/v1/managed-ne/{ne_id}", params=None) - - return ToolSpec( - name="netx_get_managed_ne", - description=( - "读取 netx 单条纳管网元详情（GET /v1/managed-ne/{ne_id}）。" - "含 connect_status、connect_message、connect_detail（连通测试日志）、跳板 hop_* 配置摘要。" - ), - parameters={ - "type": "object", - "properties": { - "ne_id": {"type": "string", "description": "网元 UUID（列表 items[].id）"}, - }, - "required": ["ne_id"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "managed_ne", "read_only"}), - risk_level="low", - read_only=True, - ) - - -def _netx_exec_max_commands() -> int: - """Mirror netx NETX_NE_EXEC_MAX_COMMANDS (default 5, hard cap 50).""" - try: - raw = int(os.getenv("NETX_NE_EXEC_MAX_COMMANDS") or 5) - except ValueError: - raw = 5 - return max(1, min(50, raw)) - - -def netx_exec_managed_ne_tool() -> ToolSpec: - """Run read-only CLI on a managed NE via netx.""" - max_cmds = _netx_exec_max_commands() - - def handler(args: dict[str, Any]) -> dict[str, Any]: - ne_id = str(args.get("ne_id") or "").strip() - if not ne_id: - return {"ok": False, "error": "ne_id_required", "error_code": "ne_id_required"} - raw_cmds = args.get("commands") - if not isinstance(raw_cmds, list) or not raw_cmds: - return {"ok": False, "error": "commands_required", "error_code": "commands_required"} - commands = [str(c).strip() for c in raw_cmds if str(c).strip()] - if not commands: - return {"ok": False, "error": "commands_required", "error_code": "commands_required"} - if len(commands) > _netx_exec_max_commands(): - return {"ok": False, "error": "too_many_commands", "error_code": "too_many_commands"} - body: dict[str, Any] = {"ne_id": ne_id, "commands": commands} - rts = args.get("read_timeout_sec") - if rts is not None: - body["read_timeout_sec"] = int(rts) - out = _http_post_json("/v1/managed-ne/exec", body, timeout=300.0) - if not out.get("ok"): - return out - data = out.get("data") or {} - if isinstance(data, dict) and data.get("ok") is False: - return {"ok": False, "data": data, "error": str(data.get("error") or "exec_failed")} - return {"ok": True, "data": data} - - return ToolSpec( - name="netx_exec_managed_ne", - description=( - "经 netx 登录「网元管理」中的设备并执行只读 CLI（POST /v1/managed-ne/exec）。" - "每条命令须以 show / display / ping / ping6 / traceroute / tracert / trace / trace6 开头；" - f"允许白名单管道过滤；禁止分号及改配置类命令；单次最多 {max_cmds} 条" - "（NETX_NE_EXEC_MAX_COMMANDS，硬上限 50）；默认读超时 60s。" - "返回合并输出（含命令回显）；失败时含 error/detail。" - "先 netx_list_managed_ne 解析 ne_id；若 connect_status 非 pass 可先 netx_get_managed_ne 看 connect_detail。" - ), - parameters={ - "type": "object", - "properties": { - "ne_id": {"type": "string", "description": "纳管网元 UUID"}, - "commands": { - "type": "array", - "items": {"type": "string"}, - "minItems": 1, - "maxItems": max_cmds, - "description": "只读 CLI 列表，如 show version、display interface brief", - }, - "read_timeout_sec": { - "type": "integer", - "minimum": 10, - "maximum": 120, - "description": "单条命令 Netmiko 读超时（秒），默认 60", - }, - }, - "required": ["ne_id", "commands"], - "additionalProperties": False, - }, - handler=handler, - tags=frozenset({"netx", "ops", "managed_ne", "cli", "exec"}), - risk_level="medium", - read_only=False, - ) - - -__all__: list[str] = [] - -_LEGACY_EXPORTS = [ - "netx_query_ume_alarms_tool", - "netx_aggregate_ume_alarms_tool", - "netx_run_ume_diagnostics_tool", - "netx_query_ume_ne_inventory_tool", - "netx_get_ume_ne_tool", - "netx_query_ume_alarms_raw_tool", - "netx_aggregate_ume_alarms_raw_tool", - "netx_list_ume_alarm_fields_tool", - "netx_sql_query_ume_tool", - "netx_list_managed_ne_tool", - "netx_get_managed_ne_tool", - "netx_exec_managed_ne_tool", -] - - -def _apply_legacy_exports() -> None: - import os as _os - - if str(_os.getenv("OCLAW_NETX_BUILTIN_TOOLS") or "0").strip().lower() in {"0", "false", "no", "off"}: - return - globals()["__all__"] = list(_LEGACY_EXPORTS) - - -_apply_legacy_exports() diff --git a/runtime/tools/experts/network_ops/ume_alarm_xlsx_report.py b/runtime/tools/experts/network_ops/ume_alarm_xlsx_report.py index b8794940..896bd659 100644 --- a/runtime/tools/experts/network_ops/ume_alarm_xlsx_report.py +++ b/runtime/tools/experts/network_ops/ume_alarm_xlsx_report.py @@ -1,7 +1,7 @@ """Always-on ops tool: query UME alarms → xlsx → optional channel deliverable in one call. -Independent of ``OCLAW_NETX_BUILTIN_TOOLS`` (MCP remains the primary query path; this -collapses the WhatsApp “export Excel” loop that previously needed 3–12 tool calls). +Uses shared ``netx_http`` (same REST base as MCP). MCP remains the primary interactive +query path; this collapses the WhatsApp “export Excel” loop into one tool call. """ from __future__ import annotations @@ -9,7 +9,7 @@ from __future__ import annotations from typing import Any from runtime.tools.base import ToolSpec -from runtime.tools.experts.network_ops import netx_tools as nt +from runtime.tools.experts.network_ops import netx_http as nt from runtime.tools.public.write_xlsx_tool import write_xlsx_tool _LIST_FIELDS = [ diff --git a/runtime/tools/skills_runtime/materialize_skill_tools.py b/runtime/tools/skills_runtime/materialize_skill_tools.py deleted file mode 100644 index da82ae9d..00000000 --- a/runtime/tools/skills_runtime/materialize_skill_tools.py +++ /dev/null @@ -1,63 +0,0 @@ -from __future__ import annotations - -from typing import Any - -from runtime.skills import discover_workspace_skill_manifests -from runtime.tools.base import ToolSpec - - -def materialize_executable_skill_tools(*, store: Any | None = None) -> list[ToolSpec]: - """Convert installed skill manifests with runtime into ToolSpec. - - Tool name equals skill name so the model can call it directly. - """ - _ = store - out: list[ToolSpec] = [] - for m in discover_workspace_skill_manifests(): - rt = dict(m.runtime or {}) if isinstance(m.runtime, dict) else {} - if not rt: - continue - tp = str(rt.get("type") or "").strip().lower() - entry = str(rt.get("entry") or "").strip() - if not tp or not entry: - continue - schema = rt.get("schema") if isinstance(rt.get("schema"), dict) else {"type": "object", "additionalProperties": True} - name = str(m.name or "").strip() - if not name: - continue - - def _handler(args: dict[str, Any], *, _manifest=m, _rt=rt) -> dict[str, Any]: - from runtime.tools.skills_runtime.subprocess_exec import run_skill_runtime_entry - - source_meta = {} - if isinstance(getattr(_manifest, "metadata_oclaw", None), dict): - source_raw = _manifest.metadata_oclaw.get("source") - if isinstance(source_raw, dict): - source_meta = { - "provider": str(source_raw.get("provider") or ""), - "version": str(source_raw.get("version") or ""), - "kind": str(source_raw.get("kind") or ""), - } - return run_skill_runtime_entry( - skill_name=str(_manifest.name or ""), - skill_dir=str(_manifest.skill_dir or ""), - runtime={**dict(_rt), "__source_meta": source_meta}, - args=dict(args or {}), - ) - - out.append( - ToolSpec( - name=name, - description=str(m.description or ""), - parameters=dict(schema), - handler=_handler, - tags=frozenset({"skill", "oclaw", "runtime"}), - risk_level="high", - timeout_s=60.0, - ) - ) - return out - - -__all__ = ["materialize_executable_skill_tools"] - diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.en.md b/runtime/workspaces/ops/ROLE_SYSTEM.en.md index 52ff1e70..0ecc3328 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.en.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.en.md @@ -97,6 +97,5 @@ Each turn may append a **UME alarm runtime anchor** at the end of system context ## netx managed NE (device CLI) - **MCP**: `mcp__netx__listManagedNe` / `mcp__netx__getManagedNe` / `mcp__netx__execManagedNe`. -- **Legacy builtin** (`OCLAW_NETX_BUILTIN_TOOLS=1`): `netx_list_managed_ne`, `netx_get_managed_ne`, `netx_exec_managed_ne`. netx API: MCP env `NETX_API_URL` (recommended); anchor probe also uses `OCLAW_NETX_BASE_URL`. Disable anchor inject: `OCLAW_OPS_NETX_CONTEXT_INJECT=0`. diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.md b/runtime/workspaces/ops/ROLE_SYSTEM.md index 6e487a6a..6c08ed12 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.md @@ -88,6 +88,5 @@ ## netx 纳管网元（登录设备查 CLI） - **MCP**：`mcp__netx__listManagedNe` / `mcp__netx__getManagedNe` / `mcp__netx__execManagedNe`。 -- **旧内置**（`OCLAW_NETX_BUILTIN_TOOLS=1`）：`netx_list_managed_ne`、`netx_get_managed_ne`、`netx_exec_managed_ne`。 netx 服务地址：MCP env `NETX_API_URL`（推荐）；锚点探测亦用 `OCLAW_NETX_BASE_URL`。关闭自动锚点：`OCLAW_OPS_NETX_CONTEXT_INJECT=0`。 diff --git a/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md index 7023abda..bd33a21a 100644 --- a/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md +++ b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md @@ -13,7 +13,7 @@ description: 面向 ops 专家的 netx 纳管网元（网元管理）作业手 ## 工具选择顺序 -优先 **MCP**（`mcp__netx__*`）。legacy：`netx_list_managed_ne` 等（`OCLAW_NETX_BUILTIN_TOOLS=1`）。 +使用 **MCP**（`mcp__netx__*`）。旧 inline `netx_*` 工具已移除。 1. **定位设备** - `mcp__netx__listManagedNe`：`keyword`、`connect_status=pass`（**首选定位**） diff --git a/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md b/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md index 7694b03f..b9f5b7d9 100644 --- a/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md +++ b/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md @@ -27,7 +27,7 @@ description: 面向 ops 专家的 netx UME 运维作业手册。覆盖告警查 | 拓扑路径 | `findTopologyPaths` | | 纳管/CLI（另册） | `listManagedNe` / `getManagedNe` / `execManagedNe` / `listCliTargets` | -旧名 `netx_query_ume_alarms` 等仅在 `OCLAW_NETX_BUILTIN_TOOLS=1` 时可用；**优先 MCP**。 +使用 **MCP** 名称（`mcp__netx__*`）；勿使用已移除的旧 inline `netx_*` 工具名。 ## 工具选择顺序 diff --git a/tests/test_admin_chat_stream_async_task.py b/tests/test_admin_chat_stream_async_task.py index 535434ff..4c6eb080 100644 --- a/tests/test_admin_chat_stream_async_task.py +++ b/tests/test_admin_chat_stream_async_task.py @@ -172,8 +172,8 @@ class AdminChatStreamAsyncTaskTests(unittest.TestCase): payload = json.loads(str(task.payload or "{}")) self.assertEqual(str(payload.get("selected_specialist") or ""), "generalist") - def test_user_mode_plan_version_sets_v2_feature_flag_in_store(self) -> None: - """POST /user-mode mirrors plan_agent_version to AIA_EXPERT_PLAN_AGENT_V2_ENABLED (v2→1, v1→0).""" + def test_user_mode_always_clears_plan_agent_v2_flag(self) -> None: + """Plan mode removed: POST /user-mode always keeps AIA_EXPERT_PLAN_AGENT_V2_ENABLED=0.""" token = self._login() headers = { "authorization": f"Bearer {token}", @@ -193,8 +193,8 @@ class AdminChatStreamAsyncTaskTests(unittest.TestCase): self.assertEqual(r2.status_code, 200) body2 = r2.json() self.assertTrue(body2.get("ok"), body2) - self.assertTrue(body2.get("plan_agent_v2_globally_enabled"), body2) - self.assertEqual(str(self.store.get_setting("AIA_EXPERT_PLAN_AGENT_V2_ENABLED") or "").strip(), "1") + self.assertFalse(body2.get("plan_agent_v2_globally_enabled"), body2) + self.assertEqual(str(self.store.get_setting("AIA_EXPERT_PLAN_AGENT_V2_ENABLED") or "").strip(), "0") r1 = self.client.post( "/admin/api/chat/user-mode", diff --git a/tests/test_netx_expert_registry.py b/tests/test_netx_expert_registry.py index 09484eaa..dbf95d00 100644 --- a/tests/test_netx_expert_registry.py +++ b/tests/test_netx_expert_registry.py @@ -1,9 +1,7 @@ -"""Expert registry: netx builtin tools gated by OCLAW_NETX_BUILTIN_TOOLS.""" +"""Expert registry: inline netx_* tools removed; MCP only.""" from __future__ import annotations -import os - import pytest from runtime.tools import expert_registry @@ -18,18 +16,10 @@ def _clear_expert_cache(): expert_registry._CACHED_SPECS_BY_EXPERT = None -def test_netx_tools_skipped_when_builtin_disabled(monkeypatch): - monkeypatch.delenv("OCLAW_NETX_BUILTIN_TOOLS", raising=False) - factories = expert_registry.discover_expert_tool_factories() - network_ops = factories.get("network_ops") or [] - names = {f().name for f in network_ops} - assert not any(n.startswith("netx_") for n in names) - - -def test_netx_tools_registered_when_builtin_enabled(monkeypatch): +def test_netx_inline_tools_never_registered(monkeypatch): monkeypatch.setenv("OCLAW_NETX_BUILTIN_TOOLS", "1") factories = expert_registry.discover_expert_tool_factories() network_ops = factories.get("network_ops") or [] names = {f().name for f in network_ops} - assert "netx_query_ume_alarms" in names - assert "netx_exec_managed_ne" in names + assert not any(n.startswith("netx_") for n in names) + assert "ume_alarm_xlsx_report" in names diff --git a/tests/test_netx_managed_ne_tools.py b/tests/test_netx_managed_ne_tools.py deleted file mode 100644 index cb09ed57..00000000 --- a/tests/test_netx_managed_ne_tools.py +++ /dev/null @@ -1,49 +0,0 @@ -from __future__ import annotations - -from typing import Any - -import pytest - - -def test_netx_list_managed_ne_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None: - import runtime.tools.experts.network_ops.netx_tools as nt - - calls: list[tuple[str, str, dict[str, Any] | None]] = [] - - def fake(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: - calls.append((method, path, params)) - return {"ok": True, "data": {"total": 0, "items": []}} - - monkeypatch.setattr(nt, "_http_json", fake) - spec = nt.netx_list_managed_ne_tool() - out = spec.handler({"keyword": "192.168", "connect_status": "pass", "page": 1, "page_size": 20}) - assert out.get("ok") is True - assert calls[0] == ("GET", "/v1/managed-ne", {"page": 1, "page_size": 20, "keyword": "192.168", "connect_status": "pass"}) - - -def test_netx_exec_managed_ne_posts_body(monkeypatch: pytest.MonkeyPatch) -> None: - import runtime.tools.experts.network_ops.netx_tools as nt - - bodies: list[dict[str, Any]] = [] - - def fake_post(path: str, body: dict[str, Any], *, timeout: float = 180.0) -> dict[str, Any]: - bodies.append(body) - return {"ok": True, "data": {"ok": True, "output": "R2#show version\n..."}} - - monkeypatch.setattr(nt, "_http_post_json", fake_post) - spec = nt.netx_exec_managed_ne_tool() - out = spec.handler({"ne_id": "abc", "commands": ["show version"], "read_timeout_sec": 90}) - assert out.get("ok") is True - assert bodies[0]["ne_id"] == "abc" - assert bodies[0]["commands"] == ["show version"] - assert bodies[0]["read_timeout_sec"] == 90 - - -def test_netx_exec_requires_commands(monkeypatch: pytest.MonkeyPatch) -> None: - import runtime.tools.experts.network_ops.netx_tools as nt - - monkeypatch.setattr(nt, "_http_post_json", lambda *a, **k: {"ok": True, "data": {}}) - spec = nt.netx_exec_managed_ne_tool() - out = spec.handler({"ne_id": "abc"}) - assert out.get("ok") is False - assert out.get("error_code") == "commands_required" diff --git a/tests/test_netx_protocol_localize.py b/tests/test_netx_protocol_localize.py index b45b497e..73b33689 100644 --- a/tests/test_netx_protocol_localize.py +++ b/tests/test_netx_protocol_localize.py @@ -1,6 +1,6 @@ import unittest -from runtime.tools.experts.network_ops import netx_tools as nt +from runtime.tools.experts.network_ops import netx_http as nt class NetxProtocolLocalizeTests(unittest.TestCase): diff --git a/tests/test_netx_ume_ne_tools.py b/tests/test_netx_ume_ne_tools.py deleted file mode 100644 index 20f1e770..00000000 --- a/tests/test_netx_ume_ne_tools.py +++ /dev/null @@ -1,50 +0,0 @@ -from __future__ import annotations - -from typing import Any - -import pytest - - -def test_netx_query_ume_ne_inventory_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None: - import runtime.tools.experts.network_ops.netx_tools as nt - - calls: list[tuple[str, str, dict[str, Any] | None]] = [] - - def fake(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: - calls.append((method, path, params)) - return {"ok": True, "data": {"total": 0, "page": 1, "page_size": 50, "items": []}} - - monkeypatch.setattr(nt, "_http_json", fake) - spec = nt.netx_query_ume_ne_inventory_tool() - out = spec.handler({"keyword": "10.0.0", "page": 2, "page_size": 100}) - assert out.get("ok") is True - assert len(calls) == 1 - assert calls[0][0] == "GET" - assert calls[0][1] == "/v1/ume/inventory/ne" - assert calls[0][2] == {"page": 2, "page_size": 100, "keyword": "10.0.0"} - - -def test_netx_get_ume_ne_requires_id(monkeypatch: pytest.MonkeyPatch) -> None: - import runtime.tools.experts.network_ops.netx_tools as nt - - monkeypatch.setattr(nt, "_http_json", lambda *a, **k: {"ok": True, "data": {}}) - spec = nt.netx_get_ume_ne_tool() - out = spec.handler({}) - assert out.get("ok") is False - assert out.get("error_code") == "ne_id_required" - - -def test_netx_get_ume_ne_quotes_path(monkeypatch: pytest.MonkeyPatch) -> None: - import runtime.tools.experts.network_ops.netx_tools as nt - - paths: list[str] = [] - - def fake(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: - paths.append(path) - return {"ok": True, "data": {"ne_id": "x"}} - - monkeypatch.setattr(nt, "_http_json", fake) - spec = nt.netx_get_ume_ne_tool() - nid = "550e8400-e29b-41d4-a716-446655440000" - spec.handler({"ne_id": nid}) - assert paths == [f"/v1/ume/inventory/ne/{nid}"] diff --git a/tests/test_oclaw_gateway_trace.py b/tests/test_oclaw_gateway_trace.py index 6095b9c3..9a3658ec 100644 --- a/tests/test_oclaw_gateway_trace.py +++ b/tests/test_oclaw_gateway_trace.py @@ -218,11 +218,9 @@ def test_gateway_expert_mode_uses_requested_specialist() -> None: assert chosen.get("sid") == "ops" -def test_gateway_expert_plan_execution_mode_runs_v2_with_plan_prompt(monkeypatch: pytest.MonkeyPatch) -> None: +def test_gateway_expert_plan_execution_mode_ignored_without_plan_agent(monkeypatch: pytest.MonkeyPatch) -> None: class Store: - def get_setting(self, k: str) -> str: - if str(k or "") == "AIA_EXPERT_PLAN_AGENT_V2_ENABLED": - return "1" + def get_setting(self, _k: str) -> str: return "" def set_setting(self, _k: str, _v: str) -> None: @@ -234,9 +232,6 @@ def test_gateway_expert_plan_execution_mode_runs_v2_with_plan_prompt(monkeypatch def add_trace_events_batch(self, _rows: list[dict[str, object]]) -> None: return None - def set_setting(self, _k: str, _v: str) -> None: - return None - class _Exec: model = object() tools = object() @@ -247,7 +242,7 @@ def test_gateway_expert_plan_execution_mode_runs_v2_with_plan_prompt(monkeypatch def _run_agent_core_ok(**kwargs: object) -> object: data = kwargs.get("data") captured["system_prompt"] = str(getattr(data, "system_prompt", "") or "") - return SimpleNamespace(outcome=SimpleNamespace(final_text="plan_llm_reply", turn_uuid="turn-1")) + return SimpleNamespace(outcome=SimpleNamespace(final_text="agent_reply", turn_uuid="turn-1")) monkeypatch.setattr("runtime.gateway.run_agent_core", _run_agent_core_ok) @@ -265,130 +260,10 @@ def test_gateway_expert_plan_execution_mode_runs_v2_with_plan_prompt(monkeypatch out = gw.handle_turn(msg=msg, lang="zh", executor=_Exec()) assert out.interaction_mode == "expert" assert out.dispatch_reason == "expert_direct" - assert str(out.reply_text or "") == "plan_llm_reply" + assert str(out.reply_text or "") == "agent_reply" prompt_text = str(captured.get("system_prompt") or "") - assert ("plan 模式" in prompt_text) or ("Plan mode is active" in prompt_text) - assert ("计划工作流" in prompt_text) or ("Plan Workflow" in prompt_text) - - -def test_gateway_expert_plan_mode_filters_non_readonly_tools(monkeypatch: pytest.MonkeyPatch) -> None: - class Store: - def get_setting(self, k: str) -> str: - if str(k or "") == "AIA_EXPERT_PLAN_AGENT_V2_ENABLED": - return "1" - return "" - - def set_setting(self, _k: str, _v: str) -> None: - return None - - def add_trace_event(self, **_kwargs: object) -> None: - return None - - def add_trace_events_batch(self, _rows: list[dict[str, object]]) -> None: - return None - - def _mk_tool(name: str, read_only: bool) -> ToolSpec: - return ToolSpec( - name=name, - description=name, - parameters={"type": "object", "properties": {}, "additionalProperties": True}, - handler=lambda args: {"ok": True, "args": args}, - read_only=read_only, - ) - - class _Exec: - model = object() - system_prompt = "base-system" - tools = ToolRegistry([_mk_tool("read_file", True), _mk_tool("edit_file", False)]) - - captured: dict[str, object] = {} - - def _run_agent_core_ok(**kwargs: object) -> object: - data = kwargs.get("data") - tools = getattr(data, "tools", None) - captured["tool_names"] = [t.name for t in tools.list()] if hasattr(tools, "list") else [] - return SimpleNamespace(outcome=SimpleNamespace(final_text="ok", turn_uuid="turn-1")) - - monkeypatch.setattr("runtime.gateway.run_agent_core", _run_agent_core_ok) - - gw = OclawGateway(store=Store()) - msg = StandardMessage( - session_id="sid-plan-tools", - tenant_id="t1", - user_id="u1", - role="user", - channel="admin_chat", - text="先给我一个执行计划", - attachments=[], - metadata={"interaction_mode": "expert", "selected_specialist": "generalist", "execution_mode": "plan"}, - ) - out = gw.handle_turn(msg=msg, lang="zh", executor=_Exec()) - assert str(out.reply_text or "") == "ok" - names = list(captured.get("tool_names") or []) - assert "read_file" in names - assert "edit_file" not in names - - -def test_gateway_expert_agent_mode_injects_plan_control_tools(monkeypatch: pytest.MonkeyPatch) -> None: - class Store: - def __init__(self) -> None: - self.kv: dict[str, str] = {} - - def get_setting(self, k: str) -> str: - return str(self.kv.get(k) or "") - - def set_setting(self, k: str, v: str) -> None: - self.kv[k] = str(v or "") - - def add_trace_event(self, **_kwargs: object) -> None: - return None - - def add_trace_events_batch(self, _rows: list[dict[str, object]]) -> None: - return None - - def _mk_tool(name: str, read_only: bool) -> ToolSpec: - return ToolSpec( - name=name, - description=name, - parameters={"type": "object", "properties": {}, "additionalProperties": True}, - handler=lambda args: {"ok": True, "args": args}, - read_only=read_only, - ) - - class _Exec: - model = object() - system_prompt = "base-system" - tools = ToolRegistry([_mk_tool("read_file", True)]) - - captured: dict[str, object] = {} - - def _run_agent_core_ok(**kwargs: object) -> object: - data = kwargs.get("data") - tools = getattr(data, "tools", None) - captured["tool_names"] = [t.name for t in tools.list()] if hasattr(tools, "list") else [] - return SimpleNamespace(outcome=SimpleNamespace(final_text="ok", turn_uuid="turn-1")) - - monkeypatch.setattr("runtime.gateway.run_agent_core", _run_agent_core_ok) - - store = Store() - store.kv["AIA_EXPERT_PLAN_AGENT_V2_ENABLED"] = "1" - gw = OclawGateway(store=store) - msg = StandardMessage( - session_id="sid-agent-tools", - tenant_id="t1", - user_id="u1", - role="user", - channel="admin_chat", - text="直接执行", - attachments=[], - metadata={"interaction_mode": "expert", "selected_specialist": "generalist", "execution_mode": "agent"}, - ) - out = gw.handle_turn(msg=msg, lang="zh", executor=_Exec()) - assert str(out.reply_text or "") == "ok" - names = list(captured.get("tool_names") or []) - assert "enter_plan_mode_v2" in names - assert "exit_plan_mode_v2" in names - assert store.get_setting("AIA_PLAN_AGENT_V2_DEFAULT_SESSION_ID") == "sid-agent-tools" + assert "plan 模式" not in prompt_text + assert "Plan mode is active" not in prompt_text def test_gateway_comprehensive_mode_manager_first_selects_specialist(monkeypatch: pytest.MonkeyPatch) -> None: diff --git a/tests/test_plan_agent_v2_gateway_dryrun.py b/tests/test_plan_agent_v2_gateway_dryrun.py deleted file mode 100644 index 35b1faef..00000000 --- a/tests/test_plan_agent_v2_gateway_dryrun.py +++ /dev/null @@ -1,142 +0,0 @@ -from __future__ import annotations - -from pathlib import Path - -from svc.persistence.sqlite_store import SqliteStore -from runtime.gateway import OclawGatewayResult -from runtime.plan_agent_v2 import ( - build_shadow_gateway_result, - evaluate_gateway_expert_turn_shadow, - legacy_gateway_result_keys, -) -from runtime.types import StandardMessage - - -def _msg(text: str) -> StandardMessage: - return StandardMessage( - session_id="sess-dryrun", - tenant_id="tenant-1", - user_id="user-1", - role="user", - channel="chat", - text=text, - attachments=[], - metadata={}, - ) - - -def test_gateway_shadow_stays_off_without_force_or_flag(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - out = evaluate_gateway_expert_turn_shadow( - store=store, - msg=_msg("实现一个功能"), - lang="zh", - interaction_mode="expert", - requested_specialist="generalist", - base_system_prompt="base-system", - force_flag=False, - ) - assert out.used_v2 is False - assert out.decision is None - - -def test_gateway_shadow_force_path_matches_legacy_shape(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - out = evaluate_gateway_expert_turn_shadow( - store=store, - msg=_msg("请先给计划"), - lang="zh", - interaction_mode="expert", - requested_specialist="generalist", - base_system_prompt="base-system", - force_flag=True, - ) - assert out.used_v2 is True - assert out.decision is not None - - shadow_row = build_shadow_gateway_result( - decision=out.decision, - run_id="run-1", - trace_id="trace-1", - elapsed_ms=9, - requested_specialist="generalist", - ) - assert set(shadow_row.keys()) == legacy_gateway_result_keys() - - baseline = OclawGatewayResult(run_id="run-1", reply_text="", trace_id="trace-1", elapsed_ms=9) - assert shadow_row["mode"] == baseline.mode - assert shadow_row["task_id"] == baseline.task_id - assert shadow_row["dynamic_agent_used"] == baseline.dynamic_agent_used - assert shadow_row["relay_pointer_count"] == baseline.relay_pointer_count - - -def test_gateway_shadow_confirm_path_builds_compatible_result(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - - # Enter plan - first = evaluate_gateway_expert_turn_shadow( - store=store, - msg=_msg("我要改造一下"), - lang="zh", - interaction_mode="expert", - requested_specialist="generalist", - base_system_prompt="base-system", - force_flag=True, - ) - assert first.used_v2 is True - assert first.decision is not None - assert first.decision.action == "run_agent" - assert "base-system" in str(first.decision.system_prompt_override or "") - - # Confirm plan - second = evaluate_gateway_expert_turn_shadow( - store=store, - msg=_msg("确认"), - lang="zh", - interaction_mode="expert", - requested_specialist="generalist", - base_system_prompt="base-system", - force_flag=True, - ) - assert second.used_v2 is True - assert second.decision is not None - assert second.decision.action == "stay_plan" - assert "切换到 agent 模式" in str(second.decision.reply_text or "") - - row = build_shadow_gateway_result( - decision=second.decision, - run_id="run-2", - trace_id="trace-2", - elapsed_ms=12, - requested_specialist="generalist", - ) - assert row["interaction_mode"] == "expert" - assert str(row["dispatch_reason"]).startswith("plan_agent_v2:") - - -def test_gateway_shadow_skips_v2_when_metadata_plan_agent_version_v1(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - out = evaluate_gateway_expert_turn_shadow( - store=store, - msg=StandardMessage( - session_id="sess-dryrun", - tenant_id="tenant-1", - user_id="user-1", - role="user", - channel="chat", - text="请先给计划", - attachments=[], - metadata={"plan_agent_version": "v1"}, - ), - lang="zh", - interaction_mode="expert", - requested_specialist="generalist", - base_system_prompt="base-system", - force_flag=True, - ) - assert out.used_v2 is False - assert out.decision is None - diff --git a/tests/test_plan_agent_v2_shadow.py b/tests/test_plan_agent_v2_shadow.py deleted file mode 100644 index ae359067..00000000 --- a/tests/test_plan_agent_v2_shadow.py +++ /dev/null @@ -1,450 +0,0 @@ -from __future__ import annotations - -from pathlib import Path - -from svc.persistence.sqlite_store import SqliteStore -from runtime.plan_agent_v2.adapter import evaluate_for_expert_mode -from runtime.plan_agent_v2.compat import build_shadow_gateway_result, legacy_gateway_result_keys -from runtime.plan_agent_v2.gateway_adapter import evaluate_gateway_expert_turn_shadow -from runtime.plan_agent_v2.manager import PlanModeManagerV2 -from runtime.plan_agent_v2.models import PLAN_MODE_PLAN, PlanAgentStateV2 -from runtime.plan_agent_v2.prompt_injector import build_plan_mode_prefix -from runtime.plan_agent_v2.state_store import PlanAgentStateStoreV2 -from runtime.plan_agent_v2.switch import should_route_to_v2, v2_feature_enabled -from runtime.plan_agent_v2.tool_specs import materialize_plan_mode_v2_tools -from runtime.plan_agent_v2.tool_policy import filter_tools_for_mode -from runtime.plan_agent_v2.trace import emit_plan_agent_v2_trace -from runtime.plan_agent_v2 import should_route_to_v2 as should_route_to_v2_pkg -from runtime.gateway import OclawGatewayResult -from runtime.tools.base import ToolRegistry, ToolSpec -from runtime.types import StandardMessage - - -def _dummy_tool(name: str, read_only: bool) -> ToolSpec: - def _handler(args): - return {"ok": True, "echo": args} - - return ToolSpec( - name=name, - description=name, - parameters={"type": "object", "properties": {}, "additionalProperties": True}, - handler=_handler, - read_only=read_only, - ) - - -def test_state_store_roundtrip(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - ss = PlanAgentStateStoreV2(store) - st = ss.load(session_id="s1") - assert st.mode == "normal" - saved = ss.save(session_id="s1", state=st) - loaded = ss.load(session_id="s1") - assert loaded.mode == saved.mode - - -def test_manager_enter_and_confirm(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - plan_root = tmp_path / "plans" - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(plan_root)) - mgr = PlanModeManagerV2(store=store) - st1 = mgr.enter(session_id="sess-1", owner_specialist="generalist") - assert st1.mode == PLAN_MODE_PLAN - assert st1.plan_path - assert Path(st1.plan_path).exists() - st2 = mgr.confirm(session_id="sess-1") - assert st2.mode == "normal" - assert st2.plan_confirmed is True - assert "## Goal" in str(st2.plan_content or "") - - -def test_tool_policy_filters_non_readonly_in_plan_mode() -> None: - reg = ToolRegistry([_dummy_tool("read_a", True), _dummy_tool("write_a", False)]) - out = filter_tools_for_mode(registry=reg, mode="plan") - names = {t.name for t in out} - assert "read_a" in names - assert "write_a" not in names - - -def test_tool_policy_keeps_plan_mode_control_tools() -> None: - reg = ToolRegistry([_dummy_tool("exit_plan_mode_v2", False), _dummy_tool("write_a", False)]) - out = filter_tools_for_mode(registry=reg, mode="plan") - names = {t.name for t in out} - assert "exit_plan_mode_v2" in names - assert "write_a" not in names - - -def test_adapter_agent_mode_repeated_user_injects_stall_guard(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - sid = store.create_session("stall-test").id - duplicate_line = "please handle this request" - store.add_message(session_id=sid, role="user", content=duplicate_line, event_type="user_text") - store.add_message(session_id=sid, role="assistant", content="I will analyze first…", event_type="assistant_text") - dec = evaluate_for_expert_mode( - store=store, - session_id=sid, - lang="en", - requested_specialist="generalist", - user_text=duplicate_line, - execution_mode="agent", - base_system_prompt="base", - ) - assert dec.action == "run_agent" - assert "Conversation stall guard" in str(dec.system_prompt_override or "") - assert "base" in str(dec.system_prompt_override or "") - - -def test_adapter_plan_flow(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - - d1 = evaluate_for_expert_mode( - store=store, - session_id="s1", - lang="zh", - requested_specialist="generalist", - user_text="帮我做一个功能", - execution_mode="plan", - base_system_prompt="base", - ) - assert d1.action == "run_agent" - assert isinstance(d1.plan_state, dict) - assert str(d1.plan_state.get("mode") or "") == "plan" - assert "base" in str(d1.system_prompt_override or "") - - d2 = evaluate_for_expert_mode( - store=store, - session_id="s1", - lang="zh", - requested_specialist="generalist", - user_text="确认", - execution_mode="agent", - base_system_prompt="base", - ) - assert d2.action == "run_agent" - assert "base" in str(d2.system_prompt_override or "") - - -def test_adapter_confirm_blocked_until_agent_mode(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - - _ = evaluate_for_expert_mode( - store=store, - session_id="s2", - lang="zh", - requested_specialist="generalist", - user_text="先给计划", - execution_mode="plan", - base_system_prompt="base", - ) - d2 = evaluate_for_expert_mode( - store=store, - session_id="s2", - lang="zh", - requested_specialist="generalist", - user_text="确认", - execution_mode="plan", - base_system_prompt="base", - ) - assert d2.action == "stay_plan" - assert "切换到 agent 模式" in str(d2.reply_text or "") - - -def test_adapter_confirm_strategy_auto_allows_confirm_in_plan_mode(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - store.set_setting("AIA_EXPERT_PLAN_CONFIRM_STRATEGY", "auto") - _ = evaluate_for_expert_mode( - store=store, - session_id="s-auto", - lang="zh", - requested_specialist="generalist", - user_text="先给计划", - execution_mode="plan", - base_system_prompt="base", - ) - d2 = evaluate_for_expert_mode( - store=store, - session_id="s-auto", - lang="zh", - requested_specialist="generalist", - user_text="确认", - execution_mode="plan", - base_system_prompt="base", - ) - assert d2.action == "run_agent" - assert "已确认计划" in str(d2.reply_text or "") - - -def test_adapter_confirm_strategy_off_allows_confirm_in_plan_mode(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - store.set_setting("AIA_EXPERT_PLAN_CONFIRM_STRATEGY", "off") - _ = evaluate_for_expert_mode( - store=store, - session_id="s-off", - lang="zh", - requested_specialist="generalist", - user_text="先给计划", - execution_mode="plan", - base_system_prompt="base", - ) - d2 = evaluate_for_expert_mode( - store=store, - session_id="s-off", - lang="zh", - requested_specialist="generalist", - user_text="确认", - execution_mode="plan", - base_system_prompt="base", - ) - assert d2.action == "run_agent" - assert "已确认计划" in str(d2.reply_text or "") - - -def test_adapter_plan_loop_guard_blocks_repeated_input(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - common = dict( - store=store, - session_id="s-loop", - lang="zh", - requested_specialist="generalist", - execution_mode="plan", - base_system_prompt="base", - ) - _ = evaluate_for_expert_mode(user_text="继续", **common) - d2 = evaluate_for_expert_mode(user_text="继续", **common) - assert d2.action == "stay_plan" - assert "低信息续写" in str(d2.reply_text or "") - - -def test_adapter_plan_low_signal_continue_short_circuit(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - _ = evaluate_for_expert_mode( - store=store, - session_id="s-low", - lang="zh", - requested_specialist="generalist", - user_text="先给我一版计划", - execution_mode="plan", - base_system_prompt="base", - ) - d2 = evaluate_for_expert_mode( - store=store, - session_id="s-low", - lang="zh", - requested_specialist="generalist", - user_text="继续", - execution_mode="plan", - base_system_prompt="base", - ) - assert d2.action == "stay_plan" - assert "低信息续写" in str(d2.reply_text or "") - - -def test_prompt_prefix_uses_ccmini_like_phases(tmp_path: Path) -> None: - plan_file = tmp_path / "plan.md" - plan_file.write_text("# Plan\n", encoding="utf-8") - st = PlanAgentStateV2(mode="plan", plan_path=str(plan_file)) - zh = build_plan_mode_prefix(state=st, lang="zh") - en = build_plan_mode_prefix(state=st, lang="en") - assert "阶段1：理解问题" in zh - assert "计划工作流" in zh - assert "计划模式工具" in zh - assert "执行纪律" in zh - assert "enter_plan_mode_v2" in zh - assert "Phase 1: Initial Understanding" in en - assert "Plan Workflow" in en - assert "Plan mode tools" in en - assert "Execution discipline" in en - assert "enter_plan_mode_v2" in en - - -def test_shadow_tool_specs_work(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - tools = materialize_plan_mode_v2_tools(store=store) - assert len(tools) == 2 - enter = next(t for t in tools if t.name == "enter_plan_mode_v2") - exit_tool = next(t for t in tools if t.name == "exit_plan_mode_v2") - out1 = enter.handler({"session_id": "s-1", "owner_specialist": "generalist"}) - assert out1.get("ok") is True - out2 = exit_tool.handler({"session_id": "s-1", "confirm": True}) - assert out2.get("ok") is True - assert bool((out2.get("state") or {}).get("plan_confirmed")) is True - - -def test_shadow_plan_tools_emit_trace_when_trace_id(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - tools = materialize_plan_mode_v2_tools(store=store) - enter = next(t for t in tools if t.name == "enter_plan_mode_v2") - exit_tool = next(t for t in tools if t.name == "exit_plan_mode_v2") - enter.handler( - { - "session_id": "s-tr", - "owner_specialist": "generalist", - "trace_id": "tid-1", - "parent_span_id": "ps-9", - } - ) - exit_tool.handler({"session_id": "s-tr", "confirm": False, "trace_id": "tid-1"}) - rows = store.list_trace_events_for_trace(session_id="s-tr", trace_id="tid-1") - types = [r.get("event_type") for r in rows] - assert "plan_mode_tool_enter" in types - assert "plan_mode_tool_exit" in types - enter_ev = next(r for r in rows if r.get("event_type") == "plan_mode_tool_enter") - exit_ev = next(r for r in rows if r.get("event_type") == "plan_mode_tool_exit") - assert (enter_ev.get("payload") or {}).get("tool") == "enter_plan_mode_v2" - assert (exit_ev.get("payload") or {}).get("tool") == "exit_plan_mode_v2" - assert (exit_ev.get("payload") or {}).get("confirmed") is False - - -def test_shadow_tool_specs_can_use_default_session_key(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - store.set_setting("AIA_PLAN_AGENT_V2_DEFAULT_SESSION_ID", "s-default") - tools = materialize_plan_mode_v2_tools(store=store) - enter = next(t for t in tools if t.name == "enter_plan_mode_v2") - out = enter.handler({}) - assert out.get("ok") is True - st = out.get("state") or {} - assert str(st.get("mode") or "") == "plan" - - -def test_switch_default_off_and_opt_in(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - assert v2_feature_enabled(store=store) is False - assert should_route_to_v2(store=store, interaction_mode="expert") is False - assert should_route_to_v2(store=store, interaction_mode="expert", force_flag=True) is True - store.set_setting("AIA_EXPERT_PLAN_AGENT_V2_ENABLED", "1") - assert v2_feature_enabled(store=store) is True - assert should_route_to_v2(store=store, interaction_mode="expert") is True - assert should_route_to_v2(store=store, interaction_mode="comprehensive") is False - - -def test_gateway_adapter_shadow_force_flag(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - msg = StandardMessage( - session_id="s1", - tenant_id="t1", - user_id="u1", - role="user", - channel="chat", - text="帮我实现一个功能", - attachments=[], - metadata={}, - ) - out = evaluate_gateway_expert_turn_shadow( - store=store, - msg=msg, - lang="zh", - interaction_mode="expert", - requested_specialist="generalist", - base_system_prompt="base", - force_flag=True, - ) - assert out.used_v2 is True - assert out.decision is not None - assert out.decision.action == "run_agent" - - -def test_shadow_compat_result_shape(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - dec = evaluate_for_expert_mode( - store=store, - session_id="s1", - lang="zh", - requested_specialist="generalist", - user_text="我要改造", - base_system_prompt="base", - ) - row = build_shadow_gateway_result( - decision=dec, - run_id="r1", - trace_id="t1", - elapsed_ms=12, - requested_specialist="generalist", - ) - assert set(row.keys()) == legacy_gateway_result_keys() - - -def test_trace_helper_no_crash() -> None: - events = [] - - class _S: - def add_trace_event(self, **kwargs): - events.append(kwargs) - - emit_plan_agent_v2_trace( - store=_S(), - session_id="s1", - trace_id="t1", - parent_span_id=None, - event_type="plan_mode_entered", - payload={"x": 1}, - ) - assert len(events) == 1 - assert events[0].get("event_type") == "plan_mode_entered" - - -def test_shadow_gateway_result_defaults_align_legacy_baseline(tmp_path: Path) -> None: - store = SqliteStore(str(tmp_path / "ops.sqlite")) - store.set_setting("AIA_EXPERT_PLAN_FILE_DIR", str(tmp_path / "plans")) - dec = evaluate_for_expert_mode( - store=store, - session_id="s1", - lang="zh", - requested_specialist="generalist", - user_text="继续", - base_system_prompt="base", - ) - row = build_shadow_gateway_result( - decision=dec, - run_id="r1", - trace_id="t1", - elapsed_ms=1, - requested_specialist="generalist", - ) - # Baseline invariant fields expected by legacy result dataclass. - baseline = OclawGatewayResult(run_id="r1", reply_text="", trace_id="t1", elapsed_ms=1) - assert row["mode"] == baseline.mode - assert row["task_id"] == baseline.task_id - assert row["dynamic_agent_used"] == baseline.dynamic_agent_used - assert row["dynamic_agent_name"] == baseline.dynamic_agent_name - assert row["relay_pointer_count"] == baseline.relay_pointer_count - assert row["relay_envelope_present"] == baseline.relay_envelope_present - assert row["relay_envelope_pointer_count"] == baseline.relay_envelope_pointer_count - assert row["relay_ttl_turn_count"] == baseline.relay_ttl_turn_count - assert row["relay_ttl_session_count"] == baseline.relay_ttl_session_count - assert row["relay_ttl_keep_count"] == baseline.relay_ttl_keep_count - # Package export should be wired and callable. - assert callable(should_route_to_v2_pkg) - - -def test_package_exports_stable_symbols() -> None: - import runtime.plan_agent_v2 as p - - required = [ - "PlanAgentStateV2", - "PlanAgentStateStoreV2", - "PlanModeManagerV2", - "PlanAgentV2Decision", - "GatewayPlanV2AdapterOutput", - "evaluate_for_expert_mode", - "evaluate_gateway_expert_turn_shadow", - "should_route_to_v2", - "v2_feature_enabled", - "emit_plan_agent_v2_trace", - "build_shadow_gateway_result", - "legacy_gateway_result_keys", - ] - for name in required: - assert hasattr(p, name), name - diff --git a/tests/test_skill_runtime_metadata_and_tools.py b/tests/test_skill_runtime_metadata_and_tools.py index eac2cec7..28cf2005 100644 --- a/tests/test_skill_runtime_metadata_and_tools.py +++ b/tests/test_skill_runtime_metadata_and_tools.py @@ -6,7 +6,6 @@ import unittest from pathlib import Path from runtime.skills import discover_workspace_skill_manifests -from runtime.tools.skills_runtime.materialize_skill_tools import materialize_executable_skill_tools class SkillRuntimeMetadataAndToolsTests(unittest.TestCase): @@ -52,28 +51,6 @@ class SkillRuntimeMetadataAndToolsTests(unittest.TestCase): assert m.runtime.get("entry") == "scripts/run.py" assert isinstance(m.runtime.get("schema"), dict) - def test_materialize_creates_tool_for_runtime_skill(self) -> None: - d = self.skills_root / "demo_runtime_tool" - d.mkdir(parents=True, exist_ok=True) - (d / "scripts").mkdir(parents=True, exist_ok=True) - (d / "scripts" / "run.py").write_text("print('{\"ok\": true}')\n", encoding="utf-8") - (d / "SKILL.md").write_text( - "---\n" - "name: demo_runtime_tool\n" - "description: demo\n" - "metadata:\n" - " oclaw:\n" - " runtime:\n" - " type: python\n" - " entry: scripts/run.py\n" - "---\n", - encoding="utf-8", - ) - tools = materialize_executable_skill_tools(store=None) - names = {t.name for t in tools} - assert "demo_runtime_tool" in names - if __name__ == "__main__": unittest.main() - diff --git a/tests/test_ume_alarm_xlsx_report.py b/tests/test_ume_alarm_xlsx_report.py index 7d36fc13..6f90178b 100644 --- a/tests/test_ume_alarm_xlsx_report.py +++ b/tests/test_ume_alarm_xlsx_report.py @@ -126,10 +126,9 @@ def test_filter_preset_and_row_helpers() -> None: assert meta["total"] == 2 -def test_ume_alarm_xlsx_report_registered_when_builtin_disabled(monkeypatch) -> None: +def test_ume_alarm_xlsx_report_registered_without_inline_netx(monkeypatch) -> None: from runtime.tools import expert_registry - monkeypatch.delenv("OCLAW_NETX_BUILTIN_TOOLS", raising=False) expert_registry.clear_expert_tool_cache() factories = expert_registry.discover_expert_tool_factories() names = {f().name for f in (factories.get("network_ops") or [])}