From acc958b49ac67f25105e8ff4f7bb8a4b3ef01e57 Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:40:49 +0800 Subject: [PATCH 1/4] feat(tools): validate full Draft 2020-12 schemas and reject unsafe remote refs --- tiny_harness/_tools.py | 79 +++++++++++++++++++++++++----------------- 1 file changed, 48 insertions(+), 31 deletions(-) diff --git a/tiny_harness/_tools.py b/tiny_harness/_tools.py index ee8ea81..7cb903a 100644 --- a/tiny_harness/_tools.py +++ b/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 From cc354262c0091eec5cd9ba07d5a325b0d7838327 Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:40:54 +0800 Subject: [PATCH 2/4] build: depend on Draft 2020-12 JSON Schema implementation --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 80b5d62..795bb5f 100644 --- a/pyproject.toml +++ b/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"] From b46e34f2c5ad43471091c0a9fd812c33a6b6feaa Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:41:02 +0800 Subject: [PATCH 3/4] test: reject non-object tool schemas instead of silently accepting them --- tests/test_tools.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_tools.py b/tests/test_tools.py index 09f9710..11b5e8a 100644 --- a/tests/test_tools.py +++ b/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(): From 528dd763ab9559d59a7bf1fef7dba277fbadffb1 Mon Sep 17 00:00:00 2001 From: XBetter <32125213+1998x-stack@users.noreply.github.com> Date: Wed, 28 Oct 2026 14:41:23 +0800 Subject: [PATCH 4/4] test(tools): nested arrays, union constraints, refs and modified HITL payloads --- tests/test_jsonschema_contract.py | 124 ++++++++++++++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 tests/test_jsonschema_contract.py diff --git a/tests/test_jsonschema_contract.py b/tests/test_jsonschema_contract.py new file mode 100644 index 0000000..60e3e73 --- /dev/null +++ b/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({}, {}) == []