fix(tools): enforce JSON Schema Draft 2020-12 on agent tool arguments

pyproject.toml

@@ -10,7 +10,7 @@ name = "tiny-harness"
 version = "0.1.0"
 description = "Minimal AI agent harness"
 requires-python = ">=3.11"
-dependencies = ["httpx>=0.27.0"]
+dependencies = ["httpx>=0.27.0", "jsonschema>=4.23,<5"]
 
 [project.optional-dependencies]
 dev = ["pytest>=8.0.0", "pytest-asyncio>=0.23.0"]

tests/test_jsonschema_contract.py

@@ -0,0 +1,124 @@
+"""Runtime validates nested JSON Schema, not only top-level annotations."""
+import pytest
+
+from tiny_harness._guard import FilesystemGuard
+from tiny_harness._hitl import ApprovalDecision, ApprovalGate
+from tiny_harness._tools import ToolDef, ToolExecutor, ToolRegistry, validate_schema
+
+
+COMPLEX_SCHEMA = {
+    "type": "object",
+    "required": ["jobs"],
+    "additionalProperties": False,
+    "properties": {
+        "jobs": {
+            "type": "array",
+            "minItems": 1,
+            "maxItems": 2,
+            "items": {
+                "type": "object",
+                "required": ["name", "retries"],
+                "additionalProperties": False,
+                "properties": {
+                    "name": {"type": "string", "minLength": 3, "pattern": "^[a-z]+"},
+                    "retries": {"type": "integer", "minimum": 0, "maximum": 4},
+                },
+            },
+        }
+    },
+}
+
+
+def test_nested_object_and_array_constraints_are_enforced():
+    valid = {"jobs": [{"name": "build", "retries": 0}]}
+    invalid = {"jobs": [
+        {"name": "X", "retries": True, "unexpected": "data"},
+        {"name": "ok", "retries": 6},
+        {"name": "third", "retries": 0},
+    ], "extra": "no"}
+    assert validate_schema(COMPLEX_SCHEMA, valid) == []
+    failures = validate_schema(COMPLEX_SCHEMA, invalid)
+    assert len(failures) >= 5
+    assert any("jobs.0.name" in f for f in failures)
+    assert any("jobs.0.retries" in f for f in failures)
+    assert any("jobs.1.retries" in f for f in failures)
+    assert any("unexpected" in f for f in failures)
+    assert any("extra" in f for f in failures)
+
+
+def test_anyof_oneof_and_formats():
+    schema = {"type": "object", "properties": {
+        "choice": {"oneOf": [
+            {"type": "integer", "minimum": 1},
+            {"type": "string", "minLength": 5},
+        ]},
+        "email": {"type": "string", "format": "email"},
+    }, "required": ["choice", "email"]}
+    assert validate_schema(schema, {"choice": "valid", "email": "person@example.com"}) == []
+    errors = validate_schema(schema, {"choice": 0, "email": "not-an-email"})
+    assert len(errors) == 2
+
+
+def test_internal_schema_references_work():
+    schema = {
+        "$defs": {"project": {"type": "string", "minLength": 4}},
+        "type": "object",
+        "properties": {"project": {"$ref": "#/$defs/project"}},
+    }
+    assert validate_schema(schema, {"project": "abcd"}) == []
+    assert validate_schema(schema, {"project": "a"})
+
+
+@pytest.mark.parametrize("invalid_schema", [
+    {"type": "object", "properties": {"x": {"type": "not-a-type"}}},
+    {"type": "array"},
+    {"type": "object", "properties": {"x": {"$ref": "https://example.test/schema"}}},
+])
+def test_bad_tool_schemas_rejected_at_registration(invalid_schema):
+    with pytest.raises(ValueError):
+        ToolDef("bad_tool", "invalid", invalid_schema, risk_level="read_only")
+
+
+@pytest.mark.asyncio
+async def test_nested_invalid_arguments_never_reach_handler(tmp_path):
+    executed = []
+    registry = ToolRegistry()
+    registry.register_from_def(
+        ToolDef("batch", "batch", COMPLEX_SCHEMA, risk_level="read_only"),
+        lambda args: executed.append(args) or "done",
+    )
+    executor = ToolExecutor(registry, FilesystemGuard(str(tmp_path)))
+    failure = await executor.execute("batch", {"jobs": [{"name": "x", "retries": -1}]}, "c1")
+    assert not failure.success and "jobs.0" in failure.content
+    assert executed == []
+
+
+@pytest.mark.asyncio
+async def test_approval_modified_nested_payload_is_revalidated(tmp_path):
+    executed = []
+    registry = ToolRegistry()
+    registry.register_from_def(
+        ToolDef("batch", "batch", COMPLEX_SCHEMA, risk_level="mutation"),
+        lambda args: executed.append(args) or "done",
+    )
+
+    async def approve(request):
+        assert request.args["jobs"][0]["name"] == "valid"
+        return ApprovalDecision(
+            approved=True, modified_args={"jobs": [{"name": "X", "retries": 0}]}
+        )
+
+    executor = ToolExecutor(
+        registry, FilesystemGuard(str(tmp_path)), approval_gate=ApprovalGate(handler=approve),
+    )
+    failure = await executor.execute(
+        "batch", {"jobs": [{"name": "valid", "retries": 0}]}, "c2",
+    )
+    assert not failure.success
+    assert "Invalid approved arguments" in failure.content
+    assert executed == []
+
+
+def test_all_args_must_be_objects_even_for_empty_schema():
+    assert validate_schema({}, [])
+    assert validate_schema({}, {}) == []

tests/test_tools.py

@@ -215,7 +215,8 @@ def test_validate_schema_valid():
 def test_validate_schema_non_object():
     schema = {"type": "array"}
     errors = validate_schema(schema, {})
-    assert len(errors) == 0
+    assert errors
+    assert "object" in errors[0]
 
 
 def test_validate_schema_array_type():

tiny_harness/_tools.py

@@ -6,6 +6,9 @@
 from collections.abc import Callable
 from typing import TYPE_CHECKING
 
+from jsonschema import Draft202012Validator, FormatChecker
+from jsonschema.exceptions import SchemaError
+
 from tiny_harness._events import StreamEvent
 from tiny_harness._tool_outcome import ToolOutcome
 
@@ -28,6 +31,15 @@ class ToolDef:
     def __post_init__(self) -> None:
         if self.risk_level not in ("safe", "read_only", "mutation", "destructive", "dangerous"):
             raise ValueError(f"Unknown tool risk_level: {self.risk_level!r}")
+        if not isinstance(self.parameters, dict):
+            raise ValueError("Tool parameters must be a JSON Schema object")
+        try:
+            Draft202012Validator.check_schema(self.parameters)
+        except SchemaError as exc:
+            raise ValueError(f"Invalid tool parameter JSON Schema: {exc.message}") from exc
+        if self.parameters.get("type", "object") != "object":
+            raise ValueError("Tool parameter schema must describe an object")
+        _reject_external_refs(self.parameters)
         if not isinstance(self.path_args, tuple) or len(set(self.path_args)) != len(self.path_args):
             raise ValueError("path_args must be a tuple of distinct schema property names")
         props = self.parameters.get("properties", {})
@@ -229,37 +241,42 @@ def _format(self, raw) -> str:
         return result
 
 
+def _reject_external_refs(value) -> None:
+    """Tool descriptions must not initiate network retrieval during validation."""
+    if isinstance(value, dict):
+        for key, item in value.items():
+            if key in ("$ref", "$dynamicRef") and isinstance(item, str) and not item.startswith("#"):
+                raise ValueError("External JSON Schema references are not supported")
+            _reject_external_refs(item)
+    elif isinstance(value, list):
+        for item in value:
+            _reject_external_refs(item)
+
+
 def validate_schema(schema: dict, args: dict) -> list[str]:
-    errors = []
-    schema_type = schema.get("type")
-    if schema_type != "object":
-        return errors
+    """Validate tool arguments against JSON Schema Draft 2020-12.
+
+    Tool calls must always be objects, even for an empty legacy schema. All
+    nested/array/combination constraints are enforced before HITL and again
+    after approval-time argument modification.
+    """
     if not isinstance(args, dict):
         return [f"arguments should be an object, got {type(args).__name__}"]
-    properties = schema.get("properties", {})
-    required = schema.get("required", [])
-    for field in required:
-        if field not in args:
-            errors.append(f"'{field}' is required but was not provided")
-    if schema.get("additionalProperties") is False:
-        for key in args.keys() - properties.keys():
-            errors.append(f"'{key}' is not an allowed argument")
-    for key, value in args.items():
-        if key in properties:
-            prop = properties[key]
-            expected_type = prop.get("type")
-            if expected_type == "string" and not isinstance(value, str):
-                errors.append(f"'{key}' should be a string, got {type(value).__name__}")
-            elif expected_type == "integer" and (not isinstance(value, int) or isinstance(value, bool)):
-                errors.append(f"'{key}' should be an integer, got {type(value).__name__}")
-            elif expected_type == "number" and (not isinstance(value, (int, float)) or isinstance(value, bool)):
-                errors.append(f"'{key}' should be a number, got {type(value).__name__}")
-            elif expected_type == "boolean" and not isinstance(value, bool):
-                errors.append(f"'{key}' should be a boolean, got {type(value).__name__}")
-            elif expected_type == "array" and not isinstance(value, list):
-                errors.append(f"'{key}' should be an array, got {type(value).__name__}")
-            elif expected_type == "object" and not isinstance(value, dict):
-                errors.append(f"'{key}' should be an object, got {type(value).__name__}")
-            if "enum" in prop and value not in prop["enum"]:
-                errors.append(f"'{key}' must be one of {prop['enum']}, got {value!r}")
-    return errors
+    if not isinstance(schema, dict) or schema.get("type", "object") != "object":
+        return ["Tool parameter schema must describe an object"]
+    try:
+        _reject_external_refs(schema)
+        validator = Draft202012Validator(
+            {"type": "object", **schema},
+            format_checker=FormatChecker(),
+        )
+        failures = list(validator.iter_errors(args))
+    except Exception as exc:
+        return [f"Invalid tool schema or reference: {exc}"]
+    failures.sort(key=lambda e: ("/".join(map(str, e.absolute_path)), e.message))
+    messages = []
+    for err in failures:
+        path = ".".join(str(k) for k in err.absolute_path)
+        where = f"'{path}' " if path else ""
+        messages.append(where + err.message)
+    return messages