fix: post-merge audit follow-ups for #81138/#81139/#81141/#81148

Four fix-forwards from the adversarial post-merge audit of the Aug 7
unreviewed merge batch:

- estop (#81148): is_engaged() now fails SAFE (engaged) on stat errors;
  the gateway estop gate lets recognized slash commands and replies owned
  by in-flight work (update prompts, clarify, slash-confirm, tool
  approvals, running sessions) through instead of consuming them; new
  gateway /pause [reason|off] command gives messaging-only operators an
  in-band engage/resume path (busy_policy=dispatch so it works mid-run).
- cron monitor mode (#81138): execution-mode invariants (monitor x
  no_agent, monitor_script x monitor_url, no_agent-requires-script) now
  have ONE owner (_validate_job_mode_invariants) called from BOTH
  create_job and update_job, so the create-time invariant can no longer
  be silently violated through the update door.
- cron notepad (#81139): remove_job now clears the job's notepad rows
  (clear_notepad was dead code -> orphaned KV state forever); clear is
  best-effort and no-ops without creating notepad.db.
- delegation batch gate (#81141): template-marker regex narrowed to
  multi-word placeholder shapes only (<feature name>, {file_path}) so
  generics (Vec<T>), HTML tags, JSON snippets, glob braces and f-string
  style no longer reject legitimate batches; duplicate-goal rejection
  removed (best-of-N fan-outs are legitimate).

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:

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
 

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),)

gateway/run.py

@@ -14364,6 +14364,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 "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 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             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 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # 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 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             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 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     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()

hermes_cli/commands.py

@@ -139,6 +139,9 @@ COMMAND_REGISTRY: list[CommandDef] = [
                cli_only=True, args_hint="<archive.tar.gz> [--name <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 @@ _SLACK_PRIORITY_ALIASES = ("btw", "bg")
 #   - 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:

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
 # ---------------------------------------------------------------------------

tests/cron/test_notepad.py

@@ -96,6 +96,44 @@ class TestNotepadCrud:
         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):

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"

tests/tools/test_delegate_batch_validation.py

@@ -50,22 +50,30 @@ 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())
+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 @@ class TestBatchPlaceholderGoals(unittest.TestCase):
         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<T> instead of raw pointers",
+            "Fix the Result<String> error propagation in the config loader",
+            "Render the sidebar inside a <div> 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 <target environment> when ready",
+            "Backfill rows for {customer id} in the billing table",
+            "Ship <FEATURE-NAME> 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)

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 (`<feature_name>`, `{file path}`,
+# `<FEATURE-NAME>`) — the shape LLM templates actually leave behind. Bare
+# single-word brackets are left alone because legitimate coding goals are
+# full of them: generics (`Vec<T>`, `Result<String>`), HTML tags (`<div>`),
+# 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)