From d8bc08ee0dc30835e18574fac9d0ecb3ce52725a Mon Sep 17 00:00:00 2001 From: oliver Date: Mon, 10 Aug 2026 23:21:23 +0800 Subject: [PATCH] Stop shell Excel builds, default write_file paths, and cap short-intent tool rounds. Refuse openpyxl/pandas-to_excel via run_command, auto-write content-only files under tmp/, and limit WhatsApp ops short intents to 8 tool rounds by default. Co-authored-by: Cursor --- runtime/gateway.py | 24 +++++++++++++- runtime/tools/public/run_command_tool.py | 40 +++++++++++++++++++++-- runtime/tools/public/write_file_tool.py | 34 ++++++++++++++------ tests/test_oclaw_gateway_trace.py | 41 ++++++++++++++++++++++++ tests/test_run_command_aliases.py | 15 +++++++++ tests/test_write_file_tool_aliases.py | 21 +++++++++++- 6 files changed, 162 insertions(+), 13 deletions(-) diff --git a/runtime/gateway.py b/runtime/gateway.py index c34c1bf8..d8b6d8fe 100644 --- a/runtime/gateway.py +++ b/runtime/gateway.py @@ -484,6 +484,25 @@ class OclawGateway: return str(maybe_ops_short_intent_system_hint(text=str(msg.text or ""), lang=lang) or "").strip() + def _resolve_max_tool_rounds(self, msg: StandardMessage, *, base: int) -> int: + """Cap tool rounds for WhatsApp/WeChat ops short intents (cut 12+ tool loops).""" + rounds = max(1, min(int(base), 300)) + if not self._is_channel_delivery_channel(msg): + return rounds + md = msg.metadata if isinstance(msg.metadata, dict) else {} + from runtime.application.gateway.ops_short_intent import detect_ops_short_intent + + intent = detect_ops_short_intent(str(msg.text or md.get("raw_inbound_text") or "")) + if not intent: + return rounds + try: + raw = str(self.store.get_setting("AIA_OPS_SHORT_INTENT_MAX_TOOL_ROUNDS") or "").strip() + cap = int(raw) if raw.isdigit() else 8 + except Exception: + cap = 8 + cap = max(3, min(int(cap), 30)) + return min(rounds, cap) + @staticmethod def _group_focus_system_hint(msg: StandardMessage, lang: str) -> str: """Only for shared group transcripts (legacy session_scope=chat / __group__).""" @@ -1242,7 +1261,10 @@ class OclawGateway: parent_span_id=None, run_id=rid, max_messages=_get_int_setting("AIA_TURN_MAX_CONTEXT_MESSAGES", 80, 10, 400), - max_tool_rounds=_get_int_setting("AIA_TURN_MAX_TOOL_ROUNDS", 100, 1, 300), + max_tool_rounds=self._resolve_max_tool_rounds( + msg, + base=_get_int_setting("AIA_TURN_MAX_TOOL_ROUNDS", 100, 1, 300), + ), max_tool_workers=_get_int_setting("AIA_TURN_MAX_TOOL_WORKERS", 8, 1, 32), max_attempts=_get_int_setting("AIA_OCLAW_MAX_ATTEMPTS", 2, 1, 5), memory_context=memory_context, diff --git a/runtime/tools/public/run_command_tool.py b/runtime/tools/public/run_command_tool.py index 79d8503e..e918a9d9 100644 --- a/runtime/tools/public/run_command_tool.py +++ b/runtime/tools/public/run_command_tool.py @@ -5,6 +5,26 @@ from typing import Any from runtime.tools.base import ToolSpec from runtime.tools.public.local_sdk import get_local_adapter +_XLSX_SHELL_MARKERS = ( + "openpyxl", + "xlsxwriter", + "to_excel(", + "workbook(", + "load_workbook(", +) + + +def _looks_like_xlsx_via_shell(command: str) -> bool: + """Detect agents trying to build Excel via shell/python instead of write_xlsx.""" + low = str(command or "").lower() + if not low: + return False + if any(m in low for m in _XLSX_SHELL_MARKERS): + return True + if ".xlsx" in low and any(tok in low for tok in ("python", "pip", "pandas", "-c ", "import ")): + return True + return False + def run_command_tool() -> ToolSpec: def _handler(args: dict[str, Any]) -> dict[str, Any]: @@ -19,13 +39,29 @@ def run_command_tool() -> ToolSpec: "hint": "Pass command (aliases: cmd, shell).", "example": {"command": "echo hello", "timeout": 60}, } + if _looks_like_xlsx_via_shell(command): + return { + "ok": False, + "error_code": "xlsx_via_shell_forbidden", + "error": "xlsx_via_shell_forbidden", + "failure_class": "schema_validation", + "retry_forbidden": True, + "hint": ( + "Do not build Excel via run_command/openpyxl/pandas. " + "Use ume_alarm_xlsx_report or write_xlsx(deliverable=true) instead." + ), + "fallback_tools": ["ume_alarm_xlsx_report", "write_xlsx"], + } cwd = str(args.get("cwd") or args.get("workdir") or "").strip() or None timeout = int(args.get("timeout") or 300) return get_local_adapter().run_command(command=command, cwd=cwd, timeout=timeout) return ToolSpec( name="run_command", - description="Run a shell command via local backend. Prefer cmd aliases: command/cmd/shell.", + description=( + "Run a shell command via local backend. Prefer cmd aliases: command/cmd/shell. " + "Do not use this to build .xlsx (use write_xlsx / ume_alarm_xlsx_report)." + ), parameters={ "type": "object", "properties": { @@ -48,4 +84,4 @@ def run_command_tool() -> ToolSpec: ) -__all__ = ["run_command_tool"] +__all__ = ["run_command_tool", "_looks_like_xlsx_via_shell"] diff --git a/runtime/tools/public/write_file_tool.py b/runtime/tools/public/write_file_tool.py index 0e13928e..1020ca63 100644 --- a/runtime/tools/public/write_file_tool.py +++ b/runtime/tools/public/write_file_tool.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import datetime, timezone from typing import Any from runtime.tools.base import ToolSpec @@ -14,21 +15,31 @@ def _resolve_write_path(args: dict[str, Any]) -> str: return "" +def _default_write_path() -> str: + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return f"tmp/write_{stamp}.txt" + + def write_file_tool() -> ToolSpec: def _handler(args: dict[str, Any]) -> dict[str, Any]: raw = _resolve_write_path(args) - if not raw: - return { - "ok": False, - "error": "path_required", - "hint": "Pass path (or file/filename) relative to workspace root.", - "example": {"path": "tmp/notes.txt", "content": "hello", "mode": "overwrite"}, - } content = args.get("content") if content is None: content = args.get("text") if content is None: content = args.get("body") + auto_path = False + if not raw: + if content is None: + return { + "ok": False, + "error": "path_required", + "hint": "Pass path (or file/filename) relative to workspace root.", + "example": {"path": "tmp/notes.txt", "content": "hello", "mode": "overwrite"}, + } + # Production WA agents often omit path; default under tmp/ instead of schema-fail looping. + raw = _default_write_path() + auto_path = True content_s = "" if content is None else str(content) mode = str(args.get("mode") or "overwrite").strip().lower() try: @@ -42,13 +53,18 @@ def write_file_tool() -> ToolSpec: p.write_text(p.read_text(encoding="utf-8", errors="replace") + content_s, encoding="utf-8") else: p.write_text(content_s, encoding="utf-8") - return {"ok": True, "path": str(p), "bytes": p.stat().st_size} + out: dict[str, Any] = {"ok": True, "path": str(p), "bytes": p.stat().st_size} + if auto_path: + out["auto_path"] = True + out["hint"] = f"No path provided; wrote content to {raw}." + return out return ToolSpec( name="write_file", description=( "Write text content to a workspace file (overwrite or append). " - "Path aliases: file, filename, file_path." + "Path aliases: file, filename, file_path. " + "If path is omitted but content is provided, writes to tmp/write_.txt." ), parameters={ "type": "object", diff --git a/tests/test_oclaw_gateway_trace.py b/tests/test_oclaw_gateway_trace.py index db999046..06231cc2 100644 --- a/tests/test_oclaw_gateway_trace.py +++ b/tests/test_oclaw_gateway_trace.py @@ -993,3 +993,44 @@ def test_group_focus_system_hint_only_for_shared_group_scope() -> None: assert "current sender" in hint assert OclawGateway._group_focus_system_hint(per_user, "en") == "" assert OclawGateway._group_focus_system_hint(dm, "en") == "" + + +def test_ops_short_intent_caps_tool_rounds(tmp_path) -> None: + from runtime.types import StandardMessage + from svc.persistence.sqlite_store import SqliteStore + + store = SqliteStore(str(tmp_path / "rounds.sqlite")) + gw = OclawGateway(store=store) + short = StandardMessage( + session_id="s1", + tenant_id="t1", + user_id="u1", + role="member", + channel="whatsapp", + text="fiber cut report", + attachments=[], + metadata={}, + ) + long = StandardMessage( + session_id="s1", + tenant_id="t1", + user_id="u1", + role="member", + channel="whatsapp", + text="Please investigate the full OSPF adjacency flap history across all PE routers and draft a long RCA.", + attachments=[], + metadata={}, + ) + admin = StandardMessage( + session_id="s1", + tenant_id="t1", + user_id="u1", + role="member", + channel="admin", + text="fiber cut report", + attachments=[], + metadata={}, + ) + assert gw._resolve_max_tool_rounds(short, base=100) == 8 + assert gw._resolve_max_tool_rounds(long, base=100) == 100 + assert gw._resolve_max_tool_rounds(admin, base=100) == 100 diff --git a/tests/test_run_command_aliases.py b/tests/test_run_command_aliases.py index a6ab25d1..933cce99 100644 --- a/tests/test_run_command_aliases.py +++ b/tests/test_run_command_aliases.py @@ -24,3 +24,18 @@ def test_run_command_accepts_cmd_alias(monkeypatch) -> None: out = spec.handler(filtered) assert out.get("ok") is True assert calls[0]["command"] == "echo hi" + + +def test_run_command_blocks_openpyxl_xlsx() -> None: + from runtime.tools.public.run_command_tool import _looks_like_xlsx_via_shell + + assert _looks_like_xlsx_via_shell("python -c \"import openpyxl; wb=openpyxl.Workbook()\"") + assert _looks_like_xlsx_via_shell("python -c \"df.to_excel('a.xlsx')\"") + assert not _looks_like_xlsx_via_shell("echo hello") + + spec = run_command_tool() + out = spec.handler({"command": "python -c \"import openpyxl; openpyxl.Workbook()\""}) + assert out.get("ok") is False + assert out.get("error_code") == "xlsx_via_shell_forbidden" + assert out.get("retry_forbidden") is True + assert "write_xlsx" in (out.get("fallback_tools") or []) diff --git a/tests/test_write_file_tool_aliases.py b/tests/test_write_file_tool_aliases.py index c9c3f3e0..cfb2a0d0 100644 --- a/tests/test_write_file_tool_aliases.py +++ b/tests/test_write_file_tool_aliases.py @@ -1,5 +1,7 @@ from __future__ import annotations +from pathlib import Path + from runtime.tools.public.write_file_tool import write_file_tool from runtime.tools.tool_validation import filter_arguments_to_schema, validate_tool_arguments @@ -27,7 +29,24 @@ def test_write_file_schema_allows_aliases() -> None: def test_write_file_path_required_message() -> None: spec = write_file_tool() - out = spec.handler({"content": "x"}) + out = spec.handler({}) assert out.get("ok") is False assert out.get("error") == "path_required" assert "example" in out + + +def test_write_file_defaults_path_when_content_only(tmp_path, monkeypatch) -> None: + def _resolve(raw: str): + return tmp_path / "workspace" / str(raw).replace("\\", "/").lstrip("/") + + monkeypatch.setattr( + "runtime.tools.public.write_file_tool.resolve_workspace_path", + _resolve, + ) + spec = write_file_tool() + out = spec.handler({"content": "hello-auto"}) + assert out.get("ok") is True + assert out.get("auto_path") is True + path = Path(str(out.get("path") or "")) + assert path.name.startswith("write_") + assert path.read_text(encoding="utf-8") == "hello-auto"