feat(delegation): validate batch task quality before spawning children

Reject malformed tasks=[...] batches before any child agent is spawned:

- exact-duplicate goals (case/whitespace-normalized), error names both
  task indices
- placeholder goals: bare 'TODO', bare 'task N', unexpanded <...> or
  {...} template markers, or goals shorter than 10 chars after strip
- 1-task batches, with an error pointing the model at the single
  `goal` form instead

All checks are batch-only — the single-goal form is exempt by design
(short goals like goal="test" are valid there). Error strings are
actionable: each tells the model exactly how to fix the call.

Tool schema is unchanged (byte-stable); validation is runtime-only in
the existing batch-validation region.

Existing tests using terse batch goals ("A"/"B"/"C") updated to
realistic distinct goals per the new contract.

Inspired by: MoonshotAI/kimi-code agent-swarm.md validation rules (MIT)

tests/agent/test_subagent_stop_hook.py

@@ -159,7 +159,9 @@ class TestBatchMode:
             ]
             delegate_task(
                 tasks=[
-                    {"goal": "A"}, {"goal": "B"}, {"goal": "C"},
+                    {"goal": "Investigate module A"},
+                    {"goal": "Investigate module B"},
+                    {"goal": "Investigate module C"},
                 ],
                 parent_agent=_make_parent(),
             )
@@ -182,7 +184,10 @@ class TestBatchMode:
                  "_child_role": None},
             ]
             delegate_task(
-                tasks=[{"goal": "A"}, {"goal": "B"}],
+                tasks=[
+                    {"goal": "Investigate module A"},
+                    {"goal": "Investigate module B"},
+                ],
                 parent_agent=_make_parent(),
             )
 

tests/tools/test_delegate.py

@@ -618,7 +618,11 @@ class TestSubagentCostRollup(unittest.TestCase):
             ]
             result = json.loads(
                 delegate_task(
-                    tasks=[{"goal": "A"}, {"goal": "B"}, {"goal": "C"}],
+                    tasks=[
+                        {"goal": "Investigate module A"},
+                        {"goal": "Investigate module B"},
+                        {"goal": "Investigate module C"},
+                    ],
                     parent_agent=parent,
                 )
             )
@@ -1595,7 +1599,10 @@ class TestOrchestratorEndToEnd(unittest.TestCase):
                 def _orchestrator_run(user_message=None, task_id=None, stream_callback=None):
                     # Re-entrant: orchestrator spawns two leaves
                     delegate_task(
-                        tasks=[{"goal": "leaf-A"}, {"goal": "leaf-B"}],
+                        tasks=[
+                            {"goal": "Do leaf work stream A"},
+                            {"goal": "Do leaf work stream B"},
+                        ],
                         parent_agent=m,
                     )
                     return {

tests/tools/test_delegate_batch_validation.py

@@ -0,0 +1,139 @@
+#!/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()

tools/delegate_tool.py

@@ -21,6 +21,7 @@ import enum
 import contextvars
 import json
 import logging
+import re
 
 logger = logging.getLogger(__name__)
 import os
@@ -2974,6 +2975,66 @@ def _recover_tasks_from_json_string(
     return parsed, None
 
 
+# Placeholder shapes for batch goal validation: bare 'TODO', bare 'task N'
+# labels, or goals still carrying unexpanded template markers.
+_PLACEHOLDER_GOAL_RE = re.compile(r"^(todo|task\s*\d+)$", re.IGNORECASE)
+_TEMPLATE_MARKER_RE = re.compile(r"<[^<>]+>|\{[^{}]+\}")
+_MIN_BATCH_GOAL_LEN = 10
+
+
+def _validate_batch_tasks(task_list: List[Dict[str, Any]]) -> Optional[str]:
+    """Validate a tasks=[...] batch beyond per-task goal presence.
+
+    Returns an actionable error string, or None when the batch is valid.
+    Batch-only by design: the single-`goal` form legitimately uses short
+    goals, so these checks must never run on it.
+    """
+    if len(task_list) < 2:
+        return (
+            "Batch mode requires at least 2 tasks. For a single task, use "
+            "the `goal` parameter instead of `tasks`: "
+            'delegate_task(goal="...", context="...").'
+        )
+
+    seen: Dict[str, int] = {}
+    for i, task in enumerate(task_list):
+        goal = str(task.get("goal", "")).strip()
+        normalized = " ".join(goal.lower().split())
+
+        prev = seen.get(normalized)
+        if prev is not None:
+            return (
+                f"Task {i} duplicates task {prev}: both have the goal "
+                f"{goal!r}. Each task in a batch must do distinct work — "
+                "rewrite the goals so they don't overlap, or drop the "
+                "duplicate."
+            )
+        seen[normalized] = i
+
+        if _PLACEHOLDER_GOAL_RE.match(normalized):
+            return (
+                f"Task {i} has a placeholder goal ({goal!r}). Replace it "
+                "with a specific, self-contained description of what the "
+                "subagent should accomplish."
+            )
+        marker = _TEMPLATE_MARKER_RE.search(goal)
+        if marker:
+            return (
+                f"Task {i} goal contains an unexpanded template marker "
+                f"({marker.group(0)!r}). Substitute the real value before "
+                "calling delegate_task — subagents cannot resolve "
+                "placeholders."
+            )
+        if len(goal) < _MIN_BATCH_GOAL_LEN:
+            return (
+                f"Task {i} goal is too short ({goal!r}). Write a specific, "
+                "self-contained goal of at least "
+                f"{_MIN_BATCH_GOAL_LEN} characters so the subagent knows "
+                "exactly what to do."
+            )
+    return None
+
+
 def delegate_task(
     goal: Optional[str] = None,
     context: Optional[str] = None,
@@ -3095,6 +3156,15 @@ def delegate_task(
         if not task.get("goal", "").strip():
             return tool_error(f"Task {i} is missing a 'goal'.")
 
+    # Batch-only quality gate: catch malformed fan-outs (duplicate goals,
+    # placeholder goals, 1-task batches) before any child is spawned.  The
+    # single-`goal` form is deliberately exempt — short goals are valid there.
+    # Inspired by: MoonshotAI/kimi-code agent-swarm.md validation rules (MIT).
+    if tasks is not None and isinstance(tasks, list):
+        batch_error = _validate_batch_tasks(task_list)
+        if batch_error:
+            return tool_error(batch_error)
+
     overall_start = time.monotonic()
     results = []