AgapHost/ai/validate_capability_grants.py

#!/usr/bin/env python3
"""validate_capability_grants — kb#147 (A2A-15), extended by kb#144 (A2A-12):
cross-check that agent-registry.yaml's declared tool_allowlist.mcp_servers
(server-level) AND tool_allowlist.mcp_tool_filter (per-tool level, kb#144)
for each agent match what that agent's real, git-controlled config actually
grants it -- at BOTH layers that materialize a tool bundle for Adolf:

  1. OpenClaw's own mcp.servers.*.toolFilter.include in adolf/openclaw.json
     (adolf's own MCP client surface).
  2. shared-mcp.json's per-server `enabledTools` (kb#144 verification pass,
     2026-07-22): what adolf-llm/server.js's writeMcpConfig() seeds into
     each Kimi CLI session's project-root .mcp.json -- the layer that
     ACTUALLY determines the MODEL's tool bundle for Adolf's kimi backbone.
     Layer 1 alone shipped a false "Done" once already (kb#144 first pass):
     wire.jsonl proved Kimi's tool counts were unchanged because OpenClaw's
     toolFilter never reaches the Kimi CLI, which reads its own
     enabledTools/disabledTools (McpServerCommonFields, computeEnabledNames
     -- confirmed by decompiling the installed @moonshot-ai/kimi-code
     package's dist/main.mjs). Checking only layer 1 would pass this
     validator while leaving the real per-turn token bloat unfixed again.

Read-only. Makes no live changes and touches no running service — it just
diffs already-committed files so a registry edit that silently drifts from
an agent's real config fails loudly (exit 1) instead of rotting quietly,
which is exactly the "scattered configs" failure mode kb#147's acceptance
bar ("grants live in the agent registry, not scattered configs") exists to
prevent. kb#144's acceptance bar ("Adolf's tools are sourced from the
registry") means both layers, not just the one OpenClaw itself reads.

Only agents with a `prompt_source` pointing at a real openclaw.json-shaped
config are checked; agents that are registry-only target state (torgash,
researcher — no config file exists yet) are reported as skipped, not failed.

Usage:
    ./validate_capability_grants.py
    ./validate_capability_grants.py --openclaw-json ../adolf/openclaw.json --id adolf
"""

import argparse
import json
import re
import sys

import agent_registry as ar

HERE_ADOLF_OPENCLAW_JSON = "../adolf/openclaw.json"
SHARED_MCP_JSON = "shared-mcp.json"

# id -> path to the git-controlled OpenClaw config that IS this agent's live
# MCP surface. Only adolf has one today (claude-coder has no openclaw.json --
# it's a CLAUDE.md-driven persona, not an OpenClaw runtime; see its
# capability_grant note in agent-registry.yaml).
KNOWN_CONFIGS = {
    "adolf": HERE_ADOLF_OPENCLAW_JSON,
}

# id -> path to the shared-mcp.json this agent's backbone runtime seeds its
# session .mcp.json from (kb#144 layer-2 check, see module docstring). Only
# agents on a Kimi-CLI-shaped backbone go through this file at all.
KNOWN_SHARED_MCP = {
    "adolf": SHARED_MCP_JSON,
}


def _strip_jsonc_comments(text):
    """Drop // line comments. Good enough for this read-only check: this
    file's comments are all on their own line or trail real content with no
    '//' inside a string value today -- verified by hand."""
    out = []
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("//"):
            continue
        m = re.search(r'(?<!:)//', line)
        if m and line[: m.start()].count('"') % 2 == 0:
            line = line[: m.start()]
        out.append(line)
    return "\n".join(out)


def _find_key_brace(text, key, start=0):
    """Find `key: {` (bare or quoted key) at or after `start`; return the
    index of the matching '{'."""
    m = re.search(rf'["\']?{re.escape(key)}["\']?\s*:\s*\{{', text[start:])
    if not m:
        raise ValueError(f"key {key!r} not found from offset {start}")
    return start + m.end() - 1  # index of the '{' itself


def _matching_close_brace(text, open_idx):
    depth = 0
    for i in range(open_idx, len(text)):
        if text[i] == '{':
            depth += 1
        elif text[i] == '}':
            depth -= 1
            if depth == 0:
                return i
    raise ValueError("unbalanced braces")


def _extract_object_top_level_keys(text, key_path):
    """openclaw.json is JS-object-literal JSON5 (bare identifier keys,
    trailing commas) -- `json.loads` can't touch it and pulling in a JSON5
    parser is overkill for one narrow read. Instead: locate `key_path`
    (e.g. ["mcp", "servers"]) by finding each key's opening '{' in turn, then
    brace-depth-scan that object collecting only its DIRECT child keys
    (`name: {` at depth 1). Sufficient and honest for this validation
    script's one job; not a general JSON5 reader."""
    pos = 0
    open_idx = 0
    for key in key_path:
        open_idx = _find_key_brace(text, key, pos)
        pos = open_idx + 1
    close_idx = _matching_close_brace(text, open_idx)
    body = text[open_idx + 1 : close_idx]

    # Depth-0 identifiers immediately followed by ": {" are this object's
    # direct child keys (mcp.servers' entries are always objects). Only try
    # a key match right after a boundary ('{', ',', or start-of-body) so an
    # identifier can't be "matched" starting mid-token from inside a nested
    # value (the earlier, buggy version of this scanner did exactly that).
    keys = []
    depth = 0
    i, n = 0, len(body)
    prev_boundary = True
    key_re = re.compile(r'["\']?([A-Za-z0-9_-]+)["\']?\s*:\s*\{')
    while i < n:
        ch = body[i]
        if ch in ' \t\r\n':
            i += 1
            continue
        if depth == 0 and prev_boundary:
            m = key_re.match(body, i)
            if m:
                keys.append(m.group(1))
                i = m.end() - 1  # land on the key's '{' so the normal handling below opens depth 1
                prev_boundary = False
                continue
        if ch == '{':
            depth += 1
            prev_boundary = True
        elif ch == '}':
            depth -= 1
            prev_boundary = True
        elif ch == ',':
            prev_boundary = True
        else:
            prev_boundary = False
        i += 1
    return sorted(keys)


def load_mcp_servers(path):
    with open(path) as f:
        raw = f.read()
    text = _strip_jsonc_comments(raw)
    return _extract_object_top_level_keys(text, ["mcp", "servers"])


def _locate_object(text, key_path, start=0):
    """Chase `key_path` (e.g. ["mcp", "servers", "hindsight"]) through nested
    `key: {` objects, same navigation _extract_object_top_level_keys does
    internally, exposed standalone so other extractors (toolFilter below)
    can reuse it instead of re-deriving brace offsets."""
    pos = start
    open_idx = start
    for key in key_path:
        open_idx = _find_key_brace(text, key, pos)
        pos = open_idx + 1
    close_idx = _matching_close_brace(text, open_idx)
    return open_idx, close_idx


def _matching_close_bracket(text, open_idx):
    """Same brace-depth-scan as _matching_close_brace, for '[' / ']' — needed
    to bound a toolFilter.include array (a list, not an object)."""
    depth = 0
    for i in range(open_idx, len(text)):
        if text[i] == '[':
            depth += 1
        elif text[i] == ']':
            depth -= 1
            if depth == 0:
                return i
    raise ValueError("unbalanced brackets")


def load_tool_filter(path, server_name):
    """kb#144: extract mcp.servers.<server_name>.toolFilter.include as a
    sorted list of tool names, or None if that server has no toolFilter (or
    no include list) at all — OpenClaw's own semantics for "no toolFilter":
    every tool the server offers stays eligible (see schema-BqdpWz19.js:
    "When omitted, all server tools remain eligible unless excluded.").
    exclude-only filters are not modeled here (none of Adolf's servers use
    exclude today) and are reported as None (unrestricted) rather than
    silently mis-parsed.
    """
    with open(path) as f:
        raw = f.read()
    text = _strip_jsonc_comments(raw)
    try:
        server_open, server_close = _locate_object(text, ["mcp", "servers", server_name])
    except ValueError:
        return None  # server not present in this config at all
    body = text[server_open : server_close + 1]
    try:
        tf_open, tf_close = _locate_object(body, ["toolFilter"])
    except ValueError:
        return None  # no toolFilter -> unrestricted, by OpenClaw's own semantics
    tf_body = body[tf_open : tf_close + 1]
    m = re.search(r'["\']?include["\']?\s*:\s*\[', tf_body)
    if not m:
        return None  # exclude-only or empty toolFilter -- not modeled, treat as unrestricted
    bracket_open = tf_body.index('[', m.start())
    bracket_close = _matching_close_bracket(tf_body, bracket_open)
    arr_body = tf_body[bracket_open + 1 : bracket_close]
    return sorted(re.findall(r'["\']([A-Za-z0-9_.\-\*]+)["\']', arr_body))


def load_shared_mcp_enabled_tools(path):
    """kb#144 layer-2 check (see module docstring): shared-mcp.json is
    strict JSON (no JSON5 quirks, unlike openclaw.json), so a plain
    `json.load` is enough -- no brace-scanner needed here. Returns
    {server_name: sorted-tool-list-or-None}, None meaning no `enabledTools`
    key on that server (unfiltered -- every tool it offers stays eligible,
    same "omitted = unrestricted" semantics as OpenClaw's own toolFilter).
    """
    with open(path) as f:
        data = json.load(f)
    out = {}
    for name, cfg in (data.get("mcpServers") or {}).items():
        tools = cfg.get("enabledTools")
        out[name] = sorted(tools) if tools else None
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--registry", default=None)
    ap.add_argument("--openclaw-json", default=None, help="override path for --id's config")
    ap.add_argument("--id", default=None, help="only this agent id (default: every id in KNOWN_CONFIGS)")
    args = ap.parse_args()

    reg = ar.load_registry(args.registry)
    ids = [args.id] if args.id else list(KNOWN_CONFIGS)

    failures = 0
    for agent_id in ids:
        agent = ar.get_agent(reg, agent_id)
        declared = sorted((agent.get("tool_allowlist") or {}).get("mcp_servers") or [])
        config_path = args.openclaw_json or KNOWN_CONFIGS.get(agent_id)
        if not config_path:
            print(f"SKIP {agent_id}: no known live config to cross-check (target-state agent)")
            continue
        try:
            live = load_mcp_servers(config_path)
        except FileNotFoundError:
            print(f"SKIP {agent_id}: config not found at {config_path}")
            continue
        if declared == live:
            print(f"OK   {agent_id}: registry tool_allowlist.mcp_servers == live {config_path} mcp.servers -> {live}")
        else:
            failures += 1
            print(f"FAIL {agent_id}: registry says {declared} but {config_path} actually grants {live}")

        # kb#144: per-tool cross-check, same idea one level down. Only
        # meaningful for servers the registry actually declares a filter
        # for (mcp_tool_filter); a server absent from that map is not
        # asserted either way here (it may be intentionally unfiltered).
        declared_filters = (agent.get("tool_allowlist") or {}).get("mcp_tool_filter") or {}
        for server_name, declared_tools in declared_filters.items():
            live_tools = load_tool_filter(config_path, server_name)
            declared_sorted = sorted(declared_tools) if declared_tools else None
            if declared_sorted == live_tools:
                shown = live_tools if live_tools is not None else "(unfiltered)"
                print(f"OK   {agent_id}/{server_name}: registry mcp_tool_filter == live toolFilter.include -> {shown}")
            else:
                failures += 1
                print(f"FAIL {agent_id}/{server_name}: registry mcp_tool_filter says {declared_sorted} but live toolFilter.include is {live_tools}")

        # kb#144 layer-2: the file that actually reaches the MODEL for a
        # Kimi-backed agent (see module docstring for why layer 1 alone
        # missed the real bug once already). Servers the registry declares a
        # filter for but that don't appear in shared-mcp.json at all (e.g.
        # marketplace, which OpenClaw carries but Kimi's session never sees)
        # are not asserted here -- that's a separate, pre-existing gap
        # between what OpenClaw offers Adolf and what reaches Kimi, not a
        # drift this validator's job to catch.
        shared_mcp_path = KNOWN_SHARED_MCP.get(agent_id)
        if shared_mcp_path:
            try:
                live_shared = load_shared_mcp_enabled_tools(shared_mcp_path)
            except FileNotFoundError:
                print(f"SKIP {agent_id}: shared-mcp.json not found at {shared_mcp_path}")
            else:
                for server_name, declared_tools in declared_filters.items():
                    if server_name not in live_shared:
                        continue
                    declared_sorted = sorted(declared_tools) if declared_tools else None
                    live_tools = live_shared[server_name]
                    if declared_sorted == live_tools:
                        shown = live_tools if live_tools is not None else "(unfiltered)"
                        print(f"OK   {agent_id}/{server_name}: registry mcp_tool_filter == live shared-mcp.json enabledTools -> {shown}")
                    else:
                        failures += 1
                        print(f"FAIL {agent_id}/{server_name}: registry mcp_tool_filter says {declared_sorted} but shared-mcp.json enabledTools is {live_tools}")

    sys.exit(1 if failures else 0)


if __name__ == "__main__":
    main()