fix(metacog): N>=3 streak detector + drop redundant error-prefix list

Cleanup pass on the metacognitive nudge stack — restores pre-split
errored-counts-toward-repeat behaviour and tightens the is_error
plumbing through the per-batch advisory hook.

The per-batch hook in ``_run_loop`` was duplicating the is_error
signal: ``self._tool_error_flags`` (set by ``_report_tool_result``)
and a string-prefix tuple (``Error`` / ``JSON parse error`` / …).
Two truth sources is what got us here — bash commands that exit
non-zero with normal stdout matched the flag but not the prefix,
the deny path matched the prefix but not the flag, and the result
was that stuck-loop detection silently broke for the most common
failure mode (the model bashing the same broken command).

Single source of truth now:

- ``_execute_tools.run_one`` deny branch routes through
  ``_report_tool_result(is_error=True)`` so denied calls populate
  ``_tool_error_flags`` like every other error path.
- The error-prefix tuple is gone; the write-success-clear gate and
  the tool-error-nudge gate both read ``_tool_error_flags`` only.

Repeat-detection state moves from a ``set[str]`` (fired on the second
identical call, ignored errors entirely) to a ``RepeatDetector``
helper in ``metacognition.py`` with consecutive-streak semantics:

- Threshold raised from 2 to 3 — two-in-a-row was noisy on
  legitimate transient retries; three is the cheapest stuck-loop
  signal.
- Recording a different signature resets the count, so [A, A, B, A]
  is two short streaks of 2 and not a streak of 4. Bounded by O(1)
  state regardless of session length.
- Errored calls now count toward the streak (the split into a
  separate metacog module unintentionally introduced a "skip errors"
  branch — restored).

While there:

- ``metacognition._COOLDOWN_SECS`` default aligned to 300s (matches
  ``MemoryConfig.nudge_cooldown`` and the ``memory.nudge_cooldown``
  config-store default; was set to 30 by an earlier investigation).
- The per-batch advisory block (~80 lines of mixed orchestration
  inside ``_run_loop``) is extracted to
  ``ChatSession._apply_post_execute_advisories`` so the wired
  behaviour is testable without driving ``_run_loop`` end-to-end.
  Producer extraction to a dedicated module is deferred to a
  follow-up; advisory producers all live on ``ChatSession`` for
  now per existing convention.
- Frontend ``appendToolOutput`` (turnstone/ui/static/app.js) now
  skips rendering when the parent approval block is denied or
  the output starts with ``Denied by user`` / ``Blocked``,
  mirroring the history-replay guard at ``_build_history``.
  Previously the live SSE path didn't need this guard because
  the deny path never emitted a ``tool_result`` event; the
  is_error routing change above means it does now, so without
  this guard the badge from ``resolveApproval`` and the SSE
  output would both render.

Tests: 8 unit tests for ``RepeatDetector`` covering streak,
threshold, clear, and intervening-sig reset; 9 integration tests
for ``_apply_post_execute_advisories`` covering the wired
behaviour (3-identical fires warning + advisory + UI line, errored
calls count toward streak as a regression guard, intervening sig
resets streak, successful write clears, failed write does not,
JSON outputs tracked but not inline-warned, tool_error nudge gates
on memory_count, repeat UI line emitted on streak fire).

tests/test_metacognition.py

@@ -8,6 +8,7 @@ from turnstone.core.metacognition import (
     NUDGE_RESUME,
     NUDGE_START,
     NUDGE_TOOL_ERROR,
+    RepeatDetector,
     detect_completion,
     detect_correction,
     format_nudge,
@@ -308,3 +309,70 @@ class TestRepeatNudge:
         """Repeat nudge should fire even with zero memories."""
         state: dict[str, float] = {}
         assert should_nudge("repeat", state, message_count=5, memory_count=0) is True
+
+
+class TestRepeatDetector:
+    """Repeat-detection streak machine — fires only when the same signature
+    is recorded ``threshold`` times *consecutively* (default 3).  Recording
+    any different signature resets the streak, so an interrupted repeat
+    isn't flagged as a stuck loop."""
+
+    def test_below_threshold_does_not_fire(self):
+        det = RepeatDetector()
+        assert det.record("a") is False
+        assert det.record("a") is False  # second call still under threshold
+
+    def test_at_threshold_fires(self):
+        det = RepeatDetector()
+        det.record("a")
+        det.record("a")
+        assert det.record("a") is True
+
+    def test_continues_to_fire_past_threshold(self):
+        # Caller is responsible for clearing after a fire — until they do,
+        # subsequent identical calls keep returning True.
+        det = RepeatDetector()
+        det.record("a")
+        det.record("a")
+        assert det.record("a") is True
+        assert det.record("a") is True
+
+    def test_clear_resets_count(self):
+        det = RepeatDetector()
+        det.record("a")
+        det.record("a")
+        det.clear()
+        assert det.record("a") is False  # back to 1 after clear
+
+    def test_intervening_sig_resets_streak(self):
+        # The streak is consecutive: recording any other sig mid-streak
+        # discards the in-progress count.  An alternating pattern like
+        # [A, A, B, A, A] is two short streaks of 2, not a streak of 4.
+        det = RepeatDetector()
+        det.record("a")
+        det.record("a")
+        assert det.record("b") is False  # b at count 1; a's streak is gone
+        assert det.record("a") is False  # a starts fresh at 1
+        assert det.record("a") is False  # a at 2
+        assert det.record("a") is True  # a hits 3 — fresh streak completes
+
+    def test_errored_signature_counts_toward_repeat(self):
+        # Regression: when metacog was split out of the system message,
+        # the error-output skip got reintroduced and stuck-loop detection
+        # silently broke for tools that kept failing.  Detector itself is
+        # signature-only — error vs. success is the caller's policy.
+        det = RepeatDetector()
+        # Caller records an errored call's sig the same as a successful one;
+        # the streak is what matters.
+        for _ in range(3):
+            last = det.record("bash:ls /nonexistent")
+        assert last is True
+
+    def test_custom_threshold(self):
+        det = RepeatDetector(threshold=2)
+        assert det.record("a") is False
+        assert det.record("a") is True
+
+    def test_threshold_one_fires_immediately(self):
+        det = RepeatDetector(threshold=1)
+        assert det.record("a") is True

tests/test_session.py

@@ -2190,3 +2190,188 @@ class TestMetacognitiveBuffers:
 
         # Buffer cleared by the cancel handler — no leak into next send().
         assert session._pending_tool_advisories == []
+
+
+class TestApplyPostExecuteAdvisories:
+    """End-to-end coverage of the per-batch advisory hook in _run_loop —
+    repeat detection (with the streak semantics restored after the split)
+    and tool-error nudge.  Drives ``_apply_post_execute_advisories``
+    directly, simulating the post-_execute_tools state.
+    """
+
+    @staticmethod
+    def _tc(tc_id: str, name: str, args: str) -> dict:
+        return {"id": tc_id, "function": {"name": name, "arguments": args}}
+
+    @staticmethod
+    def _prime(session) -> None:
+        """Enable nudges and bump message_count above the should_nudge floor.
+
+        ``should_nudge`` skips nudging on message_count <= 1; in production
+        the per-batch hook runs after at least a user→assistant exchange,
+        so seed two messages to mirror that.
+        """
+        session._mem_cfg.nudges = True
+        session.messages.append({"role": "user", "content": "hi"})
+        session.messages.append({"role": "assistant", "content": "ok"})
+
+    def test_three_identical_calls_fire_warning_and_advisory(self, tmp_db):
+        session = _make_session()
+        self._prime(session)
+        for i in range(3):
+            tc_id = f"tc_{i}"
+            results = [(tc_id, "file contents")]
+            session._apply_post_execute_advisories(
+                [self._tc(tc_id, "read_file", '{"path": "x"}')],
+                results,
+            )
+            if i < 2:
+                # Streak below threshold — no inline warning, no advisory yet.
+                assert results[0][1] == "file contents"
+                assert all(t != "repeat" for t, _ in session._pending_tool_advisories)
+            else:
+                assert "⚠ Warning: this is an identical repeat" in results[0][1]
+        assert any(t == "repeat" for t, _ in session._pending_tool_advisories)
+
+    def test_errored_calls_count_toward_streak(self, tmp_db):
+        """Regression: when metacog was split out of the system message,
+        errored tool calls stopped counting toward repeats — so a model
+        stuck on a failing call wouldn't get warned. Three identical
+        bash failures must still fire the streak."""
+        session = _make_session()
+        self._prime(session)
+        with patch.object(session, "_visible_memory_count", return_value=0):
+            for i in range(3):
+                tc_id = f"tc_{i}"
+                session._tool_error_flags[tc_id] = True
+                session._apply_post_execute_advisories(
+                    [self._tc(tc_id, "bash", '{"command": "ls /missing"}')],
+                    [(tc_id, "ls: cannot access /missing")],
+                )
+        assert any(t == "repeat" for t, _ in session._pending_tool_advisories)
+
+    def test_intervening_different_sig_resets_streak(self, tmp_db):
+        """Streak semantics: [A, A, B, A] does NOT fire — B breaks the run."""
+        session = _make_session()
+        self._prime(session)
+        sequence = [
+            ("read_file", '{"path": "a"}'),
+            ("read_file", '{"path": "a"}'),
+            ("read_file", '{"path": "b"}'),  # different — resets
+            ("read_file", '{"path": "a"}'),
+        ]
+        with patch.object(session, "_visible_memory_count", return_value=0):
+            for i, (name, args) in enumerate(sequence):
+                tc_id = f"tc_{i}"
+                session._apply_post_execute_advisories(
+                    [self._tc(tc_id, name, args)],
+                    [(tc_id, "ok")],
+                )
+        assert all(t != "repeat" for t, _ in session._pending_tool_advisories)
+
+    def test_successful_write_clears_streak(self, tmp_db):
+        """A successful write tool changes file state, so re-reading the
+        same path is valid again — streak clears."""
+        session = _make_session()
+        self._prime(session)
+        with patch.object(session, "_visible_memory_count", return_value=0):
+            for i in range(2):
+                tc_id = f"r_{i}"
+                session._apply_post_execute_advisories(
+                    [self._tc(tc_id, "read_file", '{"path": "x"}')],
+                    [(tc_id, "contents")],
+                )
+            # Successful write — clears streak.
+            session._apply_post_execute_advisories(
+                [self._tc("w", "write_file", '{"path": "x", "content": "y"}')],
+                [("w", "ok")],
+            )
+            # Two more identical reads — would have been streak=4 without
+            # the clear, so a fire would prove it didn't clear.  Streak
+            # restarts at 1 instead.
+            for i in range(2):
+                tc_id = f"r2_{i}"
+                session._apply_post_execute_advisories(
+                    [self._tc(tc_id, "read_file", '{"path": "x"}')],
+                    [(tc_id, "contents")],
+                )
+        assert all(t != "repeat" for t, _ in session._pending_tool_advisories)
+
+    def test_failed_write_does_not_clear_streak(self, tmp_db):
+        """A failing bash command does NOT change state, so the streak
+        must persist — otherwise a model bashing the same broken
+        command never gets warned."""
+        session = _make_session()
+        self._prime(session)
+        with patch.object(session, "_visible_memory_count", return_value=0):
+            for i in range(3):
+                tc_id = f"b_{i}"
+                session._tool_error_flags[tc_id] = True
+                session._apply_post_execute_advisories(
+                    [self._tc(tc_id, "bash", '{"command": "ls /missing"}')],
+                    [(tc_id, "ls: cannot access /missing")],
+                )
+        assert any(t == "repeat" for t, _ in session._pending_tool_advisories)
+
+    def test_json_output_tracked_but_not_inline_warned(self, tmp_db):
+        """MCP-shape JSON outputs are tracked toward the streak but the
+        warning text is NOT appended — that would corrupt the payload."""
+        session = _make_session()
+        self._prime(session)
+        json_out = '{"result": "data"}'
+        with patch.object(session, "_visible_memory_count", return_value=0):
+            for i in range(3):
+                tc_id = f"j_{i}"
+                results = [(tc_id, json_out)]
+                session._apply_post_execute_advisories(
+                    [self._tc(tc_id, "search", '{"q": "x"}')],
+                    results,
+                )
+                if i == 2:
+                    # JSON content untouched even though streak fired.
+                    assert results[0][1] == json_out
+        assert any(t == "repeat" for t, _ in session._pending_tool_advisories)
+
+    def test_tool_error_nudge_fires_when_memories_exist(self, tmp_db):
+        session = _make_session()
+        self._prime(session)
+        tc_id = "tc"
+        session._tool_error_flags[tc_id] = True
+        with patch.object(session, "_visible_memory_count", return_value=3):
+            session._apply_post_execute_advisories(
+                [self._tc(tc_id, "bash", '{"command": "false"}')],
+                [(tc_id, "command failed")],
+            )
+        assert any(t == "tool_error" for t, _ in session._pending_tool_advisories)
+
+    def test_tool_error_nudge_skipped_with_zero_memories(self, tmp_db):
+        """Without memories the tool_error nudge has nothing useful to point
+        at — should_nudge gates it off."""
+        session = _make_session()
+        self._prime(session)
+        tc_id = "tc"
+        session._tool_error_flags[tc_id] = True
+        with patch.object(session, "_visible_memory_count", return_value=0):
+            session._apply_post_execute_advisories(
+                [self._tc(tc_id, "bash", '{"command": "false"}')],
+                [(tc_id, "command failed")],
+            )
+        assert all(t != "tool_error" for t, _ in session._pending_tool_advisories)
+
+    def test_emit_repeat_ui_line_on_streak_fire(self, tmp_db):
+        """The grey ``[repeat: tool() called with same arguments]`` UI
+        line is the user-visible signal that the warning fired."""
+        session = _make_session()
+        self._prime(session)
+        with (
+            patch.object(session.ui, "on_info") as m_info,
+            patch.object(session, "_visible_memory_count", return_value=0),
+        ):
+            for i in range(3):
+                tc_id = f"tc_{i}"
+                session._apply_post_execute_advisories(
+                    [self._tc(tc_id, "read_file", '{"path": "x"}')],
+                    [(tc_id, "ok")],
+                )
+        msgs = [c.args[0] for c in m_info.call_args_list]
+        assert any("[repeat: read_file()" in m for m in msgs)

turnstone/core/metacognition.py

@@ -5,7 +5,50 @@ from __future__ import annotations
 import re
 import time
 
-_COOLDOWN_SECS = 300  # 5 minutes between nudges of the same type
+# Default cooldown (s) between nudges of the same type.  Production
+# paths pass ``cooldown_secs`` explicitly from
+# ``MemoryConfig.nudge_cooldown`` (config-store ``memory.nudge_cooldown``,
+# default 300); this constant is the fallback for tests and unit-style
+# callers without a ``MemoryConfig`` and is kept aligned with that
+# canonical default so both paths behave the same.
+_COOLDOWN_SECS = 300
+
+# Repeat-detection threshold — number of *consecutive* identical tool
+# calls (same name + same arguments) before a repeat warning fires.
+# Two-in-a-row is too noisy because legitimate retries on transient
+# failures look identical; three-in-a-row is the cheapest signal that
+# the model is stuck on the same call.
+_REPEAT_THRESHOLD = 3
+
+
+class RepeatDetector:
+    """Detect a streak of identical tool-call signatures.
+
+    ``record(sig)`` returns ``True`` once *sig* has been recorded
+    ``threshold`` times in a row (default 3).  Recording a different
+    signature resets the streak — interleaved tool calls aren't a
+    stuck loop, only repeated identical ones are.  After a fire, the
+    caller is expected to call ``clear()`` to start a fresh streak.
+    """
+
+    def __init__(self, threshold: int = _REPEAT_THRESHOLD) -> None:
+        self._threshold = threshold
+        self._sig: str | None = None
+        self._count = 0
+
+    def record(self, sig: str) -> bool:
+        """Record *sig*; return ``True`` when the streak hits the threshold."""
+        if sig == self._sig:
+            self._count += 1
+        else:
+            self._sig = sig
+            self._count = 1
+        return self._count >= self._threshold
+
+    def clear(self) -> None:
+        self._sig = None
+        self._count = 0
+
 
 # ---------------------------------------------------------------------------
 # Nudge messages (brief, model-facing hints)

turnstone/core/session.py

@@ -81,6 +81,7 @@ from turnstone.core.memory_relevance import (
     score_memories,
 )
 from turnstone.core.metacognition import (
+    RepeatDetector,
     detect_completion,
     detect_correction,
     format_nudge,
@@ -474,8 +475,12 @@ class ChatSession:
             collections.OrderedDict()
         )
         self._queued_lock = threading.Lock()
-        # Repeat detection: track recent tool call signatures
-        self._recent_tool_sigs: set[str] = set()
+        # Repeat detection: streak counter over tool-call signatures.
+        # Fires when a (name, args) signature has been seen N times in
+        # a row; recording any different signature resets the streak.
+        # Also cleared after a write tool succeeds (state changed) or
+        # after a warning fires (clean slate, re-fire on the next streak).
+        self._repeat_detector = RepeatDetector()
         # Tool error tracking: call_id → is_error for message persistence
         self._tool_error_flags: dict[str, bool] = {}
         # Cooperative cancellation: set from outside to stop generation
@@ -1293,7 +1298,7 @@ class ChatSession:
             self._ws_id = ws_id
         self.messages = messages
         self._read_files.clear()
-        self._recent_tool_sigs.clear()
+        self._repeat_detector.clear()
         self._last_usage = None
         self._calibrated_msg_count = 0
         self._title_generated = True  # don't re-title resumed workstreams
@@ -2332,92 +2337,10 @@ class ChatSession:
                 if self._generation != my_generation:
                     return
 
-                # Repeat detection: warn when a tool is called with identical args.
-                # Skip error outputs — retrying a failed tool is valid.
-                # Skip JSON outputs (MCP structured results) — appending
-                # text would corrupt the payload.
-                _tc_by_id = {c["id"]: c for c in tool_calls}
-                _repeat_detected = False
-                _error_prefixes = (
-                    "Error",
-                    "JSON parse error",
-                    "Unknown tool",
-                    "Command timed out",
-                    "Blocked:",
-                    "Denied",
-                )
-
-                # Clear dedup sigs when a write tool executed successfully —
-                # the state has changed so re-running a read tool is valid.
-                _write_tools = frozenset({"write_file", "edit_file", "bash"})
-                if any(
-                    tc["function"]["name"] in _write_tools
-                    and not any(
-                        cid == tc["id"] and isinstance(out, str) and out.startswith(_error_prefixes)
-                        for cid, out in results
-                    )
-                    for tc in tool_calls
-                ):
-                    self._recent_tool_sigs.clear()
-                for i, (tc_id, output) in enumerate(results):
-                    tc = _tc_by_id.get(tc_id)
-                    if tc and isinstance(output, str) and not output.startswith(_error_prefixes):
-                        raw = tc["function"]["name"] + ":" + tc["function"]["arguments"]
-                        sig = hashlib.sha256(raw.encode()).hexdigest()
-                        is_json = output.lstrip().startswith(("{", "["))
-                        if sig in self._recent_tool_sigs:
-                            _repeat_detected = True
-                            if not is_json:
-                                output += (
-                                    "\n\n⚠ Warning: this is an identical repeat of a "
-                                    "previous tool call. The result is the same. "
-                                    "Try a different approach."
-                                )
-                                results[i] = (tc_id, output)
-                            self.ui.on_info(
-                                f"{GRAY}[repeat: {tc['function']['name']}() "
-                                f"called with same arguments]{RESET}"
-                            )
-                        self._recent_tool_sigs.add(sig)
-                if _repeat_detected:
-                    # Reset so the model gets a clean slate after the warning.
-                    # If it repeats again, a new warning fires.
-                    self._recent_tool_sigs.clear()
-                    if self._mem_cfg.nudges and should_nudge(
-                        "repeat",
-                        self._metacog_state,
-                        message_count=len(self.messages),
-                        cooldown_secs=self._mem_cfg.nudge_cooldown,
-                    ):
-                        self._queue_tool_advisory("repeat", format_nudge("repeat"))
-
-                # Tool-error nudge — checked here (pre-iteration) so the
-                # MetacognitiveAdvisory rides the same _collect_advisories
-                # drain pass that handles guard findings and user
-                # interjections.  Cooldown gating in should_nudge keeps
-                # this to one nudge per batch even with many failing
-                # tools.
-                if (
-                    self._mem_cfg.nudges
-                    and any(
-                        isinstance(out, str)
-                        and (
-                            out.startswith("Error")
-                            or " error: " in out[:50]
-                            or out.startswith("Command timed out")
-                            or out.startswith("Unknown tool:")
-                        )
-                        for _, out in results
-                    )
-                    and should_nudge(
-                        "tool_error",
-                        self._metacog_state,
-                        message_count=len(self.messages),
-                        memory_count=self._visible_memory_count(),
-                        cooldown_secs=self._mem_cfg.nudge_cooldown,
-                    )
-                ):
-                    self._queue_tool_advisory("tool_error", format_nudge("tool_error"))
+                # Repeat-detection + tool-error nudge.  Mutates *results*
+                # in place to inject inline warning text on identical
+                # repeats; queues advisories for the next drain pass.
+                self._apply_post_execute_advisories(tool_calls, results)
 
                 # Map tool_call_id → tool name for logging
                 from turnstone.core.tool_advisory import wrap_tool_result
@@ -3391,7 +3314,7 @@ class ChatSession:
         self.messages = [summary_user, summary_asst]
         # File contents are gone after compaction — force re-read before edit_file
         self._read_files.clear()
-        self._recent_tool_sigs.clear()
+        self._repeat_detector.clear()
 
         # Rebuild token table
         su_tok = max(1, int(self._msg_char_count(summary_user) / self._chars_per_token))
@@ -3974,7 +3897,14 @@ class ChatSession:
                 )
                 return item["call_id"], item["error"]
             if item.get("denied"):
-                return item["call_id"], item.get("denial_msg", "Denied by user")
+                msg = item.get("denial_msg", "Denied by user")
+                self._report_tool_result(
+                    item["call_id"],
+                    item.get("func_name", "unknown"),
+                    msg,
+                    is_error=True,
+                )
+                return item["call_id"], msg
             try:
                 result: tuple[str, str | list[dict[str, Any]]] = item["execute"](item)
                 return result
@@ -5441,6 +5371,94 @@ class ChatSession:
         """
         self._pending_tool_advisories.append((nudge_type, text))
 
+    def _apply_post_execute_advisories(
+        self,
+        tool_calls: list[dict[str, Any]],
+        results: list[tuple[str, str | list[dict[str, Any]]]],
+    ) -> None:
+        """Run repeat detection + tool-error nudge over a freshly-executed batch.
+
+        Mutates *results* in place when an identical-repeat warning is
+        appended to a tool's text output.  Updates ``self._repeat_detector``,
+        ``self._pending_tool_advisories``, ``self._metacog_state`` (cooldown
+        timestamp via ``should_nudge``), and emits a ``[repeat: …]`` UI line
+        on each detection.
+
+        ``_tool_error_flags`` is the authoritative is_error signal — set by
+        ``_report_tool_result`` for every error path including bash non-zero
+        exits with normal stdout, denials, parse errors, and blocked
+        commands.  Flags are read here and consumed by ``.pop()`` later in
+        the per-result loop in ``_run_loop``, so the read here is safe.
+        """
+        # Repeat detection: warn when a tool is called with identical
+        # args as a previous call in this session, regardless of
+        # whether the previous call succeeded or failed.  Repeatedly
+        # calling the same tool — even one that keeps erroring — is
+        # the stuck-loop behaviour the nudge is meant to catch.  JSON
+        # outputs (MCP structured results) are tracked but exempt
+        # from the inline warning text (appending text would corrupt
+        # the payload).
+        _tc_by_id = {c["id"]: c for c in tool_calls}
+        _repeat_detected = False
+
+        # Clear streak when a write tool executed successfully — the
+        # state has changed so re-running a read tool is valid.
+        _write_tools = frozenset({"write_file", "edit_file", "bash"})
+        if any(
+            tc["function"]["name"] in _write_tools and not self._tool_error_flags.get(tc["id"])
+            for tc in tool_calls
+        ):
+            self._repeat_detector.clear()
+
+        for i, (tc_id, output) in enumerate(results):
+            tc = _tc_by_id.get(tc_id)
+            if tc and isinstance(output, str):
+                raw = tc["function"]["name"] + ":" + tc["function"]["arguments"]
+                sig = hashlib.sha256(raw.encode()).hexdigest()
+                is_json = output.lstrip().startswith(("{", "["))
+                if self._repeat_detector.record(sig):
+                    _repeat_detected = True
+                    if not is_json:
+                        output += (
+                            "\n\n⚠ Warning: this is an identical repeat of a "
+                            "previous tool call. The result is the same. "
+                            "Try a different approach."
+                        )
+                        results[i] = (tc_id, output)
+                    self.ui.on_info(
+                        f"{GRAY}[repeat: {tc['function']['name']}() "
+                        f"called with same arguments]{RESET}"
+                    )
+
+        if _repeat_detected:
+            # Reset so the model gets a clean slate after the warning.
+            # If it repeats again, a new warning fires.
+            self._repeat_detector.clear()
+            if self._mem_cfg.nudges and should_nudge(
+                "repeat",
+                self._metacog_state,
+                message_count=len(self.messages),
+                cooldown_secs=self._mem_cfg.nudge_cooldown,
+            ):
+                self._queue_tool_advisory("repeat", format_nudge("repeat"))
+
+        # Tool-error nudge — queued so the MetacognitiveAdvisory rides
+        # the same _collect_advisories drain pass as guard findings and
+        # user interjections.  Cooldown gating in should_nudge keeps
+        # this to one nudge per batch even with many failing tools.
+        if (
+            self._mem_cfg.nudges
+            and any(self._tool_error_flags.get(tc_id) for tc_id, _ in results)
+            and should_nudge(
+                "tool_error",
+                self._metacog_state,
+                message_count=len(self.messages),
+                memory_count=self._visible_memory_count(),
+                cooldown_secs=self._mem_cfg.nudge_cooldown,
+            )
+        ):
+            self._queue_tool_advisory("tool_error", format_nudge("tool_error"))
+
     # ------------------------------------------------------------------
     # Coordinator tools — reachable only when ``kind == "coordinator"``.
     # All six dispatch through ``self._coord_client`` which is None when
@@ -9203,7 +9221,7 @@ class ChatSession:
         elif cmd == "/clear":
             self.messages.clear()
             self._read_files.clear()
-            self._recent_tool_sigs.clear()
+            self._repeat_detector.clear()
             self._last_usage = None
             self._calibrated_msg_count = 0
             self._msg_tokens = []
@@ -9214,7 +9232,7 @@ class ChatSession:
 
             self.messages.clear()
             self._read_files.clear()
-            self._recent_tool_sigs.clear()
+            self._repeat_detector.clear()
             self._last_usage = None
             self._calibrated_msg_count = 0
             self._msg_tokens = []

turnstone/ui/static/app.js

@@ -1318,6 +1318,19 @@ Pane.prototype.appendToolOutput = function (callId, name, output, isError) {
   var stripped = stripAnsi(output || "").trim();
   if (!stripped) return;
 
+  // Skip rendering for denied/blocked tool results — the ✗ denied
+  // badge from resolveApproval already shows the denial reason; the
+  // SSE tool_result event would otherwise duplicate the text.  Mirror
+  // the guard in the history-replay path (the live path used to be
+  // safe because no tool_result event was ever emitted for denied
+  // items, but we now emit one so _tool_error_flags gets set).
+  var parentBlock = target.closest(".ts-approval");
+  var isDenied =
+    (parentBlock && parentBlock.classList.contains("denied")) ||
+    /^Denied by user/.test(stripped) ||
+    /^Blocked/.test(stripped);
+  if (isDenied) return;
+
   // Detect structured media output and render interactive embed
   if (!isError) {
     var media = tryParseMedia(stripped);
@@ -1332,12 +1345,9 @@ Pane.prototype.appendToolOutput = function (callId, name, output, isError) {
   var out = renderToolOutput(stripped, isError);
 
   // Mark the parent approval block as errored
-  if (isError) {
-    var parentBlock = target.closest(".ts-approval");
-    if (parentBlock && !parentBlock.classList.contains("denied")) {
-      parentBlock.classList.add("error");
-      appendToolErrorBadge(parentBlock);
-    }
+  if (isError && parentBlock && !parentBlock.classList.contains("denied")) {
+    parentBlock.classList.add("error");
+    appendToolErrorBadge(parentBlock);
   }
 
   if (out.textContent.split("\n").length > 10) {