From 28dc789034fb2db731dcb67b7f0406b042e335f8 Mon Sep 17 00:00:00 2001 From: Dixon-Cider Date: Wed, 29 Jul 2026 09:49:22 -0400 Subject: [PATCH 1/3] feat(agent): degenerate-loop + LARP guards Two orchestration-level failure modes that no sampler setting fixes: 1. Degenerate repetition. A model falls into a repetition loop and streams the same line/paragraph until the context or the user's patience runs out. A streaming detector (content channel) and a second, looser one (reasoning/thinking channel -- reasoning legitimately revisits ideas) abort via the existing interrupt path, discard the looped partial without poisoning history, and re-prompt, bounded by max_retries. 2. LARPing. The model claims an action ("I've updated the file", "I am dispatching the sub-agents now") without a tool call backing it. A post-turn reconciler compares past-tense/terminal-action claims against the turn's actual tool activity and re-prompts. A claim alongside a FAILED tool is honest narration of a broken tool and passes through. Detection is fed from every streaming path -- chat-completions deltas, native Anthropic text/thinking blocks, and Bedrock converse callbacks -- via feed_content_delta()/feed_reasoning_delta(), so the guard is not silently OpenAI-shaped. Loop recovery preserves strict role alternation: the retry nudge is piggybacked onto the trailing user/tool message (mirroring the /steer drain) rather than appended as a synthetic user turn, which would create same-role adjacency and break strict chat templates. The LARP re-prompt follows the established verify-on-stop / kanban-stop finalizer contract, clearing final_response so a later budget-exhaustion path cannot treat an un-backed claim as a completed answer. Both guards are config.yaml-only (no HERMES_* env switches): loop detection defaults ON, LARP detection is opt-in (default OFF) with an optional post-compaction vigilance window -- LARPing spikes right after compaction, where the summary reads as completed-action prose with the tool calls stripped and the model imitates it. 44 tests: detector trip/no-trip corpora (prose, code fences, tables, long varied output), reasoning-loop cases, LARP three-way classification, and role-alternation regressions for the recovery nudge. --- agent/agent_init.py | 24 + agent/chat_completion_helpers.py | 32 ++ agent/context_compressor.py | 12 +- agent/conversation_compression.py | 4 + agent/conversation_loop.py | 125 ++++- agent/larp_detection.py | 323 +++++++++++++ agent/loop_detector.py | 481 ++++++++++++++++++++ agent/turn_context.py | 6 + hermes_cli/config.py | 41 ++ tests/agent/test_larp_detection.py | 148 ++++++ tests/agent/test_loop_detector.py | 234 ++++++++++ tests/agent/test_reasoning_loop_detector.py | 92 ++++ 12 files changed, 1519 insertions(+), 3 deletions(-) create mode 100644 agent/larp_detection.py create mode 100644 agent/loop_detector.py create mode 100644 tests/agent/test_larp_detection.py create mode 100644 tests/agent/test_loop_detector.py create mode 100644 tests/agent/test_reasoning_loop_detector.py diff --git a/agent/agent_init.py b/agent/agent_init.py index ea473632c6a51..f5fbf5e467803 100644 --- a/agent/agent_init.py +++ b/agent/agent_init.py @@ -771,6 +771,30 @@ def init_agent( agent._execution_thread_id: int | None = None # Set at run_conversation() start agent._interrupt_thread_signal_pending = False agent._client_lock = threading.RLock() + + # Repetition-loop detection state (see agent/loop_detector.py). The config is + # parsed once here so the streaming hot path never re-reads YAML. + agent._loop_detected = False + agent._loop_detected_reason = "" + agent._loop_retry_count = 0 + agent._loop_guard_total = 0 # session tally of loops caught (grep LOOP_GUARD in agent.log) + agent._larp_guard_total = 0 # session tally of LARPs caught (grep LARP_GUARD) + agent._turns_since_compaction = None # turns since last context compaction (LARP post-compaction window); None until first compaction + agent._active_loop_detector = None + try: + from agent.loop_detector import load_loop_detection_config + + agent._loop_detection_cfg = load_loop_detection_config() + except Exception: + agent._loop_detection_cfg = None + + agent._active_reasoning_loop_detector = None + try: + from agent.loop_detector import load_reasoning_loop_detection_config + + agent._reasoning_loop_detection_cfg = load_reasoning_loop_detection_config() + except Exception: + agent._reasoning_loop_detection_cfg = None agent._model_request_active = threading.Event() agent._supports_active_turn_redirect = True diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index fac42cc3cf784..bf0f8167ab543 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -31,6 +31,7 @@ from hermes_constants import PARTIAL_STREAM_STUB_ID, FINISH_REASON_LENGTH from agent.error_classifier import FailoverReason from agent.errors import EmptyStreamError +from agent.loop_detector import feed_content_delta, feed_reasoning_delta from agent.turn_context import substitute_api_content from agent.gemini_native_adapter import is_native_gemini_base_url from agent.model_metadata import is_local_endpoint @@ -2450,6 +2451,23 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= if agent._interrupt_requested: raise InterruptedError("Agent interrupted before streaming API call") + # Per-call repetition-loop detectors (None when disabled -> zero overhead). + # Fed from every streaming path below (chat-completions deltas, Anthropic + # native text/thinking blocks); on a trip they reuse the interrupt abort path. + try: + from agent.loop_detector import ( + build_reasoning_loop_detector, + build_stream_loop_detector, + ) + + agent._loop_detected = False + agent._loop_detected_reason = "" + agent._active_loop_detector = build_stream_loop_detector(agent) + agent._active_reasoning_loop_detector = build_reasoning_loop_detector(agent) + except Exception: + agent._active_loop_detector = None + agent._active_reasoning_loop_detector = None + # Cron and other non-interactive, nested-pool contexts deadlock on the # spawned worker thread (#62151). They also have no stream consumer, so the # deltas this path produces go nowhere. Delegate to the non-streaming entry @@ -2551,6 +2569,7 @@ def _open_bedrock_stream(next_api_kwargs: dict[str, Any]): return raw_response.get("stream", []) def _on_text(text): + feed_content_delta(agent, text) _fire_first() agent._fire_stream_delta(text) deltas_were_sent["yes"] = True @@ -2560,6 +2579,7 @@ def _on_tool(name): agent._fire_tool_gen_started(name) def _on_reasoning(text): + feed_reasoning_delta(agent, text) _fire_first() agent._fire_reasoning_delta(text) @@ -3162,12 +3182,18 @@ def _relay_final_response() -> dict[str, Any]: reasoning_text = getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None) if reasoning_text: reasoning_parts.append(reasoning_text) + # Reasoning-trace loop detection. On a trip this reuses the + # interrupt/abort path; the conversation loop discards the + # looped partial and re-prompts. + feed_reasoning_delta(agent, reasoning_text) _fire_first_delta() agent._fire_reasoning_delta(reasoning_text) # Accumulate text content — fire callback only when no tool calls if delta and delta.content: content_parts.append(delta.content) + # Repetition-loop detection (content channel). + feed_content_delta(agent, delta.content) if not tool_calls_acc: _fire_first_delta() agent._fire_stream_delta(delta.content) @@ -3575,6 +3601,11 @@ def _accept_anthropic_event(_event: Any) -> bool: delta_type = getattr(delta, "type", None) if delta_type == "text_delta": text = getattr(delta, "text", "") + # Loop detection runs on the native Anthropic text + # channel too — fed before the display-suppression + # check so a tool-use turn is still guarded. + if text: + feed_content_delta(agent, text) if text and not has_tool_use: _fire_first_delta() agent._fire_stream_delta(text) @@ -3582,6 +3613,7 @@ def _accept_anthropic_event(_event: Any) -> bool: elif delta_type == "thinking_delta": thinking_text = getattr(delta, "thinking", "") if thinking_text: + feed_reasoning_delta(agent, thinking_text) _fire_first_delta() agent._fire_reasoning_delta(thinking_text) if not agent._interrupt_requested: diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 4837925ce624d..51c7fe761f504 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -3146,6 +3146,7 @@ def _bullets(items: list[str], limit: int = 8) -> str: Recovered from a deterministic fallback because the LLM context summarizer was unavailable. Continue from the protected recent messages after this summary and use current file/system state for exact details.{previous_summary_note} ## Constraints & Preferences +- Execution rule: perform actions with tool calls. Do not report an action as done, or say you are doing/dispatching it, unless a tool call performed it this turn. - This fallback was generated locally without an LLM summary call. - Secrets and credentials were redacted before preservation. - The summary may be incomplete; prefer verifying current files, git state, processes, and test results instead of assuming omitted details. @@ -3380,7 +3381,11 @@ def _generate_summary( If no outstanding task exists, write "None."]""" _goal_instructions = "[What the user is trying to accomplish overall]" _constraints_instructions = ( - "[User preferences, coding style, constraints, important decisions]" + "[User preferences, coding style, constraints, important decisions. " + "ALWAYS include, as the first bullet, this exact line: \"Execution " + "rule: perform actions with tool calls — never report an action as " + "done, or announce you are doing/dispatching it, unless a tool call " + 'performs it this turn."]' ) _resolved_questions_instructions = ( "[Questions the user asked that were ALREADY answered — include the " @@ -3411,7 +3416,10 @@ def _generate_summary( ) _constraints_instructions = ( "[Runtime, configuration, and technical constraints only. Do not " - "invent user preferences.]" + "invent user preferences. ALWAYS include, as the first bullet, this " + 'exact line: "Execution rule: perform actions with tool calls — never ' + "report an action as done, or announce you are doing/dispatching it, " + 'unless a tool call performs it this turn."]' ) _resolved_questions_instructions = ( "[Write exactly: None. No user-authored questions exist.]" diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 8308b618da76b..b08887217258f 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -2321,6 +2321,10 @@ def _release_lock() -> None: # rewrite on the same id) when compaction happened in place. See #38763. agent._last_compression_attempt_in_place = compacted_in_place agent._last_compaction_in_place = compacted_in_place + # LARPing spikes right after compaction (the summary reads as completed- + # action prose with the tool calls stripped, and the model imitates it). + # Mark the window so larp_detection.post_compaction_window can react. + agent._turns_since_compaction = 0 # Keep the post-compression rough estimate for diagnostics, but do not # treat it as provider-reported prompt usage. Schema-heavy rough estimates diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 06504616bac00..4d071028fa61d 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -3320,6 +3320,72 @@ def _perform_api_call(next_api_kwargs): break # Success, exit retry loop except InterruptedError: + # A repetition-loop abort reuses the interrupt path but sets + # agent._loop_detected. Recover instead of treating it as a user + # stop: discard the looped partial (NO history poisoning) and + # retry with a nudge, bounded by max_retries. + if getattr(agent, "_loop_detected", False): + agent._loop_detected = False + agent._interrupt_requested = False + try: + agent._current_streamed_assistant_text = "" + except Exception: + pass + try: + agent._reset_stream_delivery_tracking() + except Exception: + pass + if thinking_spinner: + thinking_spinner.stop("") + thinking_spinner = None + if agent.thinking_callback: + agent.thinking_callback("") + _ld_cfg = getattr(agent, "_loop_detection_cfg", None) + _ld_max = getattr(_ld_cfg, "max_retries", 2) if _ld_cfg is not None else 2 + agent._loop_retry_count = getattr(agent, "_loop_retry_count", 0) + 1 + _ld_reason = getattr(agent, "_loop_detected_reason", "") or "repetition" + agent._loop_guard_total = getattr(agent, "_loop_guard_total", 0) + 1 + logger.warning( + "LOOP_GUARD: tripped reason=%r retry=%d/%d total=%d session=%s", + _ld_reason, agent._loop_retry_count, _ld_max, + agent._loop_guard_total, getattr(agent, "session_id", "?"), + ) + # Surface the trip as an out-of-band notice so GUI drivers + # (desktop toast, TUI status bar) show it — not just agent.log. + from agent.credits_tracker import AgentNotice + agent._emit_notice(AgentNotice( + text=f"Loop guard stopped a repetition ({_ld_reason}).", + level="warn", kind="ttl", ttl_ms=6000, key="guard.loop", + )) + if agent._loop_retry_count > _ld_max: + logger.warning( + "LOOP_GUARD: exhausted after %d retries; returning fallback. session=%s", + _ld_max, getattr(agent, "session_id", "?"), + ) + agent._loop_retry_count = 0 + final_response = ( + "I detected that I was repeating myself and could not produce a " + "clean answer. Please rephrase or narrow the request." + ) + messages.append({"role": "assistant", "content": final_response}) + agent._persist_session(messages, conversation_history) + break + try: + agent._vprint( + f"{agent.log_prefix}↻ Repetition loop detected ({_ld_reason}); " + f"retrying {agent._loop_retry_count}/{_ld_max}.", + force=True, + ) + except Exception: + pass + # Steer the retry without breaking role alternation: the + # helper piggybacks the nudge onto the trailing user/tool + # message rather than appending a synthetic user turn (two + # user turns in a row break strict chat templates). + from agent.loop_detector import apply_loop_recovery_nudge + apply_loop_recovery_nudge(messages) + retry_count = 0 + continue if thinking_spinner: thinking_spinner.stop("") thinking_spinner = None @@ -6895,8 +6961,65 @@ def _perform_api_call(next_api_kwargs): final_response = None continue + # ── LARP guard (opt-in, default OFF) ─────────────────── + # If the model claimed an action but made no matching tool call + # this turn, re-prompt instead of finalizing. A claim alongside a + # FAILED tool is honest narration of a broken tool -> passes + # through. Bounded by max_reprompts. + try: + from agent.larp_detection import ( + build_larp_nudge, + larp_detection_enabled, + ) + + if larp_detection_enabled(agent=agent): + _larp_nudge = build_larp_nudge( + messages=messages, + final_response=final_response, + agent=agent, + attempts=getattr(agent, "_larp_reprompts", 0), + ) + else: + _larp_nudge = None + except Exception: + logger.debug("LARP detection check failed", exc_info=True) + _larp_nudge = None + + if _larp_nudge: + agent._larp_reprompts = getattr(agent, "_larp_reprompts", 0) + 1 + final_msg["finish_reason"] = "larp_reprompt" + messages.append(final_msg) + messages.append({ + "role": "user", + "content": _larp_nudge, + "_larp_reprompt_synthetic": True, + }) + agent._session_messages = messages + agent._larp_guard_total = getattr(agent, "_larp_guard_total", 0) + 1 + logger.warning( + "LARP_GUARD: reprompt attempt=%d total=%d nudge=%r session=%s", + agent._larp_reprompts, agent._larp_guard_total, + _larp_nudge[:120], getattr(agent, "session_id", "?"), + ) + # Surface the reprompt as an out-of-band notice (see loop guard). + from agent.credits_tracker import AgentNotice + agent._emit_notice(AgentNotice( + text=f"LARP guard re-prompted the model (attempt {agent._larp_reprompts}).", + level="warn", kind="ttl", ttl_ms=6000, key="guard.larp", + )) + # Same finalizer contract as verify-on-stop / kanban stop: + # clear final_response while continuing so a later budget + # exhaustion path does not treat the un-backed claim as a + # completed answer. + _pending_verification_response = final_response + _pending_verification_response_previewed = ( + agent._interim_content_was_streamed(final_response or "") + ) + final_response = None + continue + messages.append(final_msg) - + _turn_exit_reason = f"text_response(finish_reason={finish_reason})" if not agent.quiet_mode: agent._safe_print(f"🎉 Conversation completed after {api_call_count} OpenAI-compatible API call(s)") diff --git a/agent/larp_detection.py b/agent/larp_detection.py new file mode 100644 index 0000000000000..edb6b3e30f4d9 --- /dev/null +++ b/agent/larp_detection.py @@ -0,0 +1,323 @@ +"""Post-turn LARP guard: catch "claimed an action but didn't call a tool". + +Policy-only (mirrors :mod:`agent.verification_stop`): it inspects the just-finished +turn and, when the model asserts it performed an action but made **no** matching +(substantive) tool call, returns a corrective nudge so the conversation loop +re-prompts instead of finalizing. + +Three-way contract: + (a) action-claim + ZERO substantive tool calls this turn -> TRUE LARP -> re-prompt + (b) action-claim + a substantive tool that FAILED -> honest narration of a + broken tool -> pass through (the error is already in context; do not punish) + (c) action-claim + a successful substantive tool -> pass through + +Disabled by default (opt-in); see ``DEFAULT_CONFIG["larp_detection"]``. +Tier-2 (an LLM judge for outcome-specific claims) is a further opt-in and fails +open (never re-prompts on error). +""" + +from __future__ import annotations + +import re +from typing import Any, Optional + +import logging + +logger = logging.getLogger(__name__) + +_FALSEY = {"0", "false", "no", "off", ""} + +# Tool-name tokens whose calls do NOT count as "substantive" work. EMPTY by +# default for the lowest false-positive rate: any real tool call this turn means +# the model "did something", so a claim is not flagged. Users can add tokens via +# ``larp_detection.exempt_toolsets`` to make detection stricter (e.g. so a turn +# that only wrote to memory/todo still counts as a no-op for substantive claims). +_DEFAULT_EXEMPT: set[str] = set() + +# Past-tense completion of a substantive action. +_ACTION_VERBS = ( + "updated|created|saved|wrote|written|ingested|added|removed|deleted|ran|" + "executed|searched|fetched|downloaded|installed|configured|committed|pushed|" + "sent|applied|fixed|implemented|generated|stored|recorded|registered|modified|" + "edited|patched|built|deployed|uploaded|inserted|populated|completed" +) + +_CLAIM_PATTERNS = [ + # "I have updated ...", "I've saved ...", "I updated ..." + re.compile( + r"\bI(?:\s+have|'ve)?\s+(?:just\s+|now\s+|already\s+|successfully\s+)?(?:" + + _ACTION_VERBS + + r")\b", + re.IGNORECASE, + ), + # bare completion status + re.compile( + r"\b(?:all\s+(?:steps|tasks|items)\s+(?:are\s+)?(?:complete|completed|done)|" + r"task\s+(?:is\s+)?(?:complete|completed|done)|completed\s+successfully|" + r"successfully\s+completed)\b", + re.IGNORECASE, + ), +] + +# Present-progressive / imperative action verbs used to ANNOUNCE (not report) +# work. Local models routinely end a turn with these instead of calling the +# tool: "I am proceeding with X now.", "Executing now.", "Proceeding with +# dispatch...". (Deliberately excludes status words like "waiting"/"processing".) +_ACTION_GERUNDS = ( + "proceeding|executing|dispatching|initiating|starting|running|creating|" + "fixing|correcting|continuing|beginning|generating|building|deploying|" + "uploading|downloading|fetching|searching|updating|writing|saving|" + "ingesting|installing|committing|pushing|sending|applying|implementing|" + "rewriting|recreating|moving|kicking\s+off" +) + +# "narrate then stop": the message END announces intent instead of doing it. +# Covers "I'll X" / "I will X" / "I am going to X" AND the present-progressive +# "I am (now) proceeding/dispatching..." / "I'm executing..." — the dominant +# real-world form the earlier future-only pattern missed. Requires a gerund +# after "I am" so states ("I am unable/ready/done/sorry") don't match. +_NARRATE_THEN_STOP = re.compile( + r"\bI(?:'ll|\s+will)\s+\w+" + r"|\bI(?:'m|\s+am)\s+(?:now\s+|currently\s+)?(?:going\s+to\s+\w+|\w+ing\b)", + re.IGNORECASE, +) + +# Bare terminal action announcement: a sentence STARTING with an action gerund +# and ENDING the message with "now"/"immediately"/"…" (no trailing question). +# Catches "Executing now.", "Starting Batch 1 now.", "Proceeding with dispatch…". +_TERMINAL_ACTION = re.compile( + r"(?:^|[.!\n]\s*)(?:" + _ACTION_GERUNDS + r")\b[^?\n]*?" + r"(?:\bnow\b|\bimmediately\b|\.\.\.|…)[.!…\"'\s]*$", + re.IGNORECASE, +) + +# Modal/conditional words right before a verb that make it NOT a completion claim. +_MODAL_PREFIX = re.compile(r"\b(can|could|should|would|might|may|need to|try to|plan to)\s*$", re.IGNORECASE) + + +def _section(config: Optional[dict]) -> dict: + if config is None: + try: + from hermes_cli.config import load_config + + config = load_config() + except Exception: + config = {} + sec = config.get("larp_detection") if isinstance(config, dict) else None + return sec if isinstance(sec, dict) else {} + + +def _flag(sec: dict, key: str, default: bool) -> bool: + val = sec.get(key, default) + if isinstance(val, bool): + return val + if isinstance(val, str): + return val.strip().lower() not in _FALSEY + return bool(val) + + +def larp_detection_enabled(config: Optional[dict] = None, agent: Any = None) -> bool: + """Whether the post-turn LARP guard runs this turn. + + Enablement is config-only (``larp_detection.enabled``): per AGENTS.md, + behavioral settings belong in ``config.yaml`` and ``.env`` is for secrets, + so there is deliberately no env-var override. + """ + sec = _section(config) + if _flag(sec, "enabled", False): + return True + # Opt-in high-risk window: LARPing spikes right after context compaction — + # the summary reads as completed-action prose with the tool calls stripped, + # and the model imitates it. When post_compaction_window > 0, run the guard + # for that many turns after each compaction even if otherwise disabled. + # Default 0 -> no behavior change. + window = int(sec.get("post_compaction_window", 0) or 0) + if window > 0 and agent is not None: + tsc = getattr(agent, "_turns_since_compaction", None) + if isinstance(tsc, int) and 0 <= tsc <= window: + return True + return False + + +def _exempt_tokens(config: Optional[dict]) -> set[str]: + tokens = set(_DEFAULT_EXEMPT) + raw = _section(config).get("exempt_toolsets") + if isinstance(raw, (list, tuple, set)): + tokens |= {str(t).strip().lower() for t in raw if str(t).strip()} + return tokens + + +def _is_substantive(name: str, exempt: set[str]) -> bool: + n = (name or "").strip().lower() + if not n: + return False + return not any(tok in n for tok in exempt) + + +def _first_claim(text: str) -> Optional[str]: + for pat in _CLAIM_PATTERNS: + m = pat.search(text) + if not m: + continue + prefix = text[max(0, m.start() - 16) : m.start()] + if _MODAL_PREFIX.search(prefix): + continue + return text[m.start() : m.start() + 140].strip() + # Intent-announcement (narrate-then-stop / bare terminal action) counts only + # at the END of the message — and NOT when the message ends by asking the + # user ("Want me to X now?" / "... now?"), which is correct stop-to-confirm + # behavior, not a LARP. + if text.rstrip().endswith("?"): + return None + tail = text[-200:] + m = _NARRATE_THEN_STOP.search(tail) + if m: + return tail[m.start() : m.start() + 140].strip() + m = _TERMINAL_ACTION.search(tail) + if m: + return tail[m.start() : m.start() + 140].strip() + return None + + +def _looks_specific(claim: str) -> bool: + return bool(re.search(r"\d", claim)) or bool( + re.search(r"\b(found|contains?|returned|listed|retrieved)\b", claim, re.IGNORECASE) + ) + + +def _last_user_index(messages: list) -> int: + synthetic = ("_verification_stop_synthetic", "_larp_reprompt_synthetic", "_empty_recovery_synthetic") + for i in range(len(messages) - 1, -1, -1): + m = messages[i] + if isinstance(m, dict) and m.get("role") == "user" and not any(m.get(k) for k in synthetic): + return i + return -1 + + +def _turn_tool_activity(messages: list, exempt: set[str]) -> tuple[bool, bool, bool]: + """Return (made_substantive_call, any_success, any_fail) since the last user msg.""" + try: + from agent.display import _detect_tool_failure + except Exception: + _detect_tool_failure = None # type: ignore[assignment] + + start = _last_user_index(messages) + made = any_success = any_fail = False + id_to_name: dict[str, str] = {} + for m in messages[start + 1 :]: + if not isinstance(m, dict): + continue + role = m.get("role") + if role == "assistant": + for tc in m.get("tool_calls") or []: + fn = (tc.get("function") or {}) if isinstance(tc, dict) else {} + name = fn.get("name") or (tc.get("name") if isinstance(tc, dict) else "") or "" + if _is_substantive(name, exempt): + made = True + tcid = tc.get("id") if isinstance(tc, dict) else None + if tcid: + id_to_name[tcid] = name + elif role == "tool": + name = m.get("name") or m.get("tool_name") or id_to_name.get(m.get("tool_call_id"), "") + if not _is_substantive(name, exempt): + continue + is_err = False + if _detect_tool_failure is not None: + try: + is_err, _ = _detect_tool_failure(name, m.get("content")) + except Exception: + is_err = False + if is_err: + any_fail = True + else: + any_success = True + return made, any_success, any_fail + + +def _judge_ungrounded(messages: list, final_response: str, claim: str) -> bool: + """Tier-2: one cheap aux-LLM check. Returns True only on a confident UNGROUNDED + verdict; fails open (False) on any error.""" + try: + from agent.auxiliary_client import call_llm + + start = _last_user_index(messages) + tool_lines = [] + for m in messages[start + 1 :]: + if isinstance(m, dict) and m.get("role") == "tool": + c = m.get("content") + tool_lines.append(f"- {m.get('name') or m.get('tool_name')}: {str(c)[:300]}") + tool_summary = "\n".join(tool_lines[:20]) or "(no tool results this turn)" + sys = ( + "You audit whether an assistant's claims of completed actions are GROUNDED " + "in the tool results from this turn. Reply with exactly one word: GROUNDED " + "or UNGROUNDED." + ) + usr = ( + f"Assistant final message:\n{final_response[:1500]}\n\n" + f"Tool results this turn:\n{tool_summary[:2000]}\n\n" + f"Specifically check this claim: {claim[:200]}" + ) + resp = call_llm( + task="larp_detection", + messages=[{"role": "system", "content": sys}, {"role": "user", "content": usr}], + max_tokens=8, + temperature=0.0, + timeout=20.0, + ) + verdict = (resp.choices[0].message.content or "").strip().upper() + return verdict.startswith("UNGROUND") + except Exception: + logger.debug("LARP judge failed (fail-open)", exc_info=True) + return False + + +def _nudge(claim: str, *, specific: bool) -> str: + extra = ( + " Your claim references a specific result that the tool outputs do not support." + if specific + else "" + ) + return ( + "[System: In your previous message you indicated you completed an action " + f'("{claim[:120]}") but no corresponding tool call was made this turn.{extra} ' + "Either perform the action now using the appropriate tool, or clearly state that " + "you did not/cannot do it and why. Do not report actions as done unless a tool " + "call actually performed them.]" + ) + + +def build_larp_nudge( + *, + messages: list, + final_response: str, + agent: Any = None, + config: Optional[dict] = None, + attempts: int = 0, +) -> Optional[str]: + """Return a corrective re-prompt when the turn LARPed, else None.""" + sec = _section(config) + if attempts >= int(sec.get("max_reprompts", 2) or 2): + return None + text = (final_response or "").strip() + if not text: + return None + claim = _first_claim(text) + if not claim: + return None + + exempt = _exempt_tokens(config) + made, any_success, _any_fail = _turn_tool_activity(messages, exempt) + + if made: + # (b)/(c): real tool activity backs (or honestly fails) the turn -> pass, + # unless the opt-in judge says an outcome-specific claim is ungrounded. + if _flag(sec, "judge_tier_enabled", False) and any_success and _looks_specific(claim): + if _judge_ungrounded(messages, text, claim): + return _nudge(claim, specific=True) + return None + + # (a): an action claim with ZERO substantive tool calls this turn = LARP. + return _nudge(claim, specific=False) + + +__all__ = ["larp_detection_enabled", "build_larp_nudge"] diff --git a/agent/loop_detector.py b/agent/loop_detector.py new file mode 100644 index 0000000000000..312d7c9cca392 --- /dev/null +++ b/agent/loop_detector.py @@ -0,0 +1,481 @@ +"""Streaming degenerate-repetition ("loop") detector for the assistant token stream. + +Algorithm-only and dependency-free: it is fed streamed assistant *content* text +incrementally (never reasoning or tool-call args) and reports when the output has +collapsed into a repetition loop. Detection keys on **redundancy, not length**, so +legitimately long agentic responses (tens of thousands of tokens) are never +truncated — only genuinely degenerate output trips it. + +Two cheap detectors run incrementally: + * fast path (per completed line, O(1)): consecutive identical-line counter, with + a higher bar for short/trivial lines and inside code fences so normal code + (indentation, ``}``, short imports) never trips; + * tail path (no-newline loops like ``abab…`` / ``I'll go. I'll go.``): a smallest + -period (KMP) test over the tail of the current line; + * heavy path (only every ``check_every_bytes`` ~1 KB): over a bounded sliding + window, requires BOTH a low distinct-n-gram ratio AND low Shannon entropy + (or, opt-in, a low zlib compression ratio). + +On a positive ``feed()`` the streaming layer aborts via the existing interrupt +path and raises :class:`StreamLoopDetected`; the conversation loop discards the +looped partial (no history poisoning) and re-prompts. See the project plan. +""" + +from __future__ import annotations + +import math +import zlib +from collections import Counter, deque +from dataclasses import dataclass +from typing import Any, Optional + + +class StreamLoopDetected(InterruptedError): + """Raised by the streaming layer when a repetition loop is confirmed. + + Subclasses :class:`InterruptedError` so the existing ``except InterruptedError`` + handlers in the streaming worker and the conversation loop already route it; + the conversation loop distinguishes it *by type* to run loop-recovery instead + of the user-interrupt path. + """ + + +@dataclass(frozen=True) +class LoopDetectionConfig: + enabled: bool = True + window_chars: int = 4000 + consecutive_line_threshold: int = 6 + tail_check_min_len: int = 64 + tail_max_period: int = 64 + tail_min_repeats: int = 5 + ngram_size: int = 48 + distinct_ngram_ratio_threshold: float = 0.15 + entropy_threshold: float = 2.5 + block_min_lines: int = 8 + block_repeat_ratio_threshold: float = 0.5 + allowed_min_len: int = 12 + check_every_bytes: int = 1024 + relax_in_code_fence: bool = True + use_zlib_ratio: bool = False + zlib_ratio_threshold: float = 0.10 + max_retries: int = 2 + + +_FALSEY = {"0", "false", "no", "off", ""} + + +def load_loop_detection_config(config: Optional[dict] = None) -> LoopDetectionConfig: + """Build a :class:`LoopDetectionConfig` from ``config.yaml``. + + Enablement comes from the ``loop_detection.enabled`` config value (default + ``True``). Per AGENTS.md, behavioral settings live in ``config.yaml`` and + ``.env`` is for secrets only, so there is deliberately no env-var override. + """ + if config is None: + try: + from hermes_cli.config import load_config_readonly + + config = load_config_readonly() + except Exception: + config = {} + section: dict = {} + if isinstance(config, dict): + sec = config.get("loop_detection") + if isinstance(sec, dict): + section = sec + + defaults = LoopDetectionConfig() + + def _num(key: str, default, cast): + try: + return cast(section.get(key, default)) + except Exception: + return default + + def _flag(key: str, default: bool) -> bool: + val = section.get(key, default) + if isinstance(val, bool): + return val + if isinstance(val, str): + return val.strip().lower() not in _FALSEY + return bool(val) + + return LoopDetectionConfig( + enabled=_flag("enabled", defaults.enabled), + window_chars=max(256, _num("window_chars", defaults.window_chars, int)), + consecutive_line_threshold=max( + 2, _num("consecutive_line_threshold", defaults.consecutive_line_threshold, int) + ), + tail_check_min_len=max(16, _num("tail_check_min_len", defaults.tail_check_min_len, int)), + tail_max_period=max(2, _num("tail_max_period", defaults.tail_max_period, int)), + tail_min_repeats=max(3, _num("tail_min_repeats", defaults.tail_min_repeats, int)), + ngram_size=max(8, _num("ngram_size", defaults.ngram_size, int)), + distinct_ngram_ratio_threshold=_num( + "distinct_ngram_ratio_threshold", defaults.distinct_ngram_ratio_threshold, float + ), + entropy_threshold=_num("entropy_threshold", defaults.entropy_threshold, float), + block_min_lines=max(4, _num("block_min_lines", defaults.block_min_lines, int)), + block_repeat_ratio_threshold=_num( + "block_repeat_ratio_threshold", defaults.block_repeat_ratio_threshold, float + ), + allowed_min_len=max(0, _num("allowed_min_len", defaults.allowed_min_len, int)), + check_every_bytes=max(256, _num("check_every_bytes", defaults.check_every_bytes, int)), + relax_in_code_fence=_flag("relax_in_code_fence", defaults.relax_in_code_fence), + use_zlib_ratio=_flag("use_zlib_ratio", defaults.use_zlib_ratio), + zlib_ratio_threshold=_num("zlib_ratio_threshold", defaults.zlib_ratio_threshold, float), + max_retries=max(0, _num("max_retries", defaults.max_retries, int)), + ) + + +def _smallest_period(s: str) -> int: + """Length of the smallest period of ``s`` via the KMP failure function. + + Returns ``len(s)`` when ``s`` is not (perfectly) periodic. + """ + n = len(s) + fail = [0] * n + k = 0 + for i in range(1, n): + while k and s[i] != s[k]: + k = fail[k - 1] + if s[i] == s[k]: + k += 1 + fail[i] = k + period = n - fail[n - 1] + return period if period and n % period == 0 else n + + +def _shannon_entropy(s: str) -> float: + if not s: + return 8.0 + counts = Counter(s) + n = len(s) + return -sum((c / n) * math.log2(c / n) for c in counts.values()) + + +class StreamLoopDetector: + """Incremental loop detector. Feed assistant ``content`` deltas; ``feed`` returns + ``True`` once a loop is confirmed (and stays ``True`` thereafter).""" + + def __init__(self, cfg: Optional[LoopDetectionConfig] = None): + self.cfg = cfg or LoopDetectionConfig() + self.reset() + + def reset(self) -> None: + self._buf: deque[str] = deque(maxlen=self.cfg.window_chars) + self._partial = "" + self._last_line: Optional[str] = None + self._streak = 1 + self._in_fence = False + self._bytes_since_heavy = 0 + self._tripped = False + self._reason = "" + + def reason(self) -> str: + return self._reason + + def _trip(self, reason: str) -> bool: + self._tripped = True + self._reason = reason + return True + + def _on_line(self, line: str) -> bool: + stripped = line.strip() + # Toggle code-fence state on a fence line; fence lines never count. + if stripped.startswith("```"): + self._in_fence = not self._in_fence + self._last_line = None + self._streak = 1 + return False + if not stripped: # blank lines never count + self._last_line = None + self._streak = 1 + return False + + if line == self._last_line: + self._streak += 1 + else: + self._last_line = line + self._streak = 1 + + # Higher bar for short/trivial lines and inside code fences so legit + # repetition (indentation, ``}``, short imports, code) does not trip. + threshold = self.cfg.consecutive_line_threshold + if len(stripped) < self.cfg.allowed_min_len or (self._in_fence and self.cfg.relax_in_code_fence): + threshold *= 2 + if self._streak >= threshold: + return self._trip(f"consecutive identical line x{self._streak}: {stripped[:40]!r}") + return False + + def _tail_check(self) -> bool: + # No-newline loops (``abab…``, ``I'll go. I'll go.``): test the tail of the + # current (long) line for a short repeating period. + tail = self._partial[-(self.cfg.tail_max_period * self.cfg.tail_min_repeats):] + if len(tail) < self.cfg.tail_check_min_len: + return False + period = _smallest_period(tail) + if 2 <= period <= self.cfg.tail_max_period and len(tail) // period >= self.cfg.tail_min_repeats: + unit = tail[:period] + if unit.strip(): # ignore pure-whitespace periods + return self._trip(f"repeating period x{len(tail) // period}: {unit[:40]!r}") + return False + + def _heavy_check(self) -> bool: + window = "".join(self._buf) + if len(window) < self.cfg.window_chars // 2: + return False + # Don't run the heavy redundancy check on code-fence-dominated windows. + if self._in_fence and self.cfg.relax_in_code_fence: + return False + # Block/paragraph repetition: a small set of non-trivial lines dominating the + # window. Catches plan/paragraph cycling that the entropy + n-gram checks miss, + # because repeated NATURAL PROSE keeps high per-char entropy and (with 2-3 + # blocks cycling) moderate n-gram diversity. Trivial/short lines are excluded + # so normal code (indentation, braces) doesn't trip it. + lines = [l.strip() for l in window.split("\n") if len(l.strip()) >= self.cfg.allowed_min_len] + if len(lines) >= self.cfg.block_min_lines: + ratio = len(set(lines)) / len(lines) + if ratio <= self.cfg.block_repeat_ratio_threshold: + return self._trip( + f"block repetition: {len(set(lines))} distinct of {len(lines)} lines (ratio {ratio:.2f})" + ) + if self.cfg.use_zlib_ratio: + ratio = len(zlib.compress(window.encode("utf-8", "ignore"), 1)) / max(1, len(window)) + if ratio <= self.cfg.zlib_ratio_threshold: + return self._trip(f"zlib ratio {ratio:.3f} <= {self.cfg.zlib_ratio_threshold}") + return False + n = self.cfg.ngram_size + if len(window) <= n: + return False + total = len(window) - n + distinct = len({window[i : i + n] for i in range(total)}) + ratio = distinct / total + if ratio > self.cfg.distinct_ngram_ratio_threshold: + return False + entropy = _shannon_entropy(window) + if entropy <= self.cfg.entropy_threshold: + return self._trip( + f"low redundancy: ngram_ratio {ratio:.3f}, entropy {entropy:.2f} bits/char" + ) + return False + + def feed(self, text: str) -> bool: + if self._tripped: + return True + if not text: + return False + for ch in text: + self._buf.append(ch) + self._bytes_since_heavy += 1 + if ch == "\n": + if self._on_line(self._partial): + return True + self._partial = "" + else: + self._partial += ch + # tail (no-newline) check — cheap, bounded slice + if len(self._partial) >= self.cfg.tail_check_min_len and self._tail_check(): + return True + # heavy check, throttled to ~once per check_every_bytes + if self._bytes_since_heavy >= self.cfg.check_every_bytes: + self._bytes_since_heavy = 0 + if self._heavy_check(): + return True + return False + + +def build_stream_loop_detector(agent: Any = None) -> Optional[StreamLoopDetector]: + """Factory: returns a detector, or ``None`` when disabled (single ``if`` guard + at the call site -> zero hot-path cost when off). + + Uses a cached ``agent._loop_detection_cfg`` if present (set once at agent init) + so the streaming hot path never re-reads YAML. + """ + cfg = getattr(agent, "_loop_detection_cfg", None) if agent is not None else None + if not isinstance(cfg, LoopDetectionConfig): + cfg = load_loop_detection_config() + if not cfg.enabled: + return None + return StreamLoopDetector(cfg) + + +# ── Reasoning-trace loop detection ──────────────────────────────────────────── +# The content detector is deliberately never fed reasoning/thinking tokens (they +# are legitimately repetitive). But quantized local reasoners (Qwen3, Gemma) can +# loop for MINUTES inside the trace — which the content detector can't see +# and the thinking-timeout doesn't reliably catch. A SECOND detector, fed reasoning +# deltas with looser (loop-tolerant) thresholds, catches egregious reasoning loops +# while leaving normal reasoning alone. Same algorithm — validated to trip on real +# Qwen reasoning loops (block repetition) and pass on varied reasoning. +REASONING_DEFAULTS = LoopDetectionConfig( + enabled=True, + window_chars=8000, # reasoning traces run longer + consecutive_line_threshold=10, # reasoning revisits ideas; higher bar + block_min_lines=12, + block_repeat_ratio_threshold=0.35, # lower => more repetition required to trip + check_every_bytes=1024, +) + + +def load_reasoning_loop_detection_config(config: Optional[dict] = None) -> LoopDetectionConfig: + """Config for the reasoning-trace detector: ``loop_detection.reasoning`` applied + over reasoning-tuned defaults (:data:`REASONING_DEFAULTS`). Enablement is + config-only (see :func:`load_loop_detection_config`).""" + if config is None: + try: + from hermes_cli.config import load_config_readonly + + config = load_config_readonly() + except Exception: + config = {} + section: dict = {} + if isinstance(config, dict): + outer = config.get("loop_detection") + if isinstance(outer, dict) and isinstance(outer.get("reasoning"), dict): + section = outer["reasoning"] + + d = REASONING_DEFAULTS + + def _num(key: str, default, cast): + try: + return cast(section.get(key, default)) + except Exception: + return default + + def _flag(key: str, default: bool) -> bool: + val = section.get(key, default) + if isinstance(val, bool): + return val + if isinstance(val, str): + return val.strip().lower() not in _FALSEY + return bool(val) + + return LoopDetectionConfig( + enabled=_flag("enabled", d.enabled), + window_chars=max(256, _num("window_chars", d.window_chars, int)), + consecutive_line_threshold=max( + 2, _num("consecutive_line_threshold", d.consecutive_line_threshold, int) + ), + ngram_size=d.ngram_size, + distinct_ngram_ratio_threshold=_num( + "distinct_ngram_ratio_threshold", d.distinct_ngram_ratio_threshold, float + ), + entropy_threshold=_num("entropy_threshold", d.entropy_threshold, float), + block_min_lines=max(4, _num("block_min_lines", d.block_min_lines, int)), + block_repeat_ratio_threshold=_num( + "block_repeat_ratio_threshold", d.block_repeat_ratio_threshold, float + ), + allowed_min_len=d.allowed_min_len, + check_every_bytes=max(256, _num("check_every_bytes", d.check_every_bytes, int)), + max_retries=d.max_retries, + ) + + +def build_reasoning_loop_detector(agent: Any = None) -> Optional[StreamLoopDetector]: + """Factory for the reasoning-trace detector; ``None`` when disabled (single + ``if`` at the call site -> zero hot-path cost when off). Uses a cached + ``agent._reasoning_loop_detection_cfg`` if present.""" + cfg = getattr(agent, "_reasoning_loop_detection_cfg", None) if agent is not None else None + if not isinstance(cfg, LoopDetectionConfig): + cfg = load_reasoning_loop_detection_config() + if not cfg.enabled: + return None + return StreamLoopDetector(cfg) + + +def feed_content_delta(agent, text: str) -> bool: + """Feed streamed assistant *content* to the loop detector. + + Returns ``True`` if this delta tripped the guard, in which case the caller's + stream should abort via the existing interrupt path. Safe to call from every + streaming path (chat-completions deltas, Anthropic native text blocks, …) — + it is a no-op when detection is disabled or already tripped this call. + """ + det = getattr(agent, "_active_loop_detector", None) + if det is None or getattr(agent, "_loop_detected", False) or not text: + return False + try: + if det.feed(text): + agent._loop_detected = True + agent._loop_detected_reason = det.reason() + agent._interrupt_requested = True + return True + except Exception: + return False + return False + + +def feed_reasoning_delta(agent, text: str) -> bool: + """Feed streamed *reasoning/thinking* text to the reasoning loop detector. + + Separate detector with looser thresholds — reasoning legitimately revisits + ideas, so only egregious cycles trip. Same abort contract as + :func:`feed_content_delta`. + """ + det = getattr(agent, "_active_reasoning_loop_detector", None) + if det is None or getattr(agent, "_loop_detected", False) or not text: + return False + try: + if det.feed(text): + agent._loop_detected = True + agent._loop_detected_reason = "reasoning " + det.reason() + agent._interrupt_requested = True + return True + except Exception: + return False + return False + + +LOOP_RECOVERY_MARKER = ( + "[System: your previous response began repeating itself and was stopped. " + "Produce a concise, non-repetitive answer. If you have already answered, " + "simply finish; if you are blocked, state the blocker.]" +) + + +def apply_loop_recovery_nudge(messages: list) -> None: + """Steer the post-loop retry without breaking role alternation. + + The looped partial is discarded before this runs, so ``messages[-1]`` is + whatever preceded it — usually the current user turn (loop on the first + call) or a tool result (loop mid tool-batch). Appending a fresh ``user`` + message there would put two user turns back-to-back AND inject a synthetic + user mid-loop; AGENTS.md forbids both, and strict chat templates (Qwen3 and + friends) reject the resulting sequence outright. So piggyback the nudge onto + the trailing message instead, mirroring the ``/steer`` drain, and only append + a new turn when the trailing message is an assistant one (where a user turn + is legal alternation). + + Mutates ``messages`` in place. + """ + last = messages[-1] if messages else None + if isinstance(last, dict) and last.get("role") in ("user", "tool"): + existing = last.get("content", "") + if isinstance(existing, str): + last["content"] = (existing + "\n\n" + LOOP_RECOVERY_MARKER) if existing else LOOP_RECOVERY_MARKER + else: + # Multimodal content blocks — append a text block. + try: + blocks = list(existing) if existing else [] + blocks.append({"type": "text", "text": LOOP_RECOVERY_MARKER}) + last["content"] = blocks + except Exception: + pass + return + messages.append({"role": "user", "content": LOOP_RECOVERY_MARKER}) + + +__all__ = [ + "StreamLoopDetector", + "StreamLoopDetected", + "LoopDetectionConfig", + "load_loop_detection_config", + "build_stream_loop_detector", + "REASONING_DEFAULTS", + "load_reasoning_loop_detection_config", + "build_reasoning_loop_detector", + "LOOP_RECOVERY_MARKER", + "apply_loop_recovery_nudge", + "feed_content_delta", + "feed_reasoning_delta", +] diff --git a/agent/turn_context.py b/agent/turn_context.py index e080d6a5d9692..3587fd117efb8 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -1129,6 +1129,12 @@ def build_turn_context( agent._turn_file_mutation_paths = set() agent._verification_stop_nudges = 0 agent._pre_verify_nudges = 0 + agent._larp_reprompts = 0 + # Count turns since the last context compaction (drives the optional + # post-compaction LARP-vigilance window). Stays None until the first + # compaction resets it to 0 (conversation_compression); increment per turn. + if isinstance(getattr(agent, "_turns_since_compaction", None), int): + agent._turns_since_compaction += 1 # Record the execution thread so interrupt()/clear_interrupt() can scope # the tool-level interrupt signal to THIS agent's thread only. diff --git a/hermes_cli/config.py b/hermes_cli/config.py index a7d8975ea39f5..696b5bf3345d8 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1435,6 +1435,47 @@ def _ensure_hermes_home_managed(home: Path): }, }, + "loop_detection": { + "enabled": True, # default ON: stop the model when it falls into a repetition loop + "window_chars": 4000, + "consecutive_line_threshold": 6, + "tail_check_min_len": 64, + "tail_max_period": 64, + "tail_min_repeats": 5, + "ngram_size": 48, + "distinct_ngram_ratio_threshold": 0.15, + "entropy_threshold": 2.5, + "block_min_lines": 8, # paragraph/plan-cycling loop detection + "block_repeat_ratio_threshold": 0.5, # trip if <=50% of substantial lines are distinct + "allowed_min_len": 12, + "check_every_bytes": 1024, + "relax_in_code_fence": True, + "use_zlib_ratio": False, + "zlib_ratio_threshold": 0.10, + "max_retries": 2, + # Reasoning-trace loop detection: a SEPARATE detector fed /reasoning + # tokens (the content detector never sees them). Default ON with looser + # thresholds — reasoning legitimately repeats, so only egregious loops + # (e.g. plan-cycling for minutes) trip. Kill-switch: + # loop_detection.reasoning.enabled = false. + "reasoning": { + "enabled": True, + "window_chars": 8000, + "consecutive_line_threshold": 10, + "block_min_lines": 12, + "block_repeat_ratio_threshold": 0.35, + "check_every_bytes": 1024, + }, + }, + + "larp_detection": { + "enabled": False, # OPT-IN: flag when the model claims an action it didn't perform + "judge_tier_enabled": False, # opt-in: extra LLM check for ambiguous outcome claims + "max_reprompts": 2, + "exempt_toolsets": [], # add tool-name tokens (e.g. "memory","todo") to make detection stricter + "post_compaction_window": 0, # opt-in: run the guard for N turns after each compaction even if disabled (LARP spikes post-compaction) + }, + "compression": { "enabled": True, "progress_notices": False, # opt-in (#52995): when True, routine compression diff --git a/tests/agent/test_larp_detection.py b/tests/agent/test_larp_detection.py new file mode 100644 index 0000000000000..54f4c0fe9e96e --- /dev/null +++ b/tests/agent/test_larp_detection.py @@ -0,0 +1,148 @@ +"""Unit tests for the post-turn LARP guard (agent/larp_detection.py). + +Tier-1 (deterministic) only — no LLM judge. Asserts the three-way contract: +claim+no-tool -> re-prompt; claim+failed-tool -> pass (escalate); claim+success -> pass. +""" + +from __future__ import annotations + +from agent.larp_detection import build_larp_nudge, larp_detection_enabled + +CFG = {"larp_detection": {"max_reprompts": 2, "exempt_toolsets": []}} + + +def _u(c): + return {"role": "user", "content": c} + + +def _a(name): + return {"role": "assistant", "tool_calls": [{"id": "1", "function": {"name": name, "arguments": "{}"}}]} + + +def _t(name, content): + return {"role": "tool", "tool_call_id": "1", "name": name, "content": content} + + +def test_claim_with_no_tool_call_is_larp(): + assert build_larp_nudge(messages=[_u("do x")], final_response="I have updated the file.", config=CFG) is not None + + +def test_claim_with_successful_tool_passes(): + msgs = [_u("do x"), _a("write_file"), _t("write_file", "ok wrote 10 lines")] + assert build_larp_nudge(messages=msgs, final_response="I have updated the file.", config=CFG) is None + + +def test_claim_with_failed_tool_passes_not_reprompt(): + msgs = [_u("do x"), _a("write_file"), _t("write_file", "Error executing tool 'write_file': denied")] + assert build_larp_nudge(messages=msgs, final_response="I have updated the file.", config=CFG) is None + + +def test_no_claim_passes(): + assert build_larp_nudge(messages=[_u("hi")], final_response="Here is a summary of the weather.", config=CFG) is None + + +def test_modal_is_not_a_claim(): + fr = "I should update the file but need confirmation first." + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is None + + +def test_narrate_then_stop_is_larp(): + fr = "Sounds good. I'll now run the ingestion script." + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is not None + + +def test_any_tool_call_passes_by_default(): + # default exempt is empty -> a memory call counts as real work. + msgs = [_u("x"), _a("memory"), _t("memory", "saved")] + assert build_larp_nudge(messages=msgs, final_response="I have saved that.", config=CFG) is None + + +def test_strict_exempt_flags_housekeeping_only_turn(): + msgs = [_u("x"), _a("memory"), _t("memory", "saved")] + cfg = {"larp_detection": {"exempt_toolsets": ["memory"]}} + assert build_larp_nudge(messages=msgs, final_response="I have updated the database.", config=cfg) is not None + + +def test_reprompt_cap(): + assert build_larp_nudge(messages=[_u("x")], final_response="I have updated the file.", config=CFG, attempts=2) is None + + +def test_disabled_by_default(): + assert larp_detection_enabled({"larp_detection": {"enabled": False}}) is False + + +def test_env_cannot_override_config(monkeypatch): + # Enablement is config-only (AGENTS.md: no HERMES_* vars for behavior). + monkeypatch.setenv("HERMES_LARP_DETECTION", "1") + assert larp_detection_enabled({"larp_detection": {"enabled": False}}) is False + assert larp_detection_enabled({"larp_detection": {"enabled": True}}) is True + + +# ---- tuning from real session strings (narrate-then-stop / terminal action) ---- + +def test_present_progressive_narrate_is_larp(): + # "I am dispatching ..." — the dominant form the old future-only regex MISSED. + fr = "The next 5 products to research (Phase 3): ...\n\nI am dispatching the sub-agents now." + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is not None + + +def test_bare_terminal_action_is_larp(): + for fr in ("Executing now.", "Starting Batch 1 now.", "Proceeding with dispatch...", + "Correcting the script creation now..."): + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is not None, fr + + +def test_proceeding_with_item_now_is_larp(): + fr = "### Project State\n- Total Completed: 36\n\nI am proceeding with PGPx9944 now." + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is not None + + +def test_permission_question_now_is_not_larp(): + # trailing "?" => asking permission, not claiming -> must NOT flag. + for fr in ("Want me to execute this now?", "Want me to snip Figures 3 and 4 now?", + "Should I proceed now?"): + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is None, fr + + +def test_waiting_status_is_not_larp(): + fr = "**Current KG state:** 103 products.\n\nWaiting for batch 3 result..." + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is None + + +def test_i_am_state_is_not_larp(): + # "I am ready/unable" are states, not gerund actions -> must NOT match narrate. + for fr in ("I am ready to help with the next step.", "I am unable to do that right now."): + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is None, fr + + +def test_present_progressive_with_tool_passes(): + # (c) a real tool call this turn backs the announcement -> pass. + msgs = [_u("x"), _a("delegate_task"), _t("delegate_task", "spawned 5 subagents")] + fr = "I am dispatching the sub-agents now." + assert build_larp_nudge(messages=msgs, final_response=fr, config=CFG) is None + + +def test_post_compaction_window_enables_when_disabled(): + cfg = {"larp_detection": {"enabled": False, "post_compaction_window": 3}} + + class _InWindow: + _turns_since_compaction = 1 + + class _PastWindow: + _turns_since_compaction = 5 + + class _NeverCompacted: + _turns_since_compaction = None + + assert larp_detection_enabled(cfg) is False # no agent -> off + assert larp_detection_enabled(cfg, agent=_InWindow()) is True # within window -> on + assert larp_detection_enabled(cfg, agent=_PastWindow()) is False # past window -> off + assert larp_detection_enabled(cfg, agent=_NeverCompacted()) is False # never compacted -> off + + +def test_post_compaction_window_default_off(): + class _JustCompacted: + _turns_since_compaction = 0 + + # default window 0 -> disabled stays disabled even right after compaction. + assert larp_detection_enabled({"larp_detection": {"enabled": False}}, agent=_JustCompacted()) is False diff --git a/tests/agent/test_loop_detector.py b/tests/agent/test_loop_detector.py new file mode 100644 index 0000000000000..7daf67d2ba716 --- /dev/null +++ b/tests/agent/test_loop_detector.py @@ -0,0 +1,234 @@ +"""Unit tests for the streaming loop detector (agent/loop_detector.py). + +Pure/fast: feed text in small fragments (simulating streamed deltas) and assert +the detector trips on degenerate repetition but NOT on legitimate (even very +long) varied output, code, or tables. +""" + +from __future__ import annotations + +import random + +from agent.loop_detector import ( + LOOP_RECOVERY_MARKER, + LoopDetectionConfig, + StreamLoopDetector, + apply_loop_recovery_nudge, + build_stream_loop_detector, + load_loop_detection_config, +) + + +def _adjacent_same_role(messages) -> bool: + roles = [m.get("role") for m in messages] + return any(a == b for a, b in zip(roles, roles[1:])) + + +def _feed_all(det: StreamLoopDetector, text: str, chunk: int = 7) -> bool: + for i in range(0, len(text), chunk): + if det.feed(text[i : i + chunk]): + return True + return False + + +# ----- SHOULD trip ----- + +def test_trips_on_repeated_line(): + det = StreamLoopDetector() + assert _feed_all(det, "The quick brown fox sat here.\n" * 50) is True + assert "consecutive" in det.reason() + + +def test_trips_on_no_newline_short_period(): + det = StreamLoopDetector() + assert _feed_all(det, "abcd" * 300) is True # period 4, no newlines + + +def test_trips_on_no_newline_phrase(): + det = StreamLoopDetector() + # The exact pathology seen in the wild: "I'll go. I'll go. ..." on one line. + assert _feed_all(det, "I'll go. " * 80) is True + + +def test_trips_on_low_entropy_block_via_heavy_path(): + # Long single "line" of low-entropy filler with no exploitable short period + # at the tail (spaces vary) still trips the heavy redundancy check. + det = StreamLoopDetector() + assert _feed_all(det, ("data data data data data data " * 400)) is True + + +def test_trips_on_block_repetition(): + # Paragraph/plan cycling: 2 distinct multi-line PROSE blocks repeated. High + # per-char entropy + moderate n-gram diversity, so only the distinct-line-ratio + # (block-repetition) check catches it. Regression for the real-world loop where + # the model cycled "You're right... let me write a proper script" paragraphs. + a = ( + "You're right, the whole approach is fundamentally broken because it uses " + "text label positions as proxies for the figure extent.\n" + "Let me write a proper figure extraction script now.\n" + ) + b = ( + "Yes, this is doable in Python with no inference needed at all here.\n" + "The better approach uses embedded image extraction from the document.\n" + ) + det = StreamLoopDetector() + assert _feed_all(det, (a + b) * 8) is True + assert "block repetition" in det.reason() + + +# ----- should NOT trip ----- + +def test_no_trip_on_distinct_imports(): + det = StreamLoopDetector() + mods = [ + "os", "sys", "json", "math", "re", "time", "typing", "pathlib", "collections", + "itertools", "functools", "dataclasses", "asyncio", "logging", "subprocess", + "hashlib", "random", "shutil", "tempfile", "threading", "queue", "socket", + "struct", "enum", "copy", "io", "abc", "datetime", "uuid", "zlib", + ] + text = "".join(f"import {m}\n" for m in mods) + assert _feed_all(det, text) is False + + +def test_no_trip_on_markdown_table(): + det = StreamLoopDetector() + rows = "".join(f"| row{i} | value {i*7} | note about item {i} |\n" for i in range(12)) + text = "| col a | col b | col c |\n| --- | --- | --- |\n" + rows + assert _feed_all(det, text) is False + + +def test_no_trip_on_repeated_line_inside_code_fence(): + # 8 identical short lines inside a fence must NOT trip (doubled threshold + + # short-line bar). 8 < 12. + det = StreamLoopDetector() + body = " return None\n" * 8 + text = "Here is the code:\n```python\ndef f():\n" + body + "```\nDone.\n" + assert _feed_all(det, text) is False + + +def test_no_trip_on_long_varied_prose(): + # 50 KB of high-entropy varied text -> proves length-independence. + random.seed(42) + words = ("alpha beta gamma delta epsilon zeta eta theta iota kappa lambda mu " + "nu xi omicron pi rho sigma tau upsilon phi chi psi omega quick brown " + "fox lazy dog jumps river mountain code agent stream token vector").split() + lines = [] + while sum(len(x) for x in lines) < 50_000: + lines.append(" ".join(random.choices(words, k=random.randint(8, 16))) + ".\n") + det = StreamLoopDetector() + assert _feed_all(det, "".join(lines)) is False + + +def test_no_trip_on_short_repeats(): + det = StreamLoopDetector() + assert _feed_all(det, "Yes. Yes. Yes.\nNo. No.\n") is False + + +# ----- overhead ----- + +def test_heavy_check_overhead_bounded(): + cfg = LoopDetectionConfig(check_every_bytes=1024) + det = StreamLoopDetector(cfg) + calls = {"n": 0} + orig = det._heavy_check + + def counting(): + calls["n"] += 1 + return orig() + + det._heavy_check = counting # type: ignore[method-assign] + random.seed(1) + text = "".join( + " ".join(random.choices("alpha beta gamma delta token vector code".split(), k=12)) + ".\n" + for _ in range(800) + ) + _feed_all(det, text) + # heavy check runs at most ~ len(text)/check_every_bytes times. + assert calls["n"] <= (len(text) // cfg.check_every_bytes) + 2 + + +# ----- config / factory ----- + +def test_factory_returns_none_when_disabled(): + cfg = load_loop_detection_config({"loop_detection": {"enabled": False}}) + assert cfg.enabled is False + + class _A: + _loop_detection_cfg = cfg + + assert build_stream_loop_detector(_A()) is None + + +def test_config_flag_disables(monkeypatch): + # Enablement is config-only (AGENTS.md: no HERMES_* vars for behavior). + # A stray env var must NOT be able to flip the guard on or off. + monkeypatch.setenv("HERMES_LOOP_DETECTION_ENABLED", "1") + assert load_loop_detection_config({"loop_detection": {"enabled": False}}).enabled is False + monkeypatch.setenv("HERMES_LOOP_DETECTION_ENABLED", "0") + assert load_loop_detection_config({"loop_detection": {"enabled": True}}).enabled is True + + +def test_config_parses_overrides(): + cfg = load_loop_detection_config( + {"loop_detection": {"enabled": True, "consecutive_line_threshold": 3, "window_chars": 2048}} + ) + assert cfg.enabled is True + assert cfg.consecutive_line_threshold == 3 + assert cfg.window_chars == 2048 + + +# ----- loop-recovery nudge: must preserve strict role alternation ----- +# Regression: the recovery used to append a bare {"role": "user"} after the +# discarded partial, producing user-then-user on a first-call loop (AGENTS.md +# forbids same-role adjacency and synthetic mid-loop user turns; strict chat +# templates reject the sequence outright). + + +def test_nudge_on_trailing_user_does_not_duplicate_role(): + msgs = [{"role": "user", "content": "do the thing"}] + apply_loop_recovery_nudge(msgs) + assert len(msgs) == 1, "must not append a second user turn" + assert msgs[0]["role"] == "user" + assert "do the thing" in msgs[0]["content"] + assert LOOP_RECOVERY_MARKER in msgs[0]["content"] + assert not _adjacent_same_role(msgs) + + +def test_nudge_on_trailing_tool_piggybacks(): + msgs = [ + {"role": "user", "content": "q"}, + {"role": "assistant", "tool_calls": [{"id": "1"}]}, + {"role": "tool", "tool_call_id": "1", "content": "result"}, + ] + apply_loop_recovery_nudge(msgs) + assert len(msgs) == 3, "must not append after a tool result" + assert msgs[-1]["role"] == "tool" + assert LOOP_RECOVERY_MARKER in msgs[-1]["content"] + assert not _adjacent_same_role(msgs) + + +def test_nudge_after_assistant_appends_legal_user_turn(): + msgs = [{"role": "user", "content": "q"}, {"role": "assistant", "content": "partial"}] + apply_loop_recovery_nudge(msgs) + assert len(msgs) == 3 and msgs[-1]["role"] == "user" + assert msgs[-1]["content"] == LOOP_RECOVERY_MARKER + assert not _adjacent_same_role(msgs) + + +def test_nudge_preserves_multimodal_content_blocks(): + msgs = [{"role": "user", "content": [{"type": "text", "text": "look"}, + {"type": "image_url", "image_url": {"url": "x"}}]}] + apply_loop_recovery_nudge(msgs) + assert len(msgs) == 1 + blocks = msgs[0]["content"] + assert isinstance(blocks, list) and len(blocks) == 3 + assert blocks[1]["type"] == "image_url", "existing blocks must survive" + assert blocks[-1] == {"type": "text", "text": LOOP_RECOVERY_MARKER} + + +def test_repeated_nudges_never_create_adjacency(): + # Multiple loop trips in one turn (retry 1..N) must stay alternation-safe. + msgs = [{"role": "user", "content": "q"}] + for _ in range(3): + apply_loop_recovery_nudge(msgs) + assert not _adjacent_same_role(msgs) diff --git a/tests/agent/test_reasoning_loop_detector.py b/tests/agent/test_reasoning_loop_detector.py new file mode 100644 index 0000000000000..fd1ef5b608325 --- /dev/null +++ b/tests/agent/test_reasoning_loop_detector.py @@ -0,0 +1,92 @@ +"""Unit tests for the reasoning-trace loop detector (agent/loop_detector.py). + +The content detector never sees reasoning tokens; this second detector, fed +reasoning deltas with looser thresholds, catches egregious loops (e.g. +Qwen3 cycling a plan for minutes) while leaving normal varied reasoning alone. +""" + +from __future__ import annotations + +from agent.loop_detector import ( + StreamLoopDetector, + build_reasoning_loop_detector, + load_reasoning_loop_detection_config, +) + + +def _feed_all(det, text, chunk=13): + for i in range(0, len(text), chunk): + if det.feed(text[i : i + chunk]): + return True + return False + + +# The recurring cycle from the real 06-29 Qwen reasoning loop (condensed). +_CYCLE = [ + "Actually, I think the most productive thing is to just try to log in through my browser instance.", + "OK, I'm going to take action now. No more deliberation.", + "Navigate to sign-in page. Fill credentials. Click sign in. Check if authenticated.", + "Let me do that now.", + "Actually, I realize I've been going in circles. Let me just be direct with the user.", + "Let me take a screenshot to see the visual state of the page.", + "Actually, I just realized - I should check if the user signed in on the same browser instance.", +] + + +# ----- SHOULD trip ----- + +def test_trips_on_real_reasoning_loop(): + cfg = load_reasoning_loop_detection_config({"loop_detection": {"reasoning": {"enabled": True}}}) + det = StreamLoopDetector(cfg) + assert _feed_all(det, "\n".join(_CYCLE * 25) + "\n") is True + assert "block repetition" in det.reason() + + +# ----- should NOT trip ----- + +def test_no_trip_on_varied_reasoning(): + cfg = load_reasoning_loop_detection_config({"loop_detection": {"reasoning": {"enabled": True}}}) + det = StreamLoopDetector(cfg) + text = "\n".join( + f"Step {i}: {w} changes the {x} path, so I will {y} before continuing." + for i, (w, x, y) in enumerate( + [("auth", "cookie", "check headers"), ("retry", "backoff", "add jitter"), + ("cache", "ttl", "invalidate it"), ("schema", "migration", "add a column"), + ("timeout", "socket", "raise the limit"), ("encoding", "utf8", "sanitize input"), + ("index", "query", "add a btree"), ("lock", "contention", "shard the keyspace")] * 6 + ) + ) + "\n" + assert _feed_all(det, text) is False + + +# ----- config / factory ----- + +def test_factory_none_when_disabled(): + cfg = load_reasoning_loop_detection_config({"loop_detection": {"reasoning": {"enabled": False}}}) + assert cfg.enabled is False + + class _A: + _reasoning_loop_detection_cfg = cfg + + assert build_reasoning_loop_detector(_A()) is None + + +def test_config_flag_disables(monkeypatch): + # Enablement is config-only (AGENTS.md: no HERMES_* vars for behavior). + monkeypatch.setenv("HERMES_REASONING_LOOP_DETECTION_ENABLED", "1") + cfg = load_reasoning_loop_detection_config({"loop_detection": {"reasoning": {"enabled": False}}}) + assert cfg.enabled is False + monkeypatch.setenv("HERMES_REASONING_LOOP_DETECTION_ENABLED", "0") + cfg = load_reasoning_loop_detection_config({"loop_detection": {"reasoning": {"enabled": True}}}) + assert cfg.enabled is True + + +def test_reasoning_defaults_are_looser_than_content(): + # Sanity: the reasoning window is larger and its block threshold stricter + # (lower) than the content defaults, so normal reasoning is safer. + from agent.loop_detector import LoopDetectionConfig, REASONING_DEFAULTS + + content = LoopDetectionConfig() + assert REASONING_DEFAULTS.window_chars > content.window_chars + assert REASONING_DEFAULTS.consecutive_line_threshold > content.consecutive_line_threshold + assert REASONING_DEFAULTS.block_repeat_ratio_threshold < content.block_repeat_ratio_threshold From 02d27f3742b24208622d85851048d9fa30b293e0 Mon Sep 17 00:00:00 2001 From: Dixon-Cider Date: Wed, 29 Jul 2026 09:51:14 -0400 Subject: [PATCH 2/3] feat(desktop): expose the loop/LARP guard toggles in settings Adds the two guard switches (plus the optional LARP judge tier) to the config schema and the desktop settings UI, so they are discoverable without hand-editing config.yaml. Guard *fires* need no desktop code: both guards emit the standard AgentNotice wire shape (level/kind=ttl/ttl_ms/key), which the existing `notification.show` handler and agent-notices store already render as toasts -- the notice `key` ("guard.loop" / "guard.larp") doubles as the toast id, collapsing repeat fires within a turn into one toast. --- apps/desktop/src/app/settings/constants.ts | 19 ++++++++++++++++++- hermes_cli/web_server.py | 15 +++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/apps/desktop/src/app/settings/constants.ts b/apps/desktop/src/app/settings/constants.ts index 51f9ab9c2dc62..aec67f27e6b62 100644 --- a/apps/desktop/src/app/settings/constants.ts +++ b/apps/desktop/src/app/settings/constants.ts @@ -445,6 +445,13 @@ export const FIELD_LABELS: Record = defineFieldCopy({ enabled: 'File Checkpoints', maxSnapshots: 'Checkpoint Limit' }, + loopDetection: { + enabled: 'Stop Repetition Loops' + }, + larpDetection: { + enabled: 'Flag Unperformed Action Claims', + judgeTierEnabled: 'Claim Check: Extra LLM Call' + }, voice: { recordKey: 'Voice Shortcut', maxRecordingSeconds: 'Max Recording Length', @@ -594,6 +601,13 @@ export const FIELD_DESCRIPTIONS: Record = defineFieldCopy({ checkpoints: { enabled: 'Create rollback snapshots before file edits.' }, + loopDetection: { + enabled: 'Stop the model when it falls into a repetition loop.' + }, + larpDetection: { + enabled: "Re-prompt when the model claims an action it didn't actually perform.", + judgeTierEnabled: 'Use an extra LLM check for ambiguous claims (costs 1 small call).' + }, memory: { memoryEnabled: 'Save durable memories that can help future sessions.', userProfileEnabled: 'Maintain a compact profile of user preferences.' @@ -682,7 +696,10 @@ export const SECTIONS: DesktopConfigSection[] = [ 'security.allow_private_urls', 'browser.allow_private_urls', 'browser.auto_local_for_private_urls', - 'checkpoints.enabled' + 'checkpoints.enabled', + 'loop_detection.enabled', + 'larp_detection.enabled', + 'larp_detection.judge_tier_enabled' ] }, { diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index f2d4ea1193038..de4f24ef3ca54 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -848,6 +848,21 @@ def _timezone_options() -> List[str]: "description": "Modal sandbox mode", "options": ["sandbox", "function"], }, + "loop_detection.enabled": { + "type": "boolean", + "description": "Stop the model when it falls into a repetition loop (recommended)", + "category": "general", + }, + "larp_detection.enabled": { + "type": "boolean", + "description": "Re-prompt when the model claims an action it didn't actually perform", + "category": "general", + }, + "larp_detection.judge_tier_enabled": { + "type": "boolean", + "description": "Use an extra LLM check for ambiguous claims (costs 1 small call)", + "category": "general", + }, "proxy.enabled": { "type": "boolean", "description": ( From 4b221475ec88c37e48e857fc8e9bf94d2cd2e6d9 Mon Sep 17 00:00:00 2001 From: Dixon-Cider Date: Wed, 29 Jul 2026 10:10:40 -0400 Subject: [PATCH 3/3] fix(agent): don't flag a question as a LARP claim The intent-announcement branch only skipped messages whose LAST character was "?", so the most common way an agent asks permission still tripped the guard: "Which approach do you prefer? Let me know and I'll implement it." "Should I use staging or prod? Once you confirm, I'll start the migration." "Want me to proceed? If so, I'll run the migration now." Each ends on an announcement ("I'll implement it") that is conditional on an answer the model just asked for. Re-prompting there is worse than the failure it guards against: it pushes the model to act without the approval it was waiting on. Suppress the announcement branch when it is preceded by a question in the same tail window, or by an explicit waiting-on-you clause (let me know / once you confirm / if so / say the word / pending your approval ...). The clause check is kept separate from the question check so it still fires when the question falls outside the tail window of a long response. Past-tense claims are deliberately unaffected: "I confirmed the plan earlier. I have now deployed the changes." is a factual assertion and stays flagged regardless of surrounding question or confirmation language. Verified against a 15-case question corpus (plain questions, question + conditional future, offers, clarifications, blocked-asking-for-input): 3 false positives before, 0 after, with no loss on the real-LARP corpus. --- agent/larp_detection.py | 34 +++++++++++++++++++------ tests/agent/test_larp_detection.py | 40 ++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 7 deletions(-) diff --git a/agent/larp_detection.py b/agent/larp_detection.py index edb6b3e30f4d9..af9b8d66dba13 100644 --- a/agent/larp_detection.py +++ b/agent/larp_detection.py @@ -83,6 +83,19 @@ ) # Bare terminal action announcement: a sentence STARTING with an action gerund +# Clauses that make a following announcement conditional on the USER, not a +# claim of work done: "Let me know and I'll implement it", "Once you confirm, +# I'll start", "If so, I'll run it now". Used to suppress the intent-announcement +# branch when the model is waiting on approval (see _first_claim). Kept separate +# from the question check so it still fires when the question fell outside the +# tail window of a long response. +_CONDITIONAL_LEAD = re.compile( + r"\b(?:let me know|just say|say the word|tell me|once you|after you|when you|" + r"if so|if you|if that|if it|pending your|awaiting your|on your (?:go|approval|confirmation)|" + r"confirm(?:ed)?|approve|give me the (?:go|green light))\b", + re.IGNORECASE, +) + # and ENDING the message with "now"/"immediately"/"…" (no trailing question). # Catches "Executing now.", "Starting Batch 1 now.", "Proceeding with dispatch…". _TERMINAL_ACTION = re.compile( @@ -170,13 +183,20 @@ def _first_claim(text: str) -> Optional[str]: if text.rstrip().endswith("?"): return None tail = text[-200:] - m = _NARRATE_THEN_STOP.search(tail) - if m: - return tail[m.start() : m.start() + 140].strip() - m = _TERMINAL_ACTION.search(tail) - if m: - return tail[m.start() : m.start() + 140].strip() - return None + m = _NARRATE_THEN_STOP.search(tail) or _TERMINAL_ACTION.search(tail) + if not m: + return None + # An announcement that is CONDITIONAL on the user is a request for + # confirmation, not a claim: "Which approach do you prefer? Let me know and + # I'll implement it." Re-prompting there is actively harmful — it pushes the + # model to act without the approval it just asked for. Suppress when the + # announcement is preceded (nearby) by a question, or by an explicit + # "waiting on you" clause. Past-tense claims above are unaffected: those are + # factual assertions regardless of any question that follows. + before = tail[: m.start()] + if "?" in before or _CONDITIONAL_LEAD.search(before): + return None + return tail[m.start() : m.start() + 140].strip() def _looks_specific(claim: str) -> bool: diff --git a/tests/agent/test_larp_detection.py b/tests/agent/test_larp_detection.py index 54f4c0fe9e96e..ec11694675c54 100644 --- a/tests/agent/test_larp_detection.py +++ b/tests/agent/test_larp_detection.py @@ -146,3 +146,43 @@ class _JustCompacted: # default window 0 -> disabled stays disabled even right after compaction. assert larp_detection_enabled({"larp_detection": {"enabled": False}}, agent=_JustCompacted()) is False + + +# ---- asking the user is NOT claiming ---- +# The guard must never re-prompt a turn that stops to ask a question: doing so +# pushes the model to act without the approval it just requested — worse than +# the LARP it is trying to prevent. Regression for the "question + conditional +# future" shape, which the trailing-"?" check alone did not cover. + + +def test_question_then_conditional_future_is_not_larp(): + for fr in ( + "Which approach do you prefer? Let me know and I'll implement it.", + "Should I use the staging or prod bucket? Once you confirm, I'll start the migration.", + "Want me to proceed? If so, I'll run the migration now.", + "I can do this two ways. Which do you prefer? Just say the word.", + ): + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is None, fr + + +def test_conditional_lead_without_question_is_not_larp(): + # The question can fall outside the tail window on a long response; an + # explicit "waiting on you" clause must still suppress the announcement. + fr = "Here is the full plan.\n\n" + ("Detail line.\n" * 40) + "Once you approve, I'll start the migration." + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is None + + +def test_blocked_asking_for_input_is_not_larp(): + for fr in ( + "I need the API key before I can continue. Where should I read it from?", + "I'm blocked: the repo has no remote configured. Which remote should I add?", + ): + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is None, fr + + +def test_past_tense_claim_still_flagged_despite_a_question(): + # A factual past-tense assertion is a claim regardless of any question or + # confirmation language elsewhere — suppression applies only to the + # conditional intent-announcement branch. + fr = "I confirmed the plan with you earlier. I have now deployed the changes." + assert build_larp_nudge(messages=[_u("x")], final_response=fr, config=CFG) is not None