diff --git a/agent/estop.py b/agent/estop.py index 1fe196bca0cee..92cd7c32da7db 100644 --- a/agent/estop.py +++ b/agent/estop.py @@ -57,11 +57,18 @@ def sentinel_path() -> Path: def is_engaged() -> bool: - """Cheap check (one stat): is the global emergency stop engaged?""" + """Cheap check (one stat): is the global emergency stop engaged? + + Fail SAFE on stat errors: if we cannot determine whether the sentinel + exists (permission error, transient I/O failure on HERMES_HOME), report + engaged. The module contract is that the pause must hold even when the + sentinel is unreadable — a fail-open here would silently lift an + operator's emergency stop exactly when the filesystem is misbehaving. + """ try: return sentinel_path().exists() except OSError: - return False + return True def engage(reason: Optional[str] = None) -> Path: diff --git a/cron/jobs.py b/cron/jobs.py index 377b34c35487a..49d2bd0d78432 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -1534,6 +1534,36 @@ def _normalized_inference_axes(job: Dict[str, Any]) -> Tuple[Optional[str], Opti ) +def _validate_job_mode_invariants( + monitor_script: Optional[str], + monitor_url: Optional[str], + no_agent: bool, + script: Optional[str], +) -> None: + """Shared create/update validation for job execution-mode invariants. + + ONE owner for the class: create_job and update_job both call this so an + invariant enforced at create time cannot be violated through the update + door (monitor jobs silently degrading when no_agent is flipped on, etc.). + """ + if monitor_script and monitor_url: + raise ValueError( + "monitor_script and monitor_url are mutually exclusive — a job " + "can only have one monitor source." + ) + if (monitor_script or monitor_url) and no_agent: + raise ValueError( + "monitor_script/monitor_url cannot be combined with no_agent=True — " + "the whole point of a monitor job is to suppress or wake the AGENT " + "based on source changes. Use a plain no_agent script job instead." + ) + if no_agent and not script: + raise ValueError( + "no_agent=True requires a script — with no agent and no script " + "there is nothing for the job to run." + ) + + def create_job( prompt: Optional[str], schedule: str, @@ -1650,26 +1680,16 @@ def create_job( # Monitor-mode validation: exactly one source, and monitor mode only # makes sense when there IS an agent to suppress/wake. - if normalized_monitor_script and normalized_monitor_url: - raise ValueError( - "monitor_script and monitor_url are mutually exclusive — a job " - "can only have one monitor source." - ) - if (normalized_monitor_script or normalized_monitor_url) and normalized_no_agent: - raise ValueError( - "monitor_script/monitor_url cannot be combined with no_agent=True — " - "the whole point of a monitor job is to suppress or wake the AGENT " - "based on source changes. Use a plain no_agent script job instead." - ) - # no_agent jobs are meaningless without a script — the script IS the job. - # Surface this as a clear ValueError at create time so bad configs never - # reach the scheduler. - if normalized_no_agent and not normalized_script: - raise ValueError( - "no_agent=True requires a script — with no agent and no script " - "there is nothing for the job to run." - ) + # Surface these as clear ValueErrors at create time so bad configs never + # reach the scheduler (shared with update_job, see + # _validate_job_mode_invariants). + _validate_job_mode_invariants( + normalized_monitor_script, + normalized_monitor_url, + normalized_no_agent, + normalized_script, + ) # Normalize context_from: accept str or list of str, store as list or None if isinstance(context_from, str): @@ -1859,8 +1879,33 @@ def update_job(job_id: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]] else: updates["workdir"] = _normalize_workdir(_wd) + # Normalize monitor fields the same way create_job does (empty + # string clears the field). + for _mon_field in ("monitor_script", "monitor_url"): + if _mon_field in updates: + _mv = updates[_mon_field] + _mv = str(_mv).strip() if isinstance(_mv, str) else None + updates[_mon_field] = _mv or None + previous_inference_axes = _normalized_inference_axes(job) updated = _apply_skill_fields({**job, **updates}) + + # Re-check execution-mode invariants on the MERGED record when + # any participating field changes, so create-time invariants + # can't be violated through the update door (e.g. flipping + # no_agent=True on a monitor job would silently disable the + # monitor: the scheduler's no_agent short-circuit runs before + # the monitor gate). Scoped to changed fields so legacy records + # untouched by this update keep loading. + if {"monitor_script", "monitor_url", "no_agent", "script"}.intersection(updates): + _upd_script = updated.get("script") + _upd_script = str(_upd_script).strip() if isinstance(_upd_script, str) else None + _validate_job_mode_invariants( + updated.get("monitor_script") or None, + updated.get("monitor_url") or None, + bool(updated.get("no_agent")), + _upd_script or None, + ) schedule_changed = "schedule" in updates inference_fields_changed = bool( {"provider", "model", "base_url", "no_agent"}.intersection(updates) @@ -2011,6 +2056,17 @@ def remove_job(job_id: str) -> bool: # Clean up output directory to prevent orphaned dirs accumulating if job_output_dir.exists(): shutil.rmtree(job_output_dir) + # Clean up the job's durable notepad (cron/notepad.db) — without + # this, removed jobs orphan their KV rows forever. Best effort: + # a notepad failure must never block the removal itself. + try: + from cron.notepad import clear_notepad + clear_notepad(canonical_id) + except Exception: + logger.debug( + "Failed to clear notepad for removed job %s", + canonical_id, exc_info=True, + ) return True return False diff --git a/cron/notepad.py b/cron/notepad.py index 9b8864bb6d547..61d747673b55b 100644 --- a/cron/notepad.py +++ b/cron/notepad.py @@ -150,7 +150,13 @@ def list_notes(job_id: str) -> List[Dict[str, Any]]: def clear_notepad(job_id: str) -> int: - """Delete every key for one job (e.g. on job removal). Returns row count.""" + """Delete every key for one job (e.g. on job removal). Returns row count. + + Called from ``cron.jobs.remove_job`` so deleted jobs don't orphan their + rows. No-ops without creating the DB when no notepad file exists yet. + """ + if not NOTEPAD_FILE.exists(): + return 0 with _transaction() as conn: cur = conn.execute( "DELETE FROM cron_notepad WHERE job_id=?", (str(job_id),) diff --git a/gateway/run.py b/gateway/run.py index 6690c5021ee52..bd2937bee9f77 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14364,6 +14364,7 @@ async def _dispatch_busy_slash_command( "restart": self._handle_restart_command, "approve": self._handle_approve_command, "deny": self._handle_deny_command, + "pause": self._handle_pause_command, "agents": self._handle_agents_command, "background": self._handle_background_command, "kanban": self._handle_kanban_command, @@ -14393,6 +14394,36 @@ async def _dispatch_busy_slash_command( f"mid-turn. Wait for the current response or `/stop` first." ) + async def _handle_pause_command(self, event: MessageEvent): + """`/pause [reason]` engages the global emergency stop; `/pause off` + (aliases: resume/stop) lifts it. + + This is the in-band resume path for messaging-only operators — the + estop gate above deliberately lets recognized slash commands through + while paused so a user without host-shell access is never locked out. + """ + from agent import estop + + args = (event.get_command_args() or "").strip() + if args.lower() in {"off", "resume", "stop", "disengage"}: + if estop.disengage(): + return "▶️ Resumed — new work is accepted again." + return "Hermes wasn't paused." + state = estop.get_state() + if state is not None and not args: + reason = state.get("reason") + suffix = f" (reason: {reason})" if reason else "" + return ( + f"⏸️ Hermes is already paused{suffix}. " + "Use `/pause off` to resume." + ) + estop.engage(reason=args or None) + suffix = f" (reason: {args})" if args else "" + return ( + f"⏸️ Paused{suffix}. New cron/kanban/gateway work is on hold; " + "in-flight work finishes normally. Use `/pause off` to resume." + ) + async def _busy_start_command(self, event: MessageEvent, quick_key: str, source): # Telegram sends /start for bot launches/deep-links. Treat it as a # platform ping, not a user command: no help dump, no agent @@ -14740,6 +14771,16 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: # gate — pause stops NEW work, it never kills or orphans running # work. Placed after auth so unauthorized senders keep the normal # silent/pairing behavior and can't probe pause state. + # + # Passthroughs (pause blocks new AGENT turns, not control traffic): + # * recognized slash commands — /status, /help, /new, /approve and + # friends must keep working while paused, and /pause off is the + # in-band resume path for messaging-only users; + # * replies owned by IN-FLIGHT work — a pending detached-update + # prompt, clarify, slash-confirm, or dangerous-command approval, + # plus any message steering a session whose agent is already + # running. Swallowing those would stall work the pause promised + # not to touch. if not is_internal: try: from agent.estop import paused_reply as _estop_paused_reply @@ -14747,12 +14788,53 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: except ImportError: _paused_notice = None if _paused_notice is not None: - logger.info( - "Gateway turn paused by global emergency stop (platform=%s chat=%s)", - getattr(getattr(source, "platform", None), "value", "unknown"), - getattr(source, "chat_id", None) or "unknown", - ) - return _paused_notice + _estop_allow = False + _estop_cmd = None + try: + _estop_cmd = event.get_command() + except Exception: + _estop_cmd = None + if _estop_cmd: + try: + from hermes_cli.commands import ( + resolve_command as _resolve_estop_cmd, + ) + _estop_allow = _resolve_estop_cmd(_estop_cmd) is not None + except Exception: + _estop_allow = False + if not _estop_allow: + try: + _estop_key = self._session_key_for_source(source) + _estop_state = self._peek_session_state(_estop_key) + if ( + _estop_state is not None + and _estop_state.persistent.update_prompt_pending + ): + _estop_allow = True + if not _estop_allow and self._is_session_running(_estop_key): + # Steering / interrupting in-flight work (which + # also covers pending clarify + tool approvals + # held by the running agent). + _estop_allow = True + if not _estop_allow: + from tools import slash_confirm as _estop_confirm_mod + if _estop_confirm_mod.get_pending(_estop_key): + _estop_allow = True + if not _estop_allow: + from tools.approval import ( + has_blocking_approval as _estop_has_approval, + ) + if _estop_has_approval(_estop_key): + _estop_allow = True + except Exception: + pass + if not _estop_allow: + logger.info( + "Gateway turn paused by global emergency stop (platform=%s chat=%s)", + getattr(getattr(source, "platform", None), "value", "unknown"), + getattr(source, "chat_id", None) or "unknown", + ) + return _paused_notice # Intercept messages that are responses to a pending /update prompt. # The update process (detached) wrote .update_prompt.json; the watcher @@ -15302,6 +15384,9 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: canonical = _cmd_def.name if _cmd_def else command break + if canonical == "pause": + return await self._handle_pause_command(event) + if canonical == "new": if await asyncio.to_thread(self._is_telegram_topic_root_lobby, source): return self._telegram_topic_root_new_message() diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py index f3323f020ae1d..8dde644b67e92 100644 --- a/hermes_cli/commands.py +++ b/hermes_cli/commands.py @@ -139,6 +139,9 @@ class CommandDef: cli_only=True, args_hint=" [--name ]"), CommandDef("stop", "Kill all running background processes", "Session", busy_policy="interrupt_then_dispatch", busy_handler="stop"), + CommandDef("pause", "Pause new work globally (emergency stop); '/pause off' resumes", "Session", + gateway_only=True, args_hint="[reason | off]", + busy_policy="dispatch"), CommandDef("approve", "Approve a pending dangerous command", "Session", gateway_only=True, args_hint="[session|always]", busy_policy="dispatch"), CommandDef("deny", "Deny a pending dangerous command (optionally with a reason)", "Session", @@ -1272,7 +1275,9 @@ def discord_skill_commands_by_category( # - refine: on-demand memory/skill review; reached via /hermes refine on # Slack. Added at the 50-cap — a native slot would clamp an existing # native slash. -_SLACK_VIA_HERMES_ONLY = frozenset({"topup", "moa", "debug", "egress", "init", "version", "diff", "update", "heartbeat", "refine"}) +# - pause: global emergency stop; reached via /hermes pause [off] on +# Slack. Added at the 50-cap — a native slot would clamp /platform. +_SLACK_VIA_HERMES_ONLY = frozenset({"topup", "moa", "debug", "egress", "init", "version", "diff", "update", "heartbeat", "refine", "pause"}) def _sanitize_slack_name(raw: str) -> str: diff --git a/tests/cron/test_monitor_kind.py b/tests/cron/test_monitor_kind.py index b9a1900e7e053..714700b450e48 100644 --- a/tests/cron/test_monitor_kind.py +++ b/tests/cron/test_monitor_kind.py @@ -153,6 +153,90 @@ def test_create_job_monitor_rejected_with_no_agent(hermes_env): ) +def test_update_job_rejects_no_agent_on_monitor_job(hermes_env): + """The create-time monitor×no_agent invariant must hold through the + update door too — the scheduler's no_agent short-circuit runs before + the monitor gate, so flipping no_agent=True on a monitor job would + silently disable the monitor (post-merge audit of #81138).""" + from cron.jobs import create_job, update_job + + _write_script(hermes_env, "mon.sh", "echo stable\n") + _write_script(hermes_env, "w.sh", "echo hi\n") + job = create_job( + prompt="React to the change", + schedule="every 5m", + monitor_script="mon.sh", + deliver="local", + ) + with pytest.raises(ValueError, match="no_agent"): + update_job(job["id"], {"no_agent": True, "script": "w.sh"}) + + +def test_update_job_rejects_adding_monitor_to_no_agent_job(hermes_env): + from cron.jobs import create_job, update_job + + _write_script(hermes_env, "w.sh", "echo hi\n") + _write_script(hermes_env, "mon.sh", "echo stable\n") + job = create_job( + prompt=None, + schedule="every 5m", + script="w.sh", + no_agent=True, + deliver="local", + ) + with pytest.raises(ValueError, match="no_agent"): + update_job(job["id"], {"monitor_script": "mon.sh"}) + + +def test_update_job_rejects_second_monitor_source(hermes_env): + from cron.jobs import create_job, update_job + + _write_script(hermes_env, "mon.sh", "echo stable\n") + job = create_job( + prompt="React", + schedule="every 5m", + monitor_script="mon.sh", + deliver="local", + ) + with pytest.raises(ValueError, match="mutually exclusive"): + update_job(job["id"], {"monitor_url": "https://example.com/status"}) + + +def test_update_job_allows_clearing_monitor_then_no_agent(hermes_env): + """Clearing the monitor and flipping no_agent in ONE update is valid — + the invariant is checked on the merged record, not per-field.""" + from cron.jobs import create_job, get_job, update_job + + _write_script(hermes_env, "mon.sh", "echo stable\n") + _write_script(hermes_env, "w.sh", "echo hi\n") + job = create_job( + prompt="React", + schedule="every 5m", + monitor_script="mon.sh", + deliver="local", + ) + update_job(job["id"], {"monitor_script": "", "no_agent": True, "script": "w.sh"}) + reloaded = get_job(job["id"]) + assert reloaded.get("monitor_script") is None + assert reloaded["no_agent"] is True + + +def test_update_job_unrelated_fields_skip_mode_validation(hermes_env): + """A legacy/odd record must keep accepting updates that don't touch the + mode fields — the invariant re-check is scoped to changed fields.""" + from cron.jobs import create_job, update_job + + _write_script(hermes_env, "mon.sh", "echo stable\n") + job = create_job( + prompt="React", + schedule="every 5m", + monitor_script="mon.sh", + deliver="local", + ) + updated = update_job(job["id"], {"name": "renamed"}) + assert updated["name"] == "renamed" + + # --------------------------------------------------------------------------- # cron.monitor: hashing + diff unit behavior # --------------------------------------------------------------------------- diff --git a/tests/cron/test_notepad.py b/tests/cron/test_notepad.py index 8f2c7204ad1a6..9d140e50995d0 100644 --- a/tests/cron/test_notepad.py +++ b/tests/cron/test_notepad.py @@ -96,6 +96,44 @@ def test_clear_notepad_removes_all_keys_for_job_only(self, notepad): assert notepad.list_notes("job-1") == [] assert notepad.get_note("job-2", "a") == "keep" + def test_clear_notepad_noop_without_db_file(self, notepad): + """Clearing a job that never used the notepad must not create the DB + (remove_job calls clear_notepad unconditionally).""" + assert notepad.clear_notepad("never-used") == 0 + assert not notepad.NOTEPAD_FILE.exists() + + +class TestJobRemovalCleanup: + def test_remove_job_clears_notepad(self, cron_env, notepad): + """remove_job must clear the job's notepad rows — without this, + deleted jobs orphan their KV state in notepad.db forever + (post-merge audit of #81139: clear_notepad was dead code).""" + from cron.jobs import create_job, remove_job + + job = create_job(prompt="Check the feed", schedule="every 1h") + other = create_job(prompt="Other job", schedule="every 2h") + notepad.set_note(job["id"], "cursor", "page=7") + notepad.set_note(other["id"], "cursor", "keep") + + assert remove_job(job["id"]) is True + assert notepad.list_notes(job["id"]) == [] + # Sibling jobs' notepads are untouched. + assert notepad.get_note(other["id"], "cursor") == "keep" + + def test_remove_job_survives_notepad_failure(self, cron_env, notepad, monkeypatch): + """Notepad cleanup is best effort — a notepad error must never block + job removal itself.""" + from cron.jobs import create_job, get_job, remove_job + + job = create_job(prompt="Check the feed", schedule="every 1h") + + def _boom(job_id): + raise RuntimeError("disk on fire") + + monkeypatch.setattr(notepad, "clear_notepad", _boom) + assert remove_job(job["id"]) is True + assert get_job(job["id"]) is None + class TestNotepadCaps: def test_value_over_per_key_cap_rejected(self, notepad): diff --git a/tests/test_estop.py b/tests/test_estop.py index 820f9aa6c4464..126738daecfcb 100644 --- a/tests/test_estop.py +++ b/tests/test_estop.py @@ -285,3 +285,94 @@ def test_status_line_when_paused(hermes_home): assert "ops" in line estop.disengage() assert _estop_status_line() is None + + +# ── post-merge audit fixes (#81148 follow-up) ─────────────────────────────── + + +def test_is_engaged_fails_safe_on_stat_error(hermes_home, monkeypatch): + """A stat failure must report ENGAGED (fail safe) — the pause has to + hold even when HERMES_HOME is misbehaving, matching the module's + corrupt-sentinel doctrine.""" + class _BoomPath: + def exists(self): + raise OSError("permission denied") + + monkeypatch.setattr(estop, "sentinel_path", lambda: _BoomPath()) + assert estop.is_engaged() is True + + +class _FakeCmdEvent(_FakeEvent): + text = "/status" + + def get_command(self): + return "status" + + def get_command_args(self): + return "" + + +@pytest.mark.asyncio +async def test_gateway_slash_commands_bypass_estop(hermes_home): + """Recognized slash commands must pass the estop gate — /pause off is + the in-band resume path for messaging-only users, and /status, /help + and friends must keep working while paused.""" + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + runner._is_user_authorized = lambda source: True + estop.engage(reason="maintenance") + # The command proceeds past the estop gate; the bare runner then blows + # up further down on missing attributes — anything but the paused + # notice proves the gate let it through. + try: + reply = await runner._handle_message(_FakeCmdEvent()) + except Exception: + return + assert reply is None or "hermes is paused" not in (reply or "").lower() + + +class _FakePauseEvent(_FakeEvent): + def __init__(self, args=""): + super().__init__() + self._args = args + self.text = f"/pause {args}".strip() + + def get_command(self): + return "pause" + + def get_command_args(self): + return self._args + + +@pytest.mark.asyncio +async def test_gateway_pause_command_engages_and_resumes(hermes_home): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + + reply = await runner._handle_pause_command(_FakePauseEvent("deploy window")) + assert "paused" in reply.lower() + assert estop.is_engaged() is True + assert estop.get_state()["reason"] == "deploy window" + + # Re-issuing without args reports already-paused instead of clobbering. + reply = await runner._handle_pause_command(_FakePauseEvent("")) + assert "already paused" in reply.lower() + + reply = await runner._handle_pause_command(_FakePauseEvent("off")) + assert "resumed" in reply.lower() + assert estop.is_engaged() is False + + reply = await runner._handle_pause_command(_FakePauseEvent("off")) + assert "wasn't paused" in reply.lower() + + +def test_pause_command_registered_for_gateway(): + from hermes_cli.commands import GATEWAY_KNOWN_COMMANDS, resolve_command + + cmd = resolve_command("pause") + assert cmd is not None and cmd.name == "pause" + assert "pause" in GATEWAY_KNOWN_COMMANDS + # Must be dispatchable while an agent is running (in-band emergency stop). + assert cmd.busy_policy == "dispatch" diff --git a/tests/tools/test_delegate_batch_validation.py b/tests/tools/test_delegate_batch_validation.py index ef366972f2468..912c1b699ee94 100644 --- a/tests/tools/test_delegate_batch_validation.py +++ b/tests/tools/test_delegate_batch_validation.py @@ -50,22 +50,30 @@ def _call(tasks): 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()) +class TestBatchDuplicateGoalsAllowed(unittest.TestCase): + """Identical-goal fan-outs are legitimate (best-of-N / ensemble sampling). - 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()) + The original gate from #81141 rejected duplicates; the post-merge audit + downgraded that — duplicates must pass validation. + """ - 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"]) + def _completed(self, idx): + return {"task_index": idx, "status": "completed", "summary": "ok", + "api_calls": 1, "duration_seconds": 1.0, "_child_role": None} + + def test_exact_duplicate_goals_accepted(self): + with patch("tools.delegate_tool._run_single_child") as mock_run: + mock_run.side_effect = [self._completed(0), self._completed(1)] + result = _call([{"goal": GOOD_A}, {"goal": GOOD_A}]) + self.assertNotIn("error", result) + self.assertEqual(len(result["results"]), 2) + + def test_case_whitespace_variant_duplicates_accepted(self): + with patch("tools.delegate_tool._run_single_child") as mock_run: + mock_run.side_effect = [self._completed(0), self._completed(1)] + result = _call([{"goal": GOOD_A}, {"goal": " " + GOOD_A.upper() + " "}]) + self.assertNotIn("error", result) + self.assertEqual(len(result["results"]), 2) class TestBatchPlaceholderGoals(unittest.TestCase): @@ -91,6 +99,39 @@ def test_unexpanded_brace_template_marker_rejected(self): self.assertIn("error", result) self.assertIn("template", result["error"].lower()) + def test_code_shaped_brackets_not_rejected(self): + """Generics, HTML tags, JSON snippets, glob braces, and f-string-style + single-word placeholders are legitimate goal content — the narrow + marker regex (post-merge audit of #81141) must not fire on them.""" + code_goals = [ + "Refactor the parser to return Vec instead of raw pointers", + "Fix the Result error propagation in the config loader", + "Render the sidebar inside a
wrapper with flex layout", + 'Update the fixture to emit {"key": 1} for the happy path', + "Add a glob rule matching src/{a,b}/*.py to the lint config", + "Rewrite the loop so {i} interpolates via f-strings correctly", + ] + for bad_free_goal in code_goals: + with patch("tools.delegate_tool._run_single_child") as mock_run: + mock_run.side_effect = [ + {"task_index": 0, "status": "completed", "summary": "ok", + "api_calls": 1, "duration_seconds": 1.0, "_child_role": None}, + {"task_index": 1, "status": "completed", "summary": "ok", + "api_calls": 1, "duration_seconds": 1.0, "_child_role": None}, + ] + result = _call([{"goal": GOOD_A}, {"goal": bad_free_goal}]) + self.assertNotIn("error", result, bad_free_goal) + + def test_multiword_placeholder_shapes_still_rejected(self): + for marker_goal in ( + "Deploy the service to when ready", + "Backfill rows for {customer id} in the billing table", + "Ship behind the beta flag", + ): + result = _call([{"goal": GOOD_A}, {"goal": marker_goal}]) + self.assertIn("error", result, marker_goal) + 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) diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py index 951ea15aff5ea..25b8c81d07039 100644 --- a/tools/delegate_tool.py +++ b/tools/delegate_tool.py @@ -3058,8 +3058,19 @@ def _recover_tasks_from_json_string( # Placeholder shapes for batch goal validation: bare 'TODO', bare 'task N' # labels, or goals still carrying unexpanded template markers. +# +# The marker regex is deliberately NARROW: it only fires on snake_case / +# space-separated placeholder identifiers (``, `{file path}`, +# ``) — the shape LLM templates actually leave behind. Bare +# single-word brackets are left alone because legitimate coding goals are +# full of them: generics (`Vec`, `Result`), HTML tags (`
`), +# JSON/dict snippets (`{"key": 1}`), glob braces (`{a,b}`), and f-string +# style (`{i}`) must never be rejected (post-merge audit of #81141). _PLACEHOLDER_GOAL_RE = re.compile(r"^(todo|task\s*\d+)$", re.IGNORECASE) -_TEMPLATE_MARKER_RE = re.compile(r"<[^<>]+>|\{[^{}]+\}") +_TEMPLATE_MARKER_RE = re.compile( + r"<[A-Za-z][A-Za-z0-9]*(?:[ _-][A-Za-z0-9]+)+>" + r"|\{[A-Za-z][A-Za-z0-9]*(?:[ _-][A-Za-z0-9]+)+\}" +) _MIN_BATCH_GOAL_LEN = 10 @@ -3069,6 +3080,10 @@ def _validate_batch_tasks(task_list: List[Dict[str, Any]]) -> Optional[str]: 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. + + Duplicate goals are deliberately NOT rejected: identical-goal fan-outs + are a legitimate pattern (best-of-N / ensemble sampling), and blocking + them broke real workflows (post-merge audit of #81141). """ if len(task_list) < 2: return ( @@ -3077,21 +3092,10 @@ def _validate_batch_tasks(task_list: List[Dict[str, Any]]) -> Optional[str]: '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 " @@ -3241,9 +3245,10 @@ 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. + # Batch-only quality gate: catch malformed fan-outs (placeholder goals, + # unexpanded multi-word template markers, 1-task batches) before any + # child is spawned. The single-`goal` form is deliberately exempt — + # short goals are valid there. Duplicate goals are allowed (best-of-N). # 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)