hansjone/oclaw
1
0
Fork 0
mirror of https://github.com/hansjone/oclaw.git synced 2026-10-10 14:30:45 +08:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat(ops): enforce English-only replies and localize netx protocol labels

Browse source 
Append english_output_guard when lang=en, strengthen ops/generalist EN role prompts,
pass NETX_TOOL_LANG into netx tools, and map Chinese protocol buckets to English.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-05-19 23:35:46 +08:00
parent 0de5f7bc98
commit 1366459ee0
10 changed files with 146 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

42
runtime/tools/experts/network_ops/netx_tools.py
View file

@ -7,6 +7,7 @@ Configure via environment:
from __future__ import annotations
import contextvars
import os
import threading
import time
@ -16,6 +17,16 @@ import httpx
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_ne_id",
@ -65,17 +76,43 @@ def _netx_headers() -> dict[str, str]:
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_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=params or None, headers=_netx_headers())
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]}
@ -166,7 +203,8 @@ def _format_ops_netx_system_extension(r: dict[str, Any], *, lang_en: bool) -> st
tail_en = (
"- tools: netx_query_ume_alarms, netx_aggregate_ume_alarms, netx_run_ume_diagnostics, "
"netx_query_ume_ne_inventory, netx_get_ume_ne\n"
"- note: this is only runtime anchor; use tools for alarm/ne evidence."
"- note: this is only runtime anchor; use tools for alarm/ne evidence.\n"
"- English session: user-visible reply must contain NO Chinese/CJK; translate alarm text fields."
)
tail_zh = (
"- 工具: netx_query_ume_alarms、netx_aggregate_ume_alarms、netx_run_ume_diagnostics、"