Blame for harness_normalize.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Map any dataset tool call/result onto HARNESS_TOOLS Hermes shapes — or drop."""

from __future__ import annotations

import json
import re
from typing import Any

from hermes_agent_eval import hermes_tool_content
from tool_runtime import HARNESS_TOOLS

HARNESS_NAMES = frozenset(
    (t.get("function") or t).get("name")
    for t in HARNESS_TOOLS
    if (t.get("function") or t).get("name")
)

_SEARCH_RE = re.compile(
    r"(search|news|web|google|bing|wiki|lookup|find_info|get_news|whois|stock|movie|image|restaurant)",
    re.I,
)
_CALC_RE = re.compile(
    r"(calc|math|compute|convert|currency|temperature|loan|mortgage|distance|random_number|evaluate)",
    re.I,
)
_VAULT_READ_RE = re.compile(r"(vault_read|read_file|read_note|get_file|open_file)", re.I)
_VAULT_WRITE_RE = re.compile(r"(vault_write|write_file|save_note|create_file|write_note)", re.I)
_VAULT_SEARCH_RE = re.compile(r"(vault_search|search_notes|search_vault|grep_vault)", re.I)


def _as_dict(payload: Any) -> dict | None:
    if isinstance(payload, dict):
        return payload
    if isinstance(payload, str):
        try:
            obj = json.loads(payload)
        except (json.JSONDecodeError, TypeError):
            return None
        return obj if isinstance(obj, dict) else None
    return None


def _query_from_args(args: dict, human: str) -> str:
    for k in ("query", "q", "keywords", "search_query", "stock_symbol", "company", "title", "domain"):
        v = args.get(k)
        if isinstance(v, list):
            v = " ".join(str(x) for x in v if x)
        if isinstance(v, str) and v.strip():
            return v.strip()
    h = (human or "").strip()
    return h[:120] if h else "search"


def _hits_from_payload(payload: dict) -> list[dict]:
    """Build {title,url,snippet} list from common alien shapes."""
    data = payload.get("data")
    if isinstance(data, dict) and isinstance(data.get("web"), list):
        return []  # already Hermes — hermes_tool_content handles
    for key in ("results", "articles", "movies", "images", "hits", "items"):
        raw = payload.get(key)
        if not isinstance(raw, list) or not raw:
            continue
        hits = []
        for i, item in enumerate(raw):
            if isinstance(item, str):
                hits.append({"title": item, "snippet": item, "url": ""})
                continue
            if not isinstance(item, dict):
                continue
            title = (
                item.get("title")
                or item.get("name")
                or item.get("headline")
                or f"result {i + 1}"
            )
            desc = (
                item.get("description")
                or item.get("snippet")
                or item.get("excerpt")
                or item.get("summary")
                or str(title)
            )
            url = item.get("url") or item.get("link") or ""
            if url and not str(url).startswith(("http://", "https://")):
                url = "https://" + str(url).lstrip("/")
            hits.append({"title": str(title), "snippet": str(desc), "url": str(url)})
        if hits:
            return hits
    # flat scalar bags (stock_price, etc.)
    if len(payload) <= 6:
        parts = []
        for k, v in payload.items():
            if k in ("error", "success", "ok", "data"):
                continue
            if isinstance(v, (str, int, float)):
                parts.append(f"{k}: {v}")
            elif isinstance(v, list) and v and all(isinstance(x, (str, int, float)) for x in v[:5]):
                parts.append(f"{k}: {', '.join(str(x) for x in v[:5])}")
        if parts:
            text = "; ".join(parts)
            return [{"title": text[:80], "snippet": text, "url": ""}]
    return []


def _to_web_search(name: str, args: dict, payload: Any, human: str) -> tuple[str, dict, dict] | None:
    p = _as_dict(payload)
    if p is None:
        return None
    query = _query_from_args(args or {}, human)
    nargs = {"query": query, "limit": 5}
    data = p.get("data")
    if isinstance(data, dict) and isinstance(data.get("web"), list):
        content = hermes_tool_content("web_search", p)
        return "web_search", nargs, content
    hits = _hits_from_payload(p)
    if not hits:
        return None
    content = hermes_tool_content("web_search", {"results": hits})
    if not (content.get("data") or {}).get("web"):
        return None
    return "web_search", nargs, content


def _to_calculator(args: dict, payload: Any, human: str) -> tuple[str, dict, dict] | None:
    p = _as_dict(payload)
    if p is None:
        return None
    args = args or {}
    expr = (
        args.get("expression")
        or args.get("expr")
        or p.get("expression")
        or p.get("expr")
    )
    result = p.get("result")
    if result is None:
        for k in (
            "value",
            "output",
            "answer",
            "converted",
            "converted_amount",
            "converted_temperature",
            "converted_length",
            "exchange_rate",
            "stock_price",
            "gcd",
            "lcm",
        ):
            if k in p:
                result = p[k]
                break
    if expr is None:
        if "amount" in args and ("from_currency" in args or "from" in args):
            fr = args.get("from_currency") or args.get("from")
            to = args.get("to_currency") or args.get("to")
            expr = f"convert({args['amount']}, {fr!r}, {to!r})"
        elif "temperature" in args:
            expr = (
                f"convert({args['temperature']}, "
                f"{args.get('from_unit')!r}, {args.get('to_unit')!r})"
            )
        elif "length" in args:
            expr = (
                f"convert({args['length']}, "
                f"{args.get('from_unit')!r}, {args.get('to_unit')!r})"
            )
        elif "base_currency" in args and "target_currency" in args:
            expr = f"fx({args['base_currency']!r}, {args['target_currency']!r})"
        elif "celsius" in args:
            expr = f"convert({args['celsius']}, 'C', 'F')"
    if expr is None or result is None:
        return None
    try:
        if isinstance(result, str) and result.strip().startswith("$"):
            result = float(re.sub(r"[^\d.]", "", result) or "nan")
    except (TypeError, ValueError):
        pass
    content = hermes_tool_content(
        "calculator", {"expression": str(expr), "result": result}
    )
    return "calculator", {"expression": str(expr)}, content


def _to_vault(
    kind: str, args: dict, payload: Any
) -> tuple[str, dict, dict] | None:
    p = _as_dict(payload)
    if p is None:
        return None
    if kind == "vault_read":
        path = (args or {}).get("path") or p.get("path") or "notes/note.md"
        content = hermes_tool_content(
            "vault_read",
            {
                "path": str(path),
                "content": p.get("content") if "content" in p else json.dumps(p)[:800],
                "bytes": int(p.get("bytes") or p.get("total_lines") or len(str(p.get("content") or ""))),
            },
        )
        return "vault_read", {"path": str(path)}, content
    if kind == "vault_write":
        path = (args or {}).get("path") or p.get("path") or "notes/note.md"
        body = (args or {}).get("content") or ""
        content = hermes_tool_content(
            "vault_write",
            {
                "ok": True,
                "path": str(path),
                "bytes": int(p.get("bytes") or p.get("bytes_written") or len(body)),
            },
        )
        nargs = {"path": str(path), "content": str(body or f"note at {path}")}
        if "append" in (args or {}):
            nargs["append"] = bool(args["append"])
        return "vault_write", nargs, content
    # vault_search
    q = (args or {}).get("query") or p.get("query") or "notes"
    hits = p.get("hits") if isinstance(p.get("hits"), list) else []
    if not hits and isinstance(p.get("results"), list):
        hits = [
            {
                "path": (h.get("path") if isinstance(h, dict) else "notes/x.md"),
                "score": (h.get("score") if isinstance(h, dict) else 0.5),
                "excerpt": (
                    h.get("excerpt")
                    or h.get("snippet")
                    or h.get("title")
                    or str(h)
                )
                if isinstance(h, dict)
                else str(h),
            }
            for h in p["results"][:5]
        ]
    content = hermes_tool_content(
        "vault_search", {"hits": hits, "query": str(q)}
    )
    return "vault_search", {"query": str(q)}, content


def classify_harness_tool(name: str) -> str | None:
    n = (name or "").strip()
    if n in HARNESS_NAMES:
        return n
    if _VAULT_READ_RE.search(n):
        return "vault_read"
    if _VAULT_WRITE_RE.search(n):
        return "vault_write"
    if _VAULT_SEARCH_RE.search(n):
        return "vault_search"
    if _CALC_RE.search(n):
        return "calculator"
    if _SEARCH_RE.search(n):
        return "web_search"
    return None


def normalize_to_harness(
    *,
    name: str,
    args: dict | None,
    payload: Any,
    human: str = "",
    gold: str = "",
) -> dict | None:
    """Return {name, args, payload, gold} in HARNESS Hermes shapes, or None to drop."""
    args = dict(args or {})
    target = classify_harness_tool(name)
    if not target:
        return None
    if target == "web_search":
        mapped = _to_web_search(name, args, payload, human)
    elif target == "calculator":
        mapped = _to_calculator(args, payload, human)
    elif target in ("vault_read", "vault_write", "vault_search"):
        mapped = _to_vault(target, args, payload)
    else:
        return None
    if mapped is None:
        return None
    nname, nargs, npayload = mapped
    # already-canonical harness name: still force hermes_tool_content
    npayload = hermes_tool_content(nname, npayload)
    g = (gold or "").strip()
    if not g:
        return None
    return {"name": nname, "args": nargs, "payload": npayload, "gold": g}


def self_check():
    from hermes_agent_eval import hermes_tool_content as htc

    m = normalize_to_harness(
        name="search_news",
        args={"keywords": ["technology"], "category": "technology"},
        payload={
            "articles": [
                {"title": "Tech Giants", "url": "www.technews.com/a"},
                {"title": "AI News", "url": "www.technews.com/b"},
            ]
        },
        human="latest tech news",
        gold="Here are tech articles.",
    )
    assert m is not None
    assert m["name"] == "web_search"
    assert m["args"]["query"] == "technology"
    assert m["payload"]["success"] is True
    assert len(m["payload"]["data"]["web"]) == 2
    assert m["payload"]["data"]["web"][0]["url"].startswith("https://")

    m2 = normalize_to_harness(
        name="get_stock_price",
        args={"company": "Apple"},
        payload={"stock_price": "$150.75"},
        human="Apple price?",
        gold="Apple is $150.75.",
    )
    assert m2 and m2["name"] == "web_search"
    assert "150.75" in m2["payload"]["data"]["web"][0]["description"]

    m3 = normalize_to_harness(
        name="calculator",
        args={"expression": "2+2"},
        payload={"expression": "2+2", "result": 4},
        human="2+2?",
        gold="4",
    )
    assert m3 and m3["name"] == "calculator"
    assert m3["payload"] == htc("calculator", {"expression": "2+2", "result": 4})

    assert (
        normalize_to_harness(
            name="send_email",
            args={},
            payload={"ok": True},
            human="mail",
            gold="sent",
        )
        is None
    )
    print("harness_normalize self_check ok")


if __name__ == "__main__":
    self_check()