Nick
ca4f7593d7
CI Pipeline / best-practices (push) Successful in 8s
Auto-version and Release / release (push) Failing after 14s
CI Pipeline / documentation (push) Successful in 5s
CI Pipeline / lint (push) Failing after 34s
CI Pipeline / canary (push) Successful in 37s
CI Pipeline / type-check (push) Successful in 31s
CI Pipeline / gitea-offline (push) Successful in 16s
CI Pipeline / unit-tests (push) Successful in 1m4s
CI Pipeline / required-checks (push) Successful in 1s
deploy / deploy (push) Successful in 48s
CORRECTION 1: #314 reopened — recall_with_associations mode field defect CONFIRMED in current source and FIXED (mode: association_merged). CORRECTION 3: Gitea credential removed from remote URL (credential helper). TOKEN EXPOSED: yes (in diagnostic output). TOKEN ROTATED: not yet (requires Gitea web UI access). Remote URL now shows username only, no secret. CORRECTION 4: duplicate-release-commit defect ROOT CAUSE: convergence computed warranted bump from commits-since-tag without checking if master version already satisfies the bump. FIX: already-converged guard in version_convergence.py (master version > last tag → no commit). Repo-aware (args.repo passed to read_version/latest_tag). CORRECTION 6: lifecycle terminalization — execution exception now guarantees task terminalizes as failed with evidence (no immortal RUNNING). CORRECTION 7: route instruction task-scoped on RuneTask.route_instruction (NOT shared operator state). Read-only leading-verb override. Isolation tests prove no cross-contamination between interleaved tasks. PL1-01/02/03/05: pl1_records.py (route resolution, proof level, failure class, candidate status), metrics with honest NOT_YET_MEASURABLE semantics, route-instruction learning loop (PE-1789102658-0269 SEED). 1253 tests green; ratchet 352/80 (shrink); henxels clean.
419 lines
16 KiB
Python
419 lines
16 KiB
Python
"""Failure-sensitive contract tests for Rune's internal tool protocol."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
|
|
from rune import tool_executor # noqa: E402
|
|
|
|
|
|
class ToolProtocolTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
# Import per test so RED reports the missing protocol implementation while
|
|
# still allowing the legacy parser regression below to run.
|
|
self.protocol = importlib.import_module("rune.tool_protocol")
|
|
|
|
def test_standalone_nested_json_normalizes_to_tool_request(self):
|
|
raw = '{"tool":"list_directory","args":{"path":"/tmp"}}'
|
|
|
|
requests = self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
self.assertEqual(len(requests), 1)
|
|
self.assertIsInstance(requests[0], self.protocol.ToolRequest)
|
|
self.assertEqual(requests[0].name, "list_directory")
|
|
self.assertEqual(requests[0].arguments, {"path": "/tmp"})
|
|
self.assertEqual(requests[0].source, "text-fallback")
|
|
self.assertTrue(requests[0].call_id)
|
|
|
|
def test_fenced_nested_json_normalizes(self):
|
|
raw = '```json\n{"tool":"read_file","args":{"path":"/tmp/example.json","offset":2}}\n```'
|
|
|
|
request = self.protocol.normalize_text_tool_calls(raw)[0]
|
|
|
|
self.assertEqual(request.name, "read_file")
|
|
self.assertEqual(request.arguments, {"path": "/tmp/example.json", "offset": 2})
|
|
|
|
def test_native_ollama_call_normalizes_to_same_contract(self):
|
|
message = {
|
|
"content": "",
|
|
"tool_calls": [
|
|
{
|
|
"function": {
|
|
"name": "list_directory",
|
|
"arguments": {"path": "/tmp"},
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
request = self.protocol.normalize_ollama_tool_calls(message)[0]
|
|
|
|
self.assertIsInstance(request, self.protocol.ToolRequest)
|
|
self.assertIsInstance(request.call_id, str)
|
|
self.assertTrue(request.call_id.strip())
|
|
self.assertEqual(request.name, "list_directory")
|
|
self.assertEqual(request.arguments, {"path": "/tmp"})
|
|
self.assertEqual(request.source, "ollama-native")
|
|
|
|
def test_valid_request_passes_central_schema_validation(self):
|
|
request = self.protocol.ToolRequest(
|
|
call_id="call_1",
|
|
name="git_commit",
|
|
arguments={"message": "test", "files": ["tests/test_tool_protocol.py"]},
|
|
source="ollama-native",
|
|
)
|
|
|
|
self.assertEqual(self.protocol.validate_tool_request(request), request)
|
|
|
|
def test_dispatch_rejects_invalid_request_before_executor_is_called(self):
|
|
request = self.protocol.ToolRequest(
|
|
call_id="call_invalid",
|
|
name="read_file",
|
|
arguments={"offset": 1},
|
|
source="ollama-native",
|
|
)
|
|
executor = Mock()
|
|
|
|
with self.assertRaisesRegex(ValueError, "required.*path"):
|
|
self.protocol.dispatch_validated_tool(request, executor)
|
|
|
|
executor.assert_not_called()
|
|
|
|
def test_dispatch_sends_validated_request_to_executor_once(self):
|
|
request = self.protocol.ToolRequest(
|
|
call_id="call_valid",
|
|
name="list_directory",
|
|
arguments={"path": "/tmp"},
|
|
source="ollama-native",
|
|
)
|
|
expected = {"success": True, "entries": []}
|
|
executor = Mock(return_value=expected)
|
|
|
|
result = self.protocol.dispatch_validated_tool(request, executor)
|
|
|
|
self.assertEqual(result, expected)
|
|
executor.assert_called_once_with("list_directory", {"path": "/tmp"})
|
|
|
|
def test_unknown_tool_is_rejected(self):
|
|
raw = '{"tool":"delete_everything","args":{}}'
|
|
|
|
with self.assertRaisesRegex(ValueError, "unknown tool"):
|
|
self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
def test_non_object_arguments_are_rejected(self):
|
|
cases = (
|
|
'{"tool":"list_directory","args":"/tmp"}',
|
|
'{"tool":"list_directory","args":["/tmp"]}',
|
|
'{"tool":"list_directory","args":null}',
|
|
)
|
|
|
|
for raw in cases:
|
|
with (
|
|
self.subTest(raw=raw),
|
|
self.assertRaisesRegex(ValueError, "arguments.*object"),
|
|
):
|
|
self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
def test_missing_required_argument_is_rejected(self):
|
|
raw = '{"tool":"read_file","args":{"offset":1}}'
|
|
|
|
with self.assertRaisesRegex(ValueError, "required.*path"):
|
|
self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
def test_wrong_primitive_argument_types_are_rejected(self):
|
|
cases = (
|
|
('{"tool":"read_file","args":{"path":7}}', "path.*string"),
|
|
(
|
|
'{"tool":"read_file","args":{"path":"/tmp/x","limit":"ten"}}',
|
|
"limit.*integer",
|
|
),
|
|
('{"tool":"git_diff","args":{"staged":"yes"}}', "staged.*boolean"),
|
|
(
|
|
'{"tool":"git_commit","args":{"message":"test","files":"one.py"}}',
|
|
"files.*array",
|
|
),
|
|
(
|
|
'{"tool":"git_commit","args":{"message":"test","files":[1]}}',
|
|
"files.*string",
|
|
),
|
|
)
|
|
|
|
for raw, error_pattern in cases:
|
|
with (
|
|
self.subTest(raw=raw),
|
|
self.assertRaisesRegex(ValueError, error_pattern),
|
|
):
|
|
self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
def test_boolean_is_not_accepted_for_integer_argument(self):
|
|
raw = '{"tool":"read_file","args":{"path":"/tmp/x","limit":true}}'
|
|
|
|
with self.assertRaisesRegex(ValueError, "limit.*integer"):
|
|
self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
def test_unknown_argument_is_rejected(self):
|
|
raw = '{"tool":"list_directory","args":{"path":"/tmp","recursive":true}}'
|
|
|
|
with self.assertRaisesRegex(ValueError, "unknown argument.*recursive"):
|
|
self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
def test_multiple_standalone_calls_are_returned_in_source_order(self):
|
|
raw = '{"tool":"list_directory","args":{"path":"/tmp"}}\n{"tool":"read_file","args":{"path":"/tmp/example"}}'
|
|
|
|
requests = self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
self.assertEqual([request.name for request in requests], ["list_directory", "read_file"])
|
|
self.assertEqual(len({request.call_id for request in requests}), 2)
|
|
|
|
def test_multiple_native_calls_are_returned_in_source_order(self):
|
|
message = {
|
|
"tool_calls": [
|
|
{
|
|
"id": "1",
|
|
"function": {
|
|
"name": "list_directory",
|
|
"arguments": {"path": "/tmp"},
|
|
},
|
|
},
|
|
{
|
|
"id": "2",
|
|
"function": {"name": "read_file", "arguments": {"path": "/tmp/x"}},
|
|
},
|
|
]
|
|
}
|
|
|
|
requests = self.protocol.normalize_ollama_tool_calls(message)
|
|
|
|
self.assertEqual([request.call_id for request in requests], ["1", "2"])
|
|
self.assertEqual([request.name for request in requests], ["list_directory", "read_file"])
|
|
|
|
def test_top_level_json_array_is_not_an_accepted_call_envelope(self):
|
|
raw = '[{"tool":"list_directory","args":{"path":"/tmp"}}]'
|
|
|
|
with self.assertRaisesRegex(ValueError, "array.*tool call"):
|
|
self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
def test_xml_wrapped_json_tool_call_normalizes(self):
|
|
raw = '<tool_call>{"tool":"list_directory","args":{"path":"/tmp"}}</tool_call>'
|
|
|
|
request = self.protocol.normalize_text_tool_calls(raw)[0]
|
|
|
|
self.assertEqual(request.name, "list_directory")
|
|
self.assertEqual(request.arguments, {"path": "/tmp"})
|
|
|
|
def test_malformed_protocol_call_is_rejected_not_ignored_as_prose(self):
|
|
raw = '<tool_call>{"tool":"list_directory","args":{"path":"/tmp"}</tool_call>'
|
|
|
|
with self.assertRaisesRegex(ValueError, "malformed.*tool"):
|
|
self.protocol.normalize_text_tool_calls(raw)
|
|
|
|
def test_malformed_standalone_before_valid_call_is_rejected_and_quarantined(self):
|
|
raw = '{"tool":"read_file","args":{"path":"/tmp"}\n{"tool":"list_directory","args":{"path":"/tmp"}}'
|
|
|
|
for operation in (
|
|
self.protocol.normalize_text_tool_calls,
|
|
self.protocol.extract_visible_text,
|
|
):
|
|
with (
|
|
self.subTest(operation=operation.__name__),
|
|
self.assertRaisesRegex(ValueError, "malformed.*tool"),
|
|
):
|
|
operation(raw)
|
|
|
|
def test_malformed_non_first_tool_key_before_valid_call_is_rejected(self):
|
|
raw = '{"note":"x","tool":"read_file","args":{"path":"/tmp"}\n{"tool":"list_directory","args":{"path":"/tmp"}}'
|
|
|
|
for operation in (
|
|
self.protocol.normalize_text_tool_calls,
|
|
self.protocol.extract_visible_text,
|
|
):
|
|
with (
|
|
self.subTest(operation=operation.__name__),
|
|
self.assertRaisesRegex(ValueError, "malformed.*tool"),
|
|
):
|
|
operation(raw)
|
|
|
|
def test_malformed_non_first_tool_key_after_valid_call_is_rejected(self):
|
|
raw = '{"tool":"list_directory","args":{"path":"/tmp"}}\n{"note":"x","tool":"read_file","args":{"path":"/tmp"}'
|
|
|
|
for operation in (
|
|
self.protocol.normalize_text_tool_calls,
|
|
self.protocol.extract_visible_text,
|
|
):
|
|
with (
|
|
self.subTest(operation=operation.__name__),
|
|
self.assertRaisesRegex(ValueError, "malformed.*tool"),
|
|
):
|
|
operation(raw)
|
|
|
|
def test_malformed_non_first_tool_key_in_array_residue_is_rejected(self):
|
|
raw = '[{"note":"x","tool":"read_file","args":{"path":"/tmp"}\n{"tool":"list_directory","args":{"path":"/tmp"}}'
|
|
|
|
for operation in (
|
|
self.protocol.normalize_text_tool_calls,
|
|
self.protocol.extract_visible_text,
|
|
):
|
|
with (
|
|
self.subTest(operation=operation.__name__),
|
|
self.assertRaisesRegex(ValueError, "malformed.*tool"),
|
|
):
|
|
operation(raw)
|
|
|
|
def test_truncated_call_with_escaped_tool_key_is_quarantined(self):
|
|
raw = '{"\\u0074ool":"read_file","args":{"path":"/tmp"'
|
|
|
|
for operation in (
|
|
self.protocol.normalize_text_tool_calls,
|
|
self.protocol.extract_visible_text,
|
|
):
|
|
with (
|
|
self.subTest(operation=operation.__name__),
|
|
self.assertRaisesRegex(ValueError, "malformed.*tool"),
|
|
):
|
|
operation(raw)
|
|
|
|
self.assertTrue(self.protocol.contains_internal_protocol(raw))
|
|
|
|
def test_malformed_xml_before_valid_call_is_rejected_and_quarantined(self):
|
|
raw = (
|
|
'<tool_call>{"tool":"read_file","args":{"path":"/tmp"}</tool_call>\n'
|
|
'{"tool":"list_directory","args":{"path":"/tmp"}}'
|
|
)
|
|
|
|
for operation in (
|
|
self.protocol.normalize_text_tool_calls,
|
|
self.protocol.extract_visible_text,
|
|
):
|
|
with (
|
|
self.subTest(operation=operation.__name__),
|
|
self.assertRaisesRegex(ValueError, "malformed.*tool"),
|
|
):
|
|
operation(raw)
|
|
|
|
def test_recognized_standalone_protocol_is_removed_from_visible_text(self):
|
|
raw = 'Before\n{"tool":"list_directory","args":{"path":"/tmp"}}\nAfter'
|
|
|
|
self.assertEqual(self.protocol.extract_visible_text(raw), "Before\nAfter")
|
|
|
|
def test_recognized_fenced_and_xml_protocol_are_removed_from_visible_text(self):
|
|
raw = (
|
|
"Before\n"
|
|
"```json\n"
|
|
'{"tool":"list_directory","args":{"path":"/tmp"}}\n'
|
|
"```\n"
|
|
'<tool_call>{"tool":"read_file","args":{"path":"/tmp/x"}}</tool_call>\n'
|
|
"After"
|
|
)
|
|
|
|
self.assertEqual(self.protocol.extract_visible_text(raw), "Before\nAfter")
|
|
|
|
def test_mixed_fence_preserves_ordinary_json_while_removing_tool_protocol(self):
|
|
raw = (
|
|
'```json\n{"theme":"dark","nested":{"enabled":true}}\n{"tool":"list_directory","args":{"path":"/tmp"}}\n```'
|
|
)
|
|
expected = '```json\n{"theme":"dark","nested":{"enabled":true}}\n```'
|
|
|
|
self.assertEqual(self.protocol.extract_visible_text(raw), expected)
|
|
|
|
def test_ordinary_non_tool_json_is_preserved_verbatim(self):
|
|
raw = 'Use this configuration:\n{"theme":"dark","nested":{"enabled":true}}'
|
|
|
|
self.assertEqual(self.protocol.normalize_text_tool_calls(raw), [])
|
|
self.assertEqual(self.protocol.extract_visible_text(raw), raw)
|
|
self.assertFalse(self.protocol.contains_internal_protocol(raw))
|
|
|
|
def test_ordinary_json_with_tool_as_value_is_preserved_verbatim(self):
|
|
raw = 'Use this configuration:\n{"category":"tool","note":"select a tool"}'
|
|
|
|
self.assertEqual(self.protocol.normalize_text_tool_calls(raw), [])
|
|
self.assertEqual(self.protocol.extract_visible_text(raw), raw)
|
|
self.assertFalse(self.protocol.contains_internal_protocol(raw))
|
|
|
|
def test_escaped_tool_text_inside_ordinary_value_is_not_protocol(self):
|
|
samples = (
|
|
'{"note":"literal \\u0074ool key text"}',
|
|
'{"note":"example: \\"\\u0074ool\\": not a key"}',
|
|
)
|
|
|
|
for raw in samples:
|
|
with self.subTest(raw=raw):
|
|
self.assertEqual(self.protocol.normalize_text_tool_calls(raw), [])
|
|
self.assertEqual(self.protocol.extract_visible_text(raw), raw)
|
|
self.assertFalse(self.protocol.contains_internal_protocol(raw))
|
|
|
|
def test_protocol_markers_and_tool_results_are_internal(self):
|
|
internal_samples = (
|
|
'{"tool":"list_directory","args":{"path":"/tmp"}}',
|
|
'<tool_call>{"tool":"list_directory","args":',
|
|
"[Tool result]\nsecret output",
|
|
"[Tool error]\nprivate failure",
|
|
)
|
|
|
|
for sample in internal_samples:
|
|
with self.subTest(sample=sample):
|
|
self.assertTrue(self.protocol.contains_internal_protocol(sample))
|
|
|
|
def test_result_markers_ignore_leading_whitespace_and_case(self):
|
|
internal_samples = (
|
|
" [tool result]\nsecret output",
|
|
"\t[TOOL ERROR]\nprivate failure",
|
|
"prefix\n [ToOl ReSuLt]\nsecret output",
|
|
)
|
|
|
|
for sample in internal_samples:
|
|
with self.subTest(sample=sample):
|
|
self.assertTrue(self.protocol.contains_internal_protocol(sample))
|
|
|
|
def test_tool_definitions_convert_to_native_ollama_shape(self):
|
|
definition = next(tool for tool in tool_executor.TOOLS if tool["name"] == "read_file")
|
|
|
|
native = self.protocol.to_ollama_tools([definition])
|
|
|
|
self.assertEqual(native[0]["type"], "function")
|
|
self.assertEqual(native[0]["function"]["name"], definition["name"])
|
|
self.assertEqual(native[0]["function"]["description"], definition["description"])
|
|
self.assertEqual(native[0]["function"]["parameters"], definition["parameters"])
|
|
|
|
|
|
class LegacyParserRegressionTests(unittest.TestCase):
|
|
def test_current_parser_accepts_prompted_standalone_nested_json(self):
|
|
raw = '{"tool":"list_directory","args":{"path":"/tmp"}}'
|
|
|
|
self.assertEqual(
|
|
tool_executor.parse_tool_call(raw),
|
|
{"tool": "list_directory", "args": {"path": "/tmp"}},
|
|
)
|
|
|
|
def test_top_level_tool_executor_can_load_protocol_parser(self):
|
|
source_dir = ROOT / "src" / "rune"
|
|
script = (
|
|
"import tool_executor; "
|
|
"assert tool_executor.parse_tool_call("
|
|
'\'{\\"tool\\":\\"list_directory\\",\''
|
|
'\'\\"args\\":{\\"path\\":\\"/tmp\\"}}\''
|
|
") == {'tool': 'list_directory', 'args': {'path': '/tmp'}}"
|
|
)
|
|
|
|
completed = subprocess.run(
|
|
[sys.executable, "-c", script],
|
|
cwd=source_dir,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
self.assertEqual(completed.returncode, 0, completed.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|