From aff11c1eaea30e2611ad7e7e441f9b6ad6e60722 Mon Sep 17 00:00:00 2001 From: oliver Date: Mon, 10 Aug 2026 22:32:44 +0800 Subject: [PATCH] Reduce WhatsApp ops tool friction with looser xlsx/file schemas and short-intent playbooks. Production usage showed schema mismatches and multi-tool loops; accept common arg aliases and route agents to <=3-call alarm recipes. Co-authored-by: Cursor --- docs/presentations/oclaw-netx-ops-expert.md | 3 +- runtime/tools/public/write_file_tool.py | 42 +++++- runtime/tools/public/write_xlsx_tool.py | 125 ++++++++++++++++-- runtime/workspaces/ops/ROLE_SYSTEM.en.md | 8 +- runtime/workspaces/ops/ROLE_SYSTEM.md | 9 +- .../ops/ops-netx-managed-ne-playbook/SKILL.md | 2 + .../ops/ops-netx-ume-playbook/SKILL.md | 22 ++- .../ops/ops-netx-ume-playbook/reference.md | 10 ++ .../public/channel-file-delivery/SKILL.md | 3 +- tests/test_write_file_tool_aliases.py | 33 +++++ tests/test_write_xlsx_tool.py | 43 ++++++ 11 files changed, 276 insertions(+), 24 deletions(-) create mode 100644 tests/test_write_file_tool_aliases.py diff --git a/docs/presentations/oclaw-netx-ops-expert.md b/docs/presentations/oclaw-netx-ops-expert.md index cae4e763..036012f7 100644 --- a/docs/presentations/oclaw-netx-ops-expert.md +++ b/docs/presentations/oclaw-netx-ops-expert.md @@ -491,7 +491,8 @@ flowchart TD 运维专家： 1. 加载 ops-netx-ume-playbook - 2. mcp__netx__aggregateUmeAlarms(severity=critical, group_by=alarm_host_name) + 2. mcp__netx__aggregateUmeAlarms(severity=critical, top_ne=5) + # 若需自定义维度：aggregateUmeAlarms(group_by=alarm_host_name, severity=critical, limit=5) 3. 返回结论 + 证据表格 ``` diff --git a/runtime/tools/public/write_file_tool.py b/runtime/tools/public/write_file_tool.py index c17e8f66..0e13928e 100644 --- a/runtime/tools/public/write_file_tool.py +++ b/runtime/tools/public/write_file_tool.py @@ -6,12 +6,30 @@ from runtime.tools.base import ToolSpec from runtime.tools.path_guard import resolve_workspace_path +def _resolve_write_path(args: dict[str, Any]) -> str: + for key in ("path", "file", "filename", "file_path", "filepath", "name"): + raw = str(args.get(key) or "").strip().strip('"').strip("'") + if raw: + return raw + return "" + + def write_file_tool() -> ToolSpec: def _handler(args: dict[str, Any]) -> dict[str, Any]: - raw = str(args.get("path") or "").strip().strip('"').strip("'") + raw = _resolve_write_path(args) if not raw: - return {"ok": False, "error": "path_required"} - content = str(args.get("content") or "") + 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") + content_s = "" if content is None else str(content) mode = str(args.get("mode") or "overwrite").strip().lower() try: p = resolve_workspace_path(raw) @@ -21,22 +39,32 @@ def write_file_tool() -> ToolSpec: if mode not in ("overwrite", "append"): return {"ok": False, "error": "invalid_mode", "allowed": ["overwrite", "append"]} if mode == "append": - p.write_text(p.read_text(encoding="utf-8", errors="replace") + content, encoding="utf-8") + p.write_text(p.read_text(encoding="utf-8", errors="replace") + content_s, encoding="utf-8") else: - p.write_text(content, encoding="utf-8") + p.write_text(content_s, encoding="utf-8") return {"ok": True, "path": str(p), "bytes": p.stat().st_size} return ToolSpec( name="write_file", - description="Write text content to a workspace file (overwrite or append).", + description=( + "Write text content to a workspace file (overwrite or append). " + "Path aliases: file, filename, file_path." + ), parameters={ "type": "object", "properties": { "path": {"type": "string", "description": "File path, relative to workspace root."}, + "file": {"type": "string", "description": "Alias for path."}, + "filename": {"type": "string", "description": "Alias for path."}, + "file_path": {"type": "string", "description": "Alias for path."}, + "filepath": {"type": "string", "description": "Alias for path."}, + "name": {"type": "string", "description": "Alias for path (when it looks like a relative file path)."}, "content": {"type": "string", "description": "Full text content to write."}, + "text": {"type": "string", "description": "Alias for content."}, + "body": {"type": "string", "description": "Alias for content."}, "mode": {"type": "string", "enum": ["overwrite", "append"], "default": "overwrite"}, }, - "required": ["path", "content"], + "required": [], "additionalProperties": False, }, handler=_handler, diff --git a/runtime/tools/public/write_xlsx_tool.py b/runtime/tools/public/write_xlsx_tool.py index f4e3790f..80062194 100644 --- a/runtime/tools/public/write_xlsx_tool.py +++ b/runtime/tools/public/write_xlsx_tool.py @@ -97,7 +97,7 @@ def _build_workbook(sheets: list[dict[str, Any]], *, freeze_header: bool, auto_w for i, sheet in enumerate(sheets, start=1): if not isinstance(sheet, dict): continue - rows, inferred_cols = _normalize_rows(sheet.get("rows")) + rows, inferred_cols = _normalize_rows(sheet.get("rows") if sheet.get("rows") is not None else sheet.get("data")) headers_raw = sheet.get("headers") if isinstance(headers_raw, list) and headers_raw: col_count = max(len(headers_raw), inferred_cols, 1) @@ -129,15 +129,62 @@ def _build_workbook(sheets: list[dict[str, Any]], *, freeze_header: bool, auto_w return buf.getvalue(), summary +_XLSX_EXAMPLE = { + "name": "alarm_summary.xlsx", + "sheets": [ + { + "name": "by_host", + "headers": ["host_name", "severity", "count"], + "rows": [["NE-A", "critical", 12], ["NE-B", "major", 5]], + } + ], + "freeze_header": True, + "auto_width": True, +} + + +def _coerce_sheets(args: dict[str, Any]) -> list[Any] | None: + """Accept common agent shapes: sheets[], or top-level headers/rows[/name].""" + sheets_raw = args.get("sheets") + if isinstance(sheets_raw, list) and sheets_raw: + return sheets_raw + # Single sheet mistaken as top-level object + if isinstance(sheets_raw, dict) and ( + "rows" in sheets_raw or "headers" in sheets_raw or "data" in sheets_raw + ): + return [sheets_raw] + headers = args.get("headers") + rows = args.get("rows") + if rows is None: + rows = args.get("data") + if isinstance(rows, list): + sheet: dict[str, Any] = {"rows": rows} + if isinstance(headers, list): + sheet["headers"] = headers + title = str(args.get("sheet_name") or args.get("sheet") or "").strip() + if title: + sheet["name"] = title + return [sheet] + return None + + def write_xlsx_tool() -> ToolSpec: def _handler(args: dict[str, Any]) -> dict[str, Any]: - sheets_raw = args.get("sheets") - if not isinstance(sheets_raw, list) or not sheets_raw: - return {"ok": False, "error": "sheets_required"} + sheets_raw = _coerce_sheets(args) + if not sheets_raw: + return { + "ok": False, + "error": "sheets_required", + "hint": "Pass sheets=[{name, headers, rows}] (or top-level headers+rows).", + "example": _XLSX_EXAMPLE, + } if len(sheets_raw) > _MAX_SHEETS: return {"ok": False, "error": "too_many_sheets", "max_sheets": _MAX_SHEETS} - filename = str(args.get("name") or "").strip() or "report.xlsx" + filename = ( + str(args.get("name") or args.get("filename") or args.get("file_name") or "").strip() + or "report.xlsx" + ) if not filename.lower().endswith(".xlsx"): filename = f"{filename}.xlsx" freeze_header = args.get("freeze_header") is not False @@ -200,7 +247,10 @@ def write_xlsx_tool() -> ToolSpec: "properties": { "sheets": { "type": "array", - "description": "One or more sheets. Each item: {name, headers[], rows[][]}.", + "description": ( + "Preferred: one or more sheets [{name, headers[], rows[][]}]. " + "If omitted, top-level headers+rows are accepted as a single sheet." + ), "items": { "type": "object", "properties": { @@ -213,17 +263,72 @@ def write_xlsx_tool() -> ToolSpec: "rows": { "type": "array", "description": "Data rows: each row is an array of cell values (string/number/bool/null).", - "items": {"type": "array"}, + "items": { + "anyOf": [ + {"type": "array"}, + {"type": "object"}, + ] + }, + }, + "data": { + "type": "array", + "description": "Alias for rows.", + "items": { + "anyOf": [ + {"type": "array"}, + {"type": "object"}, + ] + }, }, }, - "required": ["rows"], - "additionalProperties": False, + "additionalProperties": True, }, }, + "headers": { + "type": "array", + "items": {"type": "string"}, + "description": "Shortcut when sheets omitted: column headers for a single sheet.", + }, + "rows": { + "type": "array", + "description": "Shortcut when sheets omitted: data rows for a single sheet.", + "items": { + "anyOf": [ + {"type": "array"}, + {"type": "object"}, + ] + }, + }, + "data": { + "type": "array", + "description": "Alias for top-level rows.", + "items": { + "anyOf": [ + {"type": "array"}, + {"type": "object"}, + ] + }, + }, + "sheet_name": { + "type": "string", + "description": "Shortcut sheet tab name when using top-level headers/rows.", + }, + "sheet": { + "type": "string", + "description": "Alias for sheet_name.", + }, "name": { "type": "string", "description": "Download filename, e.g. alarm_summary.xlsx", }, + "filename": { + "type": "string", + "description": "Alias for name.", + }, + "file_name": { + "type": "string", + "description": "Alias for name.", + }, "path": { "type": "string", "description": "Optional workspace path to also write the .xlsx file (mirror).", @@ -239,7 +344,7 @@ def write_xlsx_tool() -> ToolSpec: "description": "Best-effort column width from sample cells (default true).", }, }, - "required": ["sheets"], + "required": [], "additionalProperties": False, }, handler=_handler, diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.en.md b/runtime/workspaces/ops/ROLE_SYSTEM.en.md index 514055bb..7c56416a 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.en.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.en.md @@ -28,9 +28,15 @@ You are the ops specialist (network operations expert). - **Use `host_name` as the primary key for every NE dimension** (first table column, Top-N keys, group-by, and how you refer to an NE in prose). After sync, netx stores it on the alarm row — prefer: - List/paged alarms: **`host_name`** from `mcp__netx__queryUmeAlarms` - Raw/SQL: **`alarm_host_name`** (over `ne_host_name` when both exist) + - Aggregate: default `by_ne` from `mcp__netx__aggregateUmeAlarms`; custom dims via `group_by=alarm_host_name` (routes to raw aggregate) - **Never** use `ne_id` / `alarm_ne_id` (UUID) as the user-facing primary key; `ne_id` is for filters and joins only. - If `host_name` is empty, fall back to `user_label` / `ne_name` with a "host_name missing" note — never bare `ne_id`. -- NE stats/aggregates: prefer `group_by=alarm_host_name` or `group_by=ne_host_name`; do not group by `alarm_ne_id` / `ne_ne_id` for user output. +- NE stats/aggregates: prefer `aggregateUmeAlarms(group_by=alarm_host_name)` or `aggregateUmeAlarmsRaw`; do not group by `alarm_ne_id` / `ne_ne_id` for user output. + +## WhatsApp interaction (mandatory) +- Short ops intents follow `ops-netx-ume-playbook` WhatsApp recipes; target **≤3 tool calls** per user message. +- Spreadsheet delivery: `write_xlsx` then `save_deliverable_attachment` — never claim a file was sent without the deliverable step. +- Call `listCliTargets` at most once per session and reuse ids; batch `execManagedNe` commands; on timeout raise `read_timeout_sec` — no blind retries. ## Required skills - For every netx/UME **alarm or NE** request, load and follow skill: `ops-netx-ume-playbook` (skill text may be Chinese; **user-facing output must still match the user's language**). diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.md b/runtime/workspaces/ops/ROLE_SYSTEM.md index b2382e76..89020539 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.md @@ -19,10 +19,15 @@ ## 告警与网元展示（强制） - **网元维度一律以 `host_name` 为主键展示**（表格首列、Top 排名键、分组维度、结论中的网元指称）。告警同步后 netx 已把 `host_name` 写入告警表，优先读： - 列表/分页：`mcp__netx__queryUmeAlarms`（或 legacy `netx_query_ume_alarms`）返回的 **`host_name`** - - 聚合：`mcp__netx__aggregateUmeAlarms` 的网元维度字段 + - 聚合：`mcp__netx__aggregateUmeAlarms` 的 `by_ne`（默认按 host）；自定义维度用 `group_by=alarm_host_name`（会路由到 Raw 聚合） - **禁止**用 `ne_id` / `alarm_ne_id`（UUID）作为对用户的主展示键；`ne_id` 仅用于工具过滤或内部关联。 - 若 `host_name` 为空，再用 `user_label` / `ne_name` 并标注「host_name 缺失」；仍不得用裸 `ne_id`。 -- 按网元统计/聚合：优先 `group_by=alarm_host_name` 或 `group_by=ne_host_name`，勿按 `alarm_ne_id` / `ne_ne_id` 对外展示。 +- 按网元统计/聚合：优先 `aggregateUmeAlarms(group_by=alarm_host_name)` 或 `aggregateUmeAlarmsRaw`；勿按 `alarm_ne_id` / `ne_ne_id` 对外展示。 + +## WhatsApp 交互（强制） +- 短句优先走 `ops-netx-ume-playbook` 的「WhatsApp 短指令配方」，控制在 ≤3 次工具调用。 +- 用户要表格/Excel：`write_xlsx` → `save_deliverable_attachment`；禁止只写文件不投递。 +- `listCliTargets` 每会话最多查一次并复用 id；`execManagedNe` 合并 commands，超时调 `read_timeout_sec`，禁止盲重试。 ## 必须加载技能 - 每次处理 netx/UME **告警或网元** 问题时，必须加载并遵循技能：`ops-netx-ume-playbook`。 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 f1f05e73..ae7ced64 100644 --- a/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md +++ b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md @@ -21,6 +21,8 @@ description: 面向 ops 专家的 netx 纳管网元（网元管理）作业手 - **UME 清单（无需逐台纳管）**：`mcp__netx__listCliTargets`（`source=ume`）或 `queryUmeNeInventory` 取 `ne_id`，再用 `ume_ne_id` 执行 CLI（需先在 netx **UME → CLI 连接** 配置统一凭据/跳板） 2. **登录查信息** - `mcp__netx__execManagedNe`：`ne_id` **或** `ume_ne_id` + `commands`（默认最多 5 条，可由 `NETX_NE_EXEC_MAX_COMMANDS` 调高，硬上限 50） + - **一次会话内**：`listCliTargets` 最多调用一次，缓存返回的 id；多条 show 合并进同一次 `commands`，禁止「list→exec→list→exec」循环 + - 超时：提高 `read_timeout_sec`（60–120）或减少命令条数，禁止对同一命令盲重试 ## CLI 约束（服务端强制） diff --git a/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md b/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md index aef70a95..1ba438a1 100644 --- a/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md +++ b/skills/_workspace/ops/ops-netx-ume-playbook/SKILL.md @@ -44,6 +44,7 @@ description: 面向 ops 专家的 netx UME 运维作业手册。覆盖告警查 - **整体态势 / Top 风险**：`runUmeDiagnostics` + `aggregateUmeAlarms`（默认已排除 missing host；看 `by_ne_missing`）。 - 高危 Top-N：`aggregateUmeAlarms(severity=critical, top_ne=10)`，勿只看总量 Top。 + - 按 host 自定义分组：`aggregateUmeAlarms(group_by=alarm_host_name, …)`（内部等同 Raw）或显式 `aggregateUmeAlarmsRaw`。 - **时间收敛**：`queryUmeAlarms` / `aggregateUmeAlarms` / raw 均支持 `time_from`/`time_to`（语义=`last_seen_at`）。先看 freshness，再填时间窗。 - **可引用证据**：`queryUmeAlarmsRaw` + `field_preset=evidence`。 - **任意字段统计**：`aggregateUmeAlarmsRaw`（按 host 分组时默认排除 missing；看 `by_ne_missing`）。 @@ -51,6 +52,23 @@ description: 面向 ops 专家的 netx UME 运维作业手册。覆盖告警查 - **critical 口类/光路类告警**：抽 1–2 个 `ne_id` → `findTopologyPaths`（默认 `detail=summary`，看 `paths[].label`）→ 再决定是否 CLI。 - **查网元身份**：`queryUmeNeInventory(keyword=host_name)`；完整 `raw_json` 用 `getUmeNe`。 +## WhatsApp 短指令配方（强制少工具） + +群聊短句（中/英）优先走下列固定路径，**目标 ≤3 次工具调用**，不要先翻页/反复 listCliTargets。 + +| 用户说法（例） | 配方 | +|----------------|------| +| 断纤 / fiber cut / LOS / 光缆中断 | ① `queryUmeAlarmsRaw(keyword=LOS\|fiber\|断纤\|光缆, field_preset=evidence, page_size=50)` 或 `keyword`+severity；② 摘要表；若要文件：`write_xlsx`→`save_deliverable_attachment` | +| 离线 / 单板离线 / offline NE | ① `queryUmeAlarms`/`Raw` + keyword `离线`/`offline`/`通信中断`；② 按 `alarm_host_name` 去重列清单；要文件同上 | +| Critical Top / 告警统计 | ① `aggregateUmeAlarms(severity=critical, top_ne=20)` 或 `group_by=alarm_host_name`；② 结论；要 xlsx 再写表 | +| 当前告警有多少 / tally | ① `runUmeDiagnostics` 或 `aggregateUmeAlarms`；② 直接报 by_severity + freshness | +| 导出 Excel / 发我表格 | 先查再 `write_xlsx(sheets=[…])`（可顶层 headers+rows）→ **必须** `save_deliverable_attachment(attachment_id=…)` | + +交付约定： +- `write_xlsx` 只入库，不发出；WhatsApp 要文件时最后一步必须 `save_deliverable_attachment`。 +- 禁止用 `run_command`+openpyxl 造 xlsx。 +- 确认类短句（YES / confirm / 继续）：承接上一任务继续，勿重新开查。 + ## 约束与护栏 - 优先非 SQL；参数表达不了再用 SQL。 @@ -76,8 +94,8 @@ description: 面向 ops 专家的 netx UME 运维作业手册。覆盖告警查 ## 推荐分析模式 -- 高风险网元：`aggregateUmeAlarmsRaw` + `group_by=alarm_host_name` + `severity`。 -- 严重度：`aggregateUmeAlarms` 或 `group_by=alarm_perceived_severity`。 +- 高风险网元：`aggregateUmeAlarms(group_by=alarm_host_name, severity=critical)` 或显式 `aggregateUmeAlarmsRaw`。 +- 严重度：`aggregateUmeAlarms`（默认 by_severity + by_ne）；自定义维度才用 `group_by`。 - 事件类型：看 diagnostics `top_event_types`；真正告警码看 `top_alarm_codes`（UME `alarmCode`）。 - 关联路径：critical Port down / LOS → `findTopologyPaths(from_ume_ne_id, to_ume_ne_id)`。 diff --git a/skills/_workspace/ops/ops-netx-ume-playbook/reference.md b/skills/_workspace/ops/ops-netx-ume-playbook/reference.md index c229cae9..69af3f7a 100644 --- a/skills/_workspace/ops/ops-netx-ume-playbook/reference.md +++ b/skills/_workspace/ops/ops-netx-ume-playbook/reference.md @@ -22,8 +22,18 @@ - `aggregateUmeAlarms`：`severity`(可选，如 critical)、`top_ne`(默认50)、`exclude_missing_host`(默认true)、`time_from`/`time_to` - 高危 Top：`severity=critical`；看 `by_ne_missing`；Top 默认不含 missing + - **也可传 `group_by=alarm_host_name`**：自动走动态聚合（等同 `aggregateUmeAlarmsRaw`） - `aggregateUmeAlarmsRaw`：`group_by=alarm_host_name` 等；按 host 分组时默认排除 missing（`by_ne_missing`） +## 3b) WhatsApp 最短路径 + +| 意图 | 调用 | +|------|------| +| Critical Top | `aggregateUmeAlarms(severity=critical, top_ne=20)` | +| 按 host 统计 | `aggregateUmeAlarms(group_by=alarm_host_name, limit=50)` | +| 断纤/离线清单 | `queryUmeAlarmsRaw(keyword=…, field_preset=evidence)` → 可选 `write_xlsx` + `save_deliverable_attachment` | +| 发 Excel | `write_xlsx`（sheets 或顶层 headers+rows）→ `save_deliverable_attachment` | + ## 4) 诊断 - `runUmeDiagnostics` diff --git a/skills/_workspace/public/channel-file-delivery/SKILL.md b/skills/_workspace/public/channel-file-delivery/SKILL.md index b1c2988e..50f01f39 100644 --- a/skills/_workspace/public/channel-file-delivery/SKILL.md +++ b/skills/_workspace/public/channel-file-delivery/SKILL.md @@ -31,7 +31,8 @@ description: "在 WhatsApp/微信等渠道会话中，把生成的附件发回 | 目标格式 | 用哪个工具 | 说明 | |----------|------------|------| -| `.xlsx` Excel | **`write_xlsx`** | 传 sheets/headers/rows；返回 `attachment_id`；**禁止**再用 `run_command`+openpyxl | +| `.xlsx` Excel | **`write_xlsx`** | 传 `sheets=[{name,headers,rows}]`，或顶层 `headers`+`rows`；返回 `attachment_id`；**禁止**再用 `run_command`+openpyxl | + | `.csv`、`.txt`、`.md`、`.json` 等纯文本 | `write_file` | 只能写文本内容 | | 图片 | `cloudflare_image_generate` 等 | 生成后用 `attachment_id` 标记 | | 视频 | 对应生成工具 | 生成后用 `attachment_id` 标记 | diff --git a/tests/test_write_file_tool_aliases.py b/tests/test_write_file_tool_aliases.py new file mode 100644 index 00000000..c9c3f3e0 --- /dev/null +++ b/tests/test_write_file_tool_aliases.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +from runtime.tools.public.write_file_tool import write_file_tool +from runtime.tools.tool_validation import filter_arguments_to_schema, validate_tool_arguments + + +def test_write_file_accepts_filename_alias(tmp_path, monkeypatch) -> None: + target = tmp_path / "workspace" / "note.txt" + monkeypatch.setattr( + "runtime.tools.public.write_file_tool.resolve_workspace_path", + lambda raw: target, + ) + spec = write_file_tool() + out = spec.handler({"filename": "note.txt", "text": "hello"}) + assert out.get("ok") is True + assert target.read_text(encoding="utf-8") == "hello" + + +def test_write_file_schema_allows_aliases() -> None: + spec = write_file_tool() + args = {"file": "a.txt", "content": "x"} + filtered = filter_arguments_to_schema(spec.parameters, args) + ok, err = validate_tool_arguments(spec.parameters, filtered) + assert ok, err + assert "file" in filtered + + +def test_write_file_path_required_message() -> None: + spec = write_file_tool() + out = spec.handler({"content": "x"}) + assert out.get("ok") is False + assert out.get("error") == "path_required" + assert "example" in out diff --git a/tests/test_write_xlsx_tool.py b/tests/test_write_xlsx_tool.py index d28e06de..c63d04bf 100644 --- a/tests/test_write_xlsx_tool.py +++ b/tests/test_write_xlsx_tool.py @@ -98,6 +98,49 @@ def test_write_xlsx_requires_sheets() -> None: out = spec.handler({}) assert out.get("ok") is False assert out.get("error") == "sheets_required" + assert "example" in out + + +def test_write_xlsx_accepts_top_level_headers_rows(tmp_path, monkeypatch) -> None: + store = AttachmentAssetStore(root_dir=tmp_path / "att") + monkeypatch.setattr( + "runtime.tools.public.write_xlsx_tool.AttachmentAssetStore", + lambda root_dir=None: store if root_dir is None else AttachmentAssetStore(root_dir=root_dir), + ) + spec = write_xlsx_tool() + out = spec.handler( + { + "filename": "fiber.xlsx", + "headers": ["host_name", "count"], + "rows": [["NE-A", 3], ["NE-B", 1]], + "freeze_header": True, + "auto_width": True, + } + ) + assert out.get("ok") is True + assert out.get("name") == "fiber.xlsx" + assert out.get("sheet_count") == 1 + blob, _ = store.load_bytes(str(out["attachment_id"])) + wb = load_workbook(io.BytesIO(blob)) + ws = wb.active + assert [c.value for c in ws[1]] == ["host_name", "count"] + assert ws["A2"].value == "NE-A" + + +def test_write_xlsx_schema_allows_freeze_and_shortcut_keys() -> None: + from runtime.tools.tool_validation import filter_arguments_to_schema, validate_tool_arguments + + spec = write_xlsx_tool() + args = { + "headers": ["a"], + "rows": [[1]], + "freeze_header": True, + "auto_width": False, + "filename": "t.xlsx", + } + filtered = filter_arguments_to_schema(spec.parameters, args) + ok, err = validate_tool_arguments(spec.parameters, filtered) + assert ok, err def test_write_xlsx_tool_result_maps_to_binary_ref() -> None: