script-build-and-framework-tmp/script_build_host/tests/test_tool_argument_normalization.py

from __future__ import annotations

from typing import Any

import pytest
from agent import ToolRegistry

from script_build_host.tools.registry import (
    _structured_object,
    _structured_object_list,
)


def test_structured_arguments_accept_native_and_provider_encoded_json() -> None:
    assert _structured_object({"name": "native"}, "payload") == {"name": "native"}
    assert _structured_object('{"name":"encoded"}', "payload") == {"name": "encoded"}
    assert _structured_object_list('[{"index":1},{"index":2}]', "items") == [
        {"index": 1},
        {"index": 2},
    ]
    assert _structured_object_list(['{"index":1}'], "items") == [{"index": 1}]


@pytest.mark.parametrize("value", ["not-json", "123", '{"not":"an-array"}'])
def test_structured_object_list_rejects_non_object_arrays(value: str) -> None:
    with pytest.raises(ValueError):
        _structured_object_list(value, "items")


def test_structured_arguments_enforce_item_and_byte_limits() -> None:
    with pytest.raises(ValueError, match="item limit"):
        _structured_object_list([{}, {}], "items", max_items=1)
    with pytest.raises(ValueError, match="byte limit"):
        _structured_object({"value": "x" * (2 * 1024 * 1024)}, "payload")


def test_planner_schema_exposes_only_semantic_inputs_and_decision_ids() -> None:
    from script_build_host.tools.registry import register_script_tools

    registry = ToolRegistry()
    register_script_tools(registry, object())  # type: ignore[arg-type]

    schema = registry.get_schemas(["plan_script_tasks"])[0]
    contracts = schema["function"]["parameters"]["properties"]["contracts"]
    contract = contracts["items"]

    assert contracts["type"] == "array"
    assert "schema_version" not in contract["properties"]
    assert "input_decision_ids" in contract["required"]
    for derived in (
        "input_decision_refs",
        "base_artifact_ref",
        "write_scope",
        "output_schema",
        "budget",
        "candidate_closure_decision_refs",
        "compose_order",
    ):
        assert derived not in contract["properties"]
    assert contract["properties"]["task_kind"]["enum"] == [
        "direction",
        "pattern-retrieval",
        "decode-retrieval",
        "external-retrieval",
        "knowledge-retrieval",
        "structure",
        "paragraph",
        "element-set",
        "compare",
        "compose",
        "candidate-portfolio",
    ]
    assert contract["additionalProperties"] is False


@pytest.mark.asyncio
async def test_retrieval_tool_freezes_and_submits_in_one_terminal_call() -> None:
    class Gateway:
        def __init__(self) -> None:
            self.calls: list[tuple[str, Any]] = []

        async def retrieve(self, source: str, tool: str, query: Any, context: Any) -> Any:
            self.calls.append(("retrieve", (source, tool, query, context)))
            return {"artifact_ref": {"uri": "script-build://artifact-versions/1"}}

        async def submit_current_attempt(self, context: Any) -> Any:
            self.calls.append(("submit", context))
            return {"attempt_id": "attempt-1", "status": "awaiting_validation"}

    from script_build_host.tools.registry import register_script_tools

    gateway = Gateway()
    registry = ToolRegistry()
    register_script_tools(registry, gateway)  # type: ignore[arg-type]
    tool = registry._tools["search_knowledge"]["func"]

    result = await tool(keyword="hard tech", max_count=3, context={"task_id": "task-1"})

    assert result.terminate_run is True
    assert [name for name, _ in gateway.calls] == ["retrieve", "submit"]
    assert "attempt-1" in result.output