140 lines
5.5 KiB
Python
140 lines
5.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Batch input validation for delegate_task(tasks=[...]).
|
|
|
|
Guards against the model wasting a whole fan-out on malformed batches:
|
|
exact-duplicate goals, placeholder goals ('TODO', 'task N', unexpanded
|
|
template markers, too-short), and 1-task batches that should have used
|
|
the single `goal` form.
|
|
|
|
All checks are BATCH-ONLY — the single-goal form is deliberately exempt
|
|
(short goals like goal="test" are valid there and widely used).
|
|
|
|
Inspired by: MoonshotAI/kimi-code agent-swarm.md validation rules (MIT)
|
|
"""
|
|
|
|
import json
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from tools.delegate_tool import delegate_task
|
|
|
|
|
|
def _make_mock_parent(depth=0):
|
|
parent = MagicMock()
|
|
parent.base_url = "https://openrouter.ai/api/v1"
|
|
parent.api_key = "test-key"
|
|
parent.provider = "openrouter"
|
|
parent.api_mode = "chat_completions"
|
|
parent.model = "anthropic/claude-sonnet-4"
|
|
parent.platform = "cli"
|
|
parent.providers_allowed = None
|
|
parent.providers_ignored = None
|
|
parent.providers_order = None
|
|
parent.provider_sort = None
|
|
parent._session_db = None
|
|
parent._delegate_depth = depth
|
|
parent._active_children = []
|
|
parent._active_children_lock = threading.Lock()
|
|
parent._print_fn = None
|
|
parent.tool_progress_callback = None
|
|
parent.thinking_callback = None
|
|
return parent
|
|
|
|
|
|
def _call(tasks):
|
|
return json.loads(delegate_task(tasks=tasks, parent_agent=_make_mock_parent()))
|
|
|
|
|
|
GOOD_A = "Refactor the login handler to use the new session helper"
|
|
GOOD_B = "Write regression tests for the session expiry watcher"
|
|
|
|
|
|
class TestBatchDuplicateGoals(unittest.TestCase):
|
|
def test_exact_duplicate_goals_rejected(self):
|
|
result = _call([{"goal": GOOD_A}, {"goal": GOOD_A}])
|
|
self.assertIn("error", result)
|
|
self.assertIn("duplicate", result["error"].lower())
|
|
|
|
def test_duplicate_detection_normalizes_case_and_whitespace(self):
|
|
result = _call([{"goal": GOOD_A}, {"goal": " " + GOOD_A.upper() + " "}])
|
|
self.assertIn("error", result)
|
|
self.assertIn("duplicate", result["error"].lower())
|
|
|
|
def test_duplicate_error_names_both_task_indices(self):
|
|
result = _call([{"goal": GOOD_A}, {"goal": GOOD_B}, {"goal": GOOD_A}])
|
|
self.assertIn("error", result)
|
|
self.assertIn("2", result["error"])
|
|
self.assertIn("0", result["error"])
|
|
|
|
|
|
class TestBatchPlaceholderGoals(unittest.TestCase):
|
|
def test_bare_todo_rejected_case_insensitive(self):
|
|
for todo in ("TODO", "todo", "ToDo"):
|
|
result = _call([{"goal": GOOD_A}, {"goal": todo}])
|
|
self.assertIn("error", result, todo)
|
|
|
|
def test_task_n_placeholder_rejected(self):
|
|
# 'Task 123456789' is >10 chars, so only the bare-'task N' shape
|
|
# check can reject it — proves the pattern check exists.
|
|
result = _call([{"goal": GOOD_A}, {"goal": "Task 123456789"}])
|
|
self.assertIn("error", result)
|
|
self.assertIn("placeholder", result["error"].lower())
|
|
|
|
def test_unexpanded_angle_template_marker_rejected(self):
|
|
result = _call([{"goal": GOOD_A}, {"goal": "Implement <feature_name> end to end"}])
|
|
self.assertIn("error", result)
|
|
self.assertIn("template", result["error"].lower())
|
|
|
|
def test_unexpanded_brace_template_marker_rejected(self):
|
|
result = _call([{"goal": GOOD_A}, {"goal": "Summarize {file_path} for the report"}])
|
|
self.assertIn("error", result)
|
|
self.assertIn("template", result["error"].lower())
|
|
|
|
def test_too_short_goal_rejected(self):
|
|
result = _call([{"goal": GOOD_A}, {"goal": "fix bug"}])
|
|
self.assertIn("error", result)
|
|
|
|
def test_placeholder_error_is_actionable(self):
|
|
result = _call([{"goal": GOOD_A}, {"goal": "TODO"}])
|
|
self.assertIn("error", result)
|
|
# Error must tell the model HOW to fix the call.
|
|
self.assertIn("specific", result["error"].lower())
|
|
|
|
|
|
class TestSingleTaskBatch(unittest.TestCase):
|
|
def test_one_task_batch_rejected_pointing_to_goal_form(self):
|
|
result = _call([{"goal": GOOD_A}])
|
|
self.assertIn("error", result)
|
|
self.assertIn("goal", result["error"])
|
|
self.assertIn("2", result["error"]) # "at least 2"
|
|
|
|
|
|
class TestValidBatchStillRuns(unittest.TestCase):
|
|
def test_two_distinct_goals_pass_validation(self):
|
|
with patch("tools.delegate_tool._run_single_child") as mock_run:
|
|
mock_run.side_effect = [
|
|
{"task_index": 0, "status": "completed", "summary": "A done",
|
|
"api_calls": 1, "duration_seconds": 1.0, "_child_role": None},
|
|
{"task_index": 1, "status": "completed", "summary": "B done",
|
|
"api_calls": 1, "duration_seconds": 1.0, "_child_role": None},
|
|
]
|
|
result = _call([{"goal": GOOD_A}, {"goal": GOOD_B}])
|
|
self.assertNotIn("error", result)
|
|
self.assertEqual(len(result["results"]), 2)
|
|
|
|
def test_single_goal_form_unaffected_by_batch_checks(self):
|
|
# goal="test" is short — must NOT trip the batch-only length check.
|
|
parent = _make_mock_parent()
|
|
with patch("tools.delegate_tool._run_single_child") as mock_run:
|
|
mock_run.return_value = {
|
|
"task_index": 0, "status": "completed", "summary": "ok",
|
|
"api_calls": 1, "duration_seconds": 1.0, "_child_role": None,
|
|
}
|
|
result = json.loads(delegate_task(goal="test", parent_agent=parent))
|
|
self.assertNotIn("error", result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|