Nick
59efd385db
Port the five audit-unmatched helpers name-preserving into the Go crate (gaps.go): _utc_now_iso (Python isoformat law — six-digit fraction iff nonzero, injectable clock seam pinned by the battery to the frozen oracle instants), _contains_tool_object_key (name-preserving twin of the outside-strings scanner), to_ollama_tools (registry → Ollama function shape), dispatch_validated_tool (validate-before-dispatch with an Executor seam; invalid calls provably never reach the executor), tool_requires_approval (fail-closed against the 12-name READ_ONLY frozenset; NOT derived from registry risk — lint_file runs a linter). Truth 17→32 cases (utc both instants, 8 scanner cases, full-registry ollama conversion, dispatch alias/invalid, approval verdict for every registered tool + alias + unknown with schema-valid minimal args); regen x2 identical modulo the pre-existing uuid call_id quirk. DEFECT FOUND+FIXED: the landed containsToolObjectKey reused ONE json.Decoder across candidates (wrong offsets after the first decode) and dropped the oracle's colon check — the new frozen true-positive cases exposed it; rewritten to the oracle's per-candidate raw-decode + colon law (fix-first, proof = battery green). go test/vet/gofmt/build clean; pytest 60+29 subtests both pins. tool_protocol 22/27 → 27/27 fn-level.
207 lines
9.4 KiB
Python
207 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
|
"""gen_tool_protocol_truth.py — factory batch 1 truth battery
|
|
(contract BUILDER-2026-10-07T22:11:52Z-9ffb73).
|
|
|
|
Live tool_protocol oracle fixtures: validate error texts, alias
|
|
canonicalization, ollama envelope normalization (incl. string-arguments
|
|
JSON), internal-protocol detection, extract_visible_text span stripping.
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(REPO / "src"))
|
|
sys.path.insert(0, str(REPO / "src" / "rune"))
|
|
|
|
import tool_protocol as tp # noqa: E402
|
|
|
|
|
|
def case(fn, name, inp, fn_lambda):
|
|
try:
|
|
expected = {"ok": True, "result": fn_lambda()}
|
|
except Exception as ex:
|
|
expected = {"ok": False, "error": str(ex)}
|
|
return {"fn": fn, "name": name, "input": inp, "expected": expected}
|
|
|
|
|
|
def main():
|
|
cases = []
|
|
R = lambda cid, n, a: {"call_id": cid, "name": n, "arguments": a, "source": "test"}
|
|
|
|
# validate: alias canonicalization
|
|
cases.append(case("validate", "alias_grep",
|
|
{"request": R("c1", "grep", {"pattern": "x", "path": "."})},
|
|
lambda: tp.validate_tool_request(tp.ToolRequest(**R("c1", "grep", {"pattern": "x", "path": "."}))).to_dict()))
|
|
cases.append(case("validate", "alias_bash",
|
|
{"request": R("c2", "bash", {"command": "ls"})},
|
|
lambda: tp.validate_tool_request(tp.ToolRequest(**R("c2", "bash", {"command": "ls"}))).to_dict()))
|
|
cases.append(case("validate", "unknown_tool",
|
|
{"request": R("c3", "nonexistent_tool", {})},
|
|
lambda: tp.validate_tool_request(tp.ToolRequest(**R("c3", "nonexistent_tool", {}))).to_dict()))
|
|
cases.append(case("validate", "missing_required",
|
|
{"request": R("c4", "read_file", {})},
|
|
lambda: tp.validate_tool_request(tp.ToolRequest(**R("c4", "read_file", {}))).to_dict()))
|
|
cases.append(case("validate", "unknown_arg",
|
|
{"request": R("c5", "read_file", {"path": "/tmp", "bogus": 1})},
|
|
lambda: tp.validate_tool_request(tp.ToolRequest(**R("c5", "read_file", {"path": "/tmp", "bogus": 1}))).to_dict()))
|
|
cases.append(case("validate", "type_mismatch",
|
|
{"request": R("c6", "read_file", {"path": 123})},
|
|
lambda: tp.validate_tool_request(tp.ToolRequest(**R("c6", "read_file", {"path": 123}))).to_dict()))
|
|
|
|
# ollama normalization: string arguments JSON
|
|
cases.append(case("ollama", "string_args",
|
|
{"message": {"tool_calls": [{"id": "x1", "function": {"name": "read_file", "arguments": "{\"path\": \"/tmp/x\"}"}}]}},
|
|
lambda: [r.to_dict() for r in tp.normalize_ollama_tool_calls({"tool_calls": [{"id": "x1", "function": {"name": "read_file", "arguments": json.dumps({"path": "/tmp/x"})}}]})]))
|
|
cases.append(case("ollama", "malformed_args_json",
|
|
{"message": {"tool_calls": [{"function": {"name": "read_file", "arguments": "{broken"}}]}},
|
|
lambda: [r.to_dict() for r in tp.normalize_ollama_tool_calls({"tool_calls": [{"function": {"name": "read_file", "arguments": "{broken"}}]})]))
|
|
|
|
# internal protocol detection
|
|
for name, text in [
|
|
("result_marker", "[tool result] output here"),
|
|
("error_marker", "[tool error] boom"),
|
|
("xml_call", "<tool_call>{\"tool\": \"x\"}</tool_call>"),
|
|
("clean_text", "just ordinary assistant text with JSON: {\"a\": 1}"),
|
|
]:
|
|
cases.append(case("internal", name, {"text": text},
|
|
lambda t=text: {"contains": tp.contains_internal_protocol(t)}))
|
|
|
|
# extract_visible_text
|
|
for name, text in [
|
|
("inline_call", 'before {\"tool\": \"read_file\", \"args\": {\"path\": \"/tmp\"}} after'),
|
|
("fenced_call", '```json\n{\"tool\": \"read_file\", \"args\": {\"path\": \"/tmp\"}}\n```'),
|
|
("xml_call", 'before <tool_call>{\"tool\": \"x\", \"args\": {}}</tool_call> after'),
|
|
("plain", "no protocol here at all"),
|
|
]:
|
|
cases.append(case("visible", name, {"text": text},
|
|
lambda t=text: {"visible": tp.extract_visible_text(t)}))
|
|
|
|
# normalize_text_tool_calls end-to-end
|
|
cases.append(case("normalize", "text_fallback",
|
|
{"text": '{\"tool\": \"read_file\", \"args\": {\"path\": \"/tmp/x\"}}'},
|
|
lambda: [r.to_dict() for r in tp.normalize_text_tool_calls(
|
|
'{\"tool\": \"read_file\", \"args\": {\"path\": \"/tmp/x\"}}')]))
|
|
|
|
# ── literal-100% residue: the five gap fns ────────────────────────────
|
|
from datetime import datetime as _dt, timezone as _tz
|
|
|
|
_frozen_cell = {"instant": None}
|
|
|
|
class _Frozen(_dt):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
base = _frozen_cell["instant"]
|
|
return base.replace(tzinfo=tz) if tz is not None else base
|
|
|
|
def _pin_utc_now(instant):
|
|
_frozen_cell["instant"] = instant
|
|
tp.datetime = _Frozen
|
|
|
|
# _utc_now_iso: Python isoformat law — six-digit fraction when
|
|
# nonzero, omitted when zero (two frozen instants).
|
|
_pin_utc_now(_dt(2026, 10, 15, 3, 24, 36, 123456, tzinfo=_tz.utc))
|
|
cases.append(case("utc_now_iso", "micros_nonzero",
|
|
{"instant": "2026-10-15T03:24:36.123456+00:00"},
|
|
lambda: tp._utc_now_iso()))
|
|
_pin_utc_now(_dt(2026, 10, 15, 3, 24, 37, 0, tzinfo=_tz.utc))
|
|
cases.append(case("utc_now_iso", "micros_zero_omitted",
|
|
{"instant": "2026-10-15T03:24:37+00:00"},
|
|
lambda: tp._utc_now_iso()))
|
|
|
|
# _contains_tool_object_key: literal/escaped key outside strings
|
|
for name, text in [
|
|
("literal_key", '{"tool": "read_file", "args": {}}'),
|
|
("spaced_key", '{ "tool" : "x" }'),
|
|
("escaped_key", '{"\\u0074ool": "x"}'),
|
|
("value_not_key", '{"name": "tool"}'),
|
|
("inside_string", 'he said {"tool": "x"} aloud'),
|
|
("not_a_key", '{"toolx": 1}'),
|
|
("colon_missing", '{"tool" "x"}'),
|
|
("empty", ''),
|
|
]:
|
|
cases.append(case("contains_key", name, {"text": text},
|
|
lambda t=text: {"contains": tp._contains_tool_object_key(t)}))
|
|
|
|
# to_ollama_tools: the full registry through the converter
|
|
cases.append(case("ollama_tools", "full_registry",
|
|
{"definitions": json.loads(json.dumps(tp.TOOLS))},
|
|
lambda: tp.to_ollama_tools(tp.TOOLS)))
|
|
cases.append(case("ollama_tools", "empty",
|
|
{"definitions": []},
|
|
lambda: tp.to_ollama_tools([])))
|
|
|
|
# dispatch_validated_tool: the alias canonicalizes BEFORE dispatch;
|
|
# invalid never reaches the executor
|
|
def _recorder(calls):
|
|
def executor(name, args):
|
|
calls.append({"name": name, "args": args})
|
|
return {"ran": name}
|
|
return executor
|
|
|
|
calls = []
|
|
cases.append(case("dispatch", "alias_canonicalized",
|
|
{"request": R("d1", "grep", {"pattern": "x", "path": "."})},
|
|
lambda: (tp.dispatch_validated_tool(
|
|
tp.ToolRequest(**R("d1", "grep", {"pattern": "x", "path": "."})),
|
|
_recorder(calls)), calls)))
|
|
calls2 = []
|
|
cases.append(case("dispatch", "invalid_never_reaches_executor",
|
|
{"request": R("d2", "nonexistent_tool", {})},
|
|
lambda: (tp.dispatch_validated_tool(
|
|
tp.ToolRequest(**R("d2", "nonexistent_tool", {})),
|
|
_recorder(calls2)), calls2)))
|
|
|
|
# tool_requires_approval: every registered tool (schema-valid minimal
|
|
# args so read-only tools reach a REAL verdict, not a required-arg
|
|
# error) + the bash alias + an unknown tool
|
|
def _valid_args(t):
|
|
params = t.get("parameters", {}) or {}
|
|
props = params.get("properties", {}) or {}
|
|
out = {}
|
|
for r in params.get("required", []) or []:
|
|
ty = (props.get(r) or {}).get("type", "string")
|
|
if ty == "integer":
|
|
out[r] = 1
|
|
elif ty == "number":
|
|
out[r] = 1.0
|
|
elif ty == "boolean":
|
|
out[r] = True
|
|
elif ty == "array":
|
|
out[r] = []
|
|
elif ty == "object":
|
|
out[r] = {}
|
|
else:
|
|
out[r] = f"zz-{r}"
|
|
return out
|
|
|
|
approval_rows = []
|
|
for t in tp.TOOLS:
|
|
args = _valid_args(t)
|
|
req = tp.ToolRequest(**R(f"a-{t['name']}", t["name"], args))
|
|
try:
|
|
approval_rows.append({"name": t["name"], "args": args,
|
|
"requires": tp.tool_requires_approval(req)})
|
|
except Exception as ex: # noqa: BLE001 — unknown-tool class errors identically
|
|
approval_rows.append({"name": t["name"], "args": args, "error": str(ex)})
|
|
for nm, args in [("bash", {"command": "ls"}), ("unknown_tool_x", {})]:
|
|
req = tp.ToolRequest(**R(f"a-{nm}", nm, args))
|
|
try:
|
|
approval_rows.append({"name": nm, "args": args,
|
|
"requires": tp.tool_requires_approval(req)})
|
|
except Exception as ex: # noqa: BLE001
|
|
approval_rows.append({"name": nm, "args": args, "error": str(ex)})
|
|
cases.append(case("approval", "every_tool_plus_alias_unknown",
|
|
{"rows": approval_rows},
|
|
lambda: approval_rows))
|
|
|
|
out = {"oracle": "src/rune/tool_protocol.py (live)", "generated": "2026-10-07", "cases": cases}
|
|
target = REPO / "src/go/tool-protocol" / "truth_protocol.json"
|
|
target.write_text(json.dumps(out, indent=1, ensure_ascii=False) + "\n")
|
|
print(f"wrote {target} ({len(cases)} cases)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|