feat: harden Qwen agentic tool orchestration

platform/mcp/patches/hass_mcp/yaml_config.py

@@ -41,6 +41,7 @@ _OPS = (
     "get",
     "read_source",
     "find_source",
+    "find_commented_blocks",
     "list_backups",
     "create",
     "update",
@@ -66,8 +67,9 @@ _SECRET_REFERENCE = re.compile(r"!secret\s+[^\s#]+", re.IGNORECASE)
         "Safely inspect and edit Home Assistant YAML. Structured CRUD is limited to "
         "automations.yaml, scripts.yaml and scenes.yaml. Raw source operations also "
         "allow configuration.yaml so commented-out blocks can be found and reviewed. "
-        "secrets.yaml and arbitrary paths are impossible. Use read_source/find_source "
-        "for comments or exact YAML text. Every mutation first returns a diff/preview "
+        "secrets.yaml and arbitrary paths are impossible. Use find_commented_blocks once "
+        "to inventory fully commented YAML entries; use read_source/find_source only for "
+        "other comments or exact YAML text. Every mutation first returns a diff/preview "
         "and one-time approval_ticket; only repeat the exact unchanged call with "
         "confirm=true after explicit user approval. Writes create a backup, are atomic, "
         "run Home Assistant config validation, roll back on failure, and reload the "
@@ -139,7 +141,7 @@ _SECRET_REFERENCE = re.compile(r"!secret\s+[^\s#]+", re.IGNORECASE)
     requires_admin=True,
     write_ops=["create", "update", "replace_source_text", "reload"],
     destructive_ops=["delete", "restore_backup"],
-    admin_ops=["list", "get", "read_source", "find_source", "list_backups"],
+    admin_ops=["list", "get", "read_source", "find_source", "find_commented_blocks", "list_backups"],
 )
 async def ha_yaml_config(
     hass: HomeAssistant,
@@ -178,6 +180,10 @@ async def ha_yaml_config(
         if not query:
             raise ToolError("op=find_source requires query")
         return await _find_source(hass, path, query, max_lines)
+    if op == "find_commented_blocks":
+        if kind not in {"automation", "scene"}:
+            raise ToolError("find_commented_blocks supports automation and scene list files")
+        return await _find_commented_blocks(hass, path, max_lines)
     if op == "list_backups":
         return await _list_backups(hass, filename, limit, offset)
     if op == "replace_source_text":
@@ -387,6 +393,86 @@ async def _find_source(hass: HomeAssistant, path: Path, query: str, max_lines: i
     }
 
 
+def _extract_commented_blocks(text: str, max_lines: int) -> tuple[list[dict[str, Any]], bool]:
+    """Return top-level YAML list entries whose every source line is commented.
+
+    This intentionally recognizes only the conservative ``# - id:`` form used
+    by Home Assistant's automations/scenes editor. Ordinary prose comments,
+    partially disabled entries and nested comments are not treated as entries.
+    """
+    lines = text.splitlines()
+    start_pattern = re.compile(r"^\s*#\s*-\s+id\s*:\s*(.*?)\s*$", re.IGNORECASE)
+    alias_pattern = re.compile(r"^\s*#\s+alias\s*:\s*(.*?)\s*$", re.IGNORECASE)
+    blocks: list[dict[str, Any]] = []
+    consumed = 0
+    index = 0
+    truncated = False
+
+    def clean_scalar(value: str) -> str:
+        value = value.strip()
+        if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
+            return value[1:-1]
+        return value
+
+    while index < len(lines):
+        match = start_pattern.match(lines[index])
+        if not match:
+            index += 1
+            continue
+        start = index
+        block_lines = [lines[index]]
+        index += 1
+        while index < len(lines):
+            if start_pattern.match(lines[index]):
+                break
+            if not lines[index].strip() or not re.match(r"^\s*#", lines[index]):
+                break
+            block_lines.append(lines[index])
+            index += 1
+        if consumed + len(block_lines) > max_lines:
+            truncated = True
+            break
+        alias = None
+        for line in block_lines:
+            alias_match = alias_pattern.match(line)
+            if alias_match:
+                alias = clean_scalar(alias_match.group(1))
+                break
+        blocks.append(
+            {
+                "start_line": start + 1,
+                "end_line": start + len(block_lines),
+                "id": clean_scalar(match.group(1)),
+                "alias": alias,
+                "source": [
+                    {"line": start + offset + 1, "text": _redact_line(line)}
+                    for offset, line in enumerate(block_lines)
+                ],
+            }
+        )
+        consumed += len(block_lines)
+    return blocks, truncated
+
+
+async def _find_commented_blocks(
+    hass: HomeAssistant, path: Path, max_lines: int
+) -> dict[str, Any]:
+    text = await _read_text(hass, path)
+    cap = max(1, min(max_lines, 400))
+    blocks, truncated = _extract_commented_blocks(text, cap)
+    return {
+        "file": path.name,
+        "sha256": _fingerprint(text),
+        "authoritative_block_count": len(blocks) if not truncated else None,
+        "returned_block_count": len(blocks),
+        "returned_source_lines": sum(len(block["source"]) for block in blocks),
+        "has_more": truncated,
+        "blocks": blocks,
+        "recognition_rule": "Only fully commented top-level '# - id:' YAML list entries are returned.",
+        "redaction_note": "Credential-like values and !secret reference names are redacted.",
+    }
+
+
 def _source_diff(filename: str, before: str, after: str) -> list[str]:
     return list(
         difflib.unified_diff(