1# tical-code -- AI Agent Platform
2# Copyright (C) 2026 zizetu
3#
4# This program is free software: you can redistribute it and/or modify
5# it under the terms of the GNU Affero General Public License as published by
6# the Free Software Foundation, either version 3 of the License, or
7# (at your option) any later version.
8#
9# This program is distributed in the hope that it will be useful,
10# but WITHOUT ANY WARRANTY; without even the implied warranty of
11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12# GNU Affero General Public License for more details.
13#
14# You should have received a copy of the GNU Affero General Public License
15# along with this program.  If not, see <https://www.gnu.org/licenses/>.
16#
17# Original repository: https://github.com/zizetu/tical-agent
18#
19
20"""
21Doom Loop Detection - AgentLoop/stagnantdetect
22==========================================
23
24Detect agent stuck in tool call loops and provide automatic recovery strategies.
25
26Core design:
27- Sliding window signature matching with four orthogonal detectors covering different loop patterns
28- Adaptive threshold: dynamically adjust detection sensitivity based on tool call frequency
29- Auto-recovery after detection: switch tool/params, rollback N steps, degrade model, force summary
30- Cross-agent loop detection: multi-agent collaboration when A→B→A round-trip loop
31- semantic similarity judgment: not only detects identical content, also detects "same substance, different surface"
32
33four types of detectors:
341. generic_repeat - same(Tool,Parameter)repeatcall
352. poll_no_progress - same-args-same-result polling (no substantive progress)
363. ping_pong - A→B→A→Balternating-oscillation
374. cross_agent - cross-agent loop calls
38
39Security design (Audit fix):
40- Bounded state: detector short-circuit logic, return on CRITICAL
41- Thread-safe: all state operations protected by threading.Lock
42- State reset mechanism: clear detector internal state after recovery
43- Detector result cache: same batch of records not recomputed
44
45Design principles:
46- Default close, explicit enable (production-grade posture)
47- Zero external dependencies, pure stdlib
48- each agent has independent detector instance, no global state
49- detection and recovery strategy decoupled, independently configurable
50
51Author: Tical (Zize Tu)
52Version: see tical_code.__version__
53"""
54
55import hashlib
56import json
57import logging
58import threading
59import time
60from collections import deque
61from dataclasses import dataclass, field
62from enum import Enum
63from typing import Any, Callable, Dict, List, Optional, Tuple
64
65logger = logging.getLogger(__name__)
66
67
68# =============================================================================
69# Constants and enums
70# =============================================================================
71
72class LoopLevel(Enum):
73    """Loop detection severity level."""
74    NORMAL = "normal"       # normal, no loop
75    WARNING = "warning"     # warning, may be stuck in loop
76    CRITICAL = "critical"   # Severe, must interrupt
77
78
79class RecoveryAction(Enum):
80    """Recovery action after loop detection."""
81    NONE = "none"                               # no needrecover
82    RETRY_DIFFERENT_ARGS = "retry_different_args"  # retry with different params
83    SWITCH_TOOL = "switch_tool"                 # switch tool
84    ROLLBACK_STEPS = "rollback_steps"           # Rollback N steps
85    DOWNGRADE_MODEL = "downgrade_model"         # Degrademodel
86    FORCE_SUMMARIZE = "force_summarize"         # Force summarycurrentstatus
87
88
89class DetectorType(Enum):
90    """Detector type."""
91    GENERIC_REPEAT = "generic_repeat"       # genericRepetition detection
92    POLL_NO_PROGRESS = "poll_no_progress"   # poll-no-progress
93    PING_PONG = "ping_pong"                 # alternating-oscillation
94    CROSS_AGENT = "cross_agent"             # cross-agent loop
95
96
97# =============================================================================
98# Data structure
99# =============================================================================
100
101@dataclass
102class ToolCallRecord:
103    """Single tool call record.
104
105    Attributes:
106        tool_name: Tool name
107        args_hash: deterministic hash of parameters (sorted JSON serialize + SHA-256)
108        result_hash: deterministic hash of result (backfilled after execution)
109        result_text: Resulttext(used forsemanticsimilarityjudge)
110        timestamp: calltimestamp
111        agent_id: caller agent ID (used for cross-agent detection)
112    """
113    tool_name: str
114    args_hash: str
115    fuzzy_hash: str = ""   # Fuzzy hash for near-duplicate detection (bash commands)
116    result_hash: str = ""
117    result_text: str = ""
118    timestamp: float = field(default_factory=time.time)
119    agent_id: str = ""
120
121
122@dataclass
123class LoopDetectionResult:
124    """Loop detection result.
125
126    Attributes:
127        stuck: whetherstuckLoop
128        level: severelevel
129        detector: triggering detector type
130        count: repeat/oscillationcount
131        message: humanreadablemessage
132        recovery: recommended recovery action
133        details: detail info (for audit and trace)
134    """
135    stuck: bool = False
136    level: LoopLevel = LoopLevel.NORMAL
137    detector: Optional[DetectorType] = None
138    count: int = 0
139    message: str = ""
140    recovery: RecoveryAction = RecoveryAction.NONE
141    details: Dict = field(default_factory=dict)
142
143    def to_dict(self) -> Dict:
144        return {
145            'stuck': self.stuck,
146            'level': self.level.value,
147            'detector': self.detector.value if self.detector else None,
148            'count': self.count,
149            'message': self.message,
150            'recovery': self.recovery.value,
151            'details': self.details,
152        }
153
154
155@dataclass
156class DoomLoopConfig:
157    """Loop detection configuration.
158
159    base threshold + adaptive adjustment strategy:
160    - high-freq calls (>1 per second) → reduce threshold (faster loop detection)
161    - low-freq calls (<0.1 per second) → increase threshold (avoid false alarms)
162
163    Attributes:
164        enabled: whetherenabledetect(Defaultclose)
165        history_size: slidingwindowsize
166        warn_threshold_base: warning threshold base value
167        critical_threshold_base: critical threshold base value
168        adaptive_enabled: whetherenableadaptivethreshold
169        cross_agent_enabled: whether to enable cross-agent detection
170        semantic_similarity_enabled: whetherenablesemanticsimilarityjudge
171        semantic_threshold: semantic similarity threshold (0-1, above this treated as "same")
172        recovery_enabled: whetherenableAuto-recovery
173    """
174    enabled: bool = True
175    history_size: int = 30
176    warn_threshold_base: int = 2
177    critical_threshold_base: int = 3
178    adaptive_enabled: bool = True
179    cross_agent_enabled: bool = True
180    semantic_similarity_enabled: bool = True
181    semantic_threshold: float = 0.85
182    recovery_enabled: bool = True
183
184
185# =============================================================================
186# coreToolfunction
187# =============================================================================
188
189def _deterministic_hash(data: Any) -> str:
190    """Compute deterministic hash of data.
191
192    Uses sorted JSON serialize to ensure same data always produces same signature,
193    unaffected by dict traversal order.
194
195    Args:
196        data: data to hash (dict/list/str/number/bool/None)
197
198    Returns:
199        SHA-256 hex digest
200    """
201    try:
202        serialized = json.dumps(data, sort_keys=True, ensure_ascii=False, default=str)
203    except (TypeError, ValueError):
204        serialized = str(data)
205    return hashlib.sha256(serialized.encode('utf-8')).hexdigest()
206
207
208def _fuzzy_args_hash(tool_name: str, args: Dict[str, Any]) -> str:
209    """Fuzzy hash for near-duplicate detection - strips variable parts from bash commands.
210
211    For 'bash' tool calls, normalizes the command by stripping:
212    - timeout values (timeout N, timeout Ns, timeout=N)
213    - heredoc content (everything after << 'PYEOF' or << 'EOF')
214
215    Other tools use exact hash (fallback to _deterministic_hash).
216    """
217    if tool_name != "bash":
218        return _deterministic_hash(args)
219
220    cmd = args.get("command", "")
221    if not cmd:
222        return _deterministic_hash(args)
223
224    import re
225    # Normalize: strip timeout prefixes
226    normalized = re.sub(r'\btimeout\s+\d+s?\b', 'timeout X', cmd)
227    normalized = re.sub(r'\btimeout=\d+\b', 'timeout=X', normalized)
228    # Strip heredoc content - keep only the marker
229    normalized = re.sub(r"(<<\s*'?PYEOF'?\s*\n).*", r'\1[...heredoc...]', normalized, flags=re.DOTALL)
230    normalized = re.sub(r"(<<\s*'?EOF'?\s*\n).*", r'\1[...heredoc...]', normalized, flags=re.DOTALL)
231    # Collapse multiple spaces
232    normalized = re.sub(r'\s+', ' ', normalized).strip()
233
234    return hashlib.sha256(normalized.encode('utf-8')).hexdigest()
235
236
237def _text_similarity(text_a: str, text_b: str) -> float:
238    """Compute semantic similarity of two text segments (based on Jaccard word set).
239
240    Not only detects fully identical text, but also "same substance, different surface" cases,
241    e.g. poll results that differ only in timestamp but have the same core content.
242
243    Optimized for Chinese: uses character bigram tokenization rather than space tokenization.
244
245    Args:
246        text_a: first segment of text
247        text_b: second segment of text
248
249    Returns:
250        similarity [0.0, 1.0]
251    """
252    if not text_a or not text_b:
253        return 0.0
254
255    def _tokenize(text: str) -> set:
256        """Mixed Chinese-English tokenize: Chinese bigrams + English words."""
257        tokens = set()
258        # Englishword
259        words = text.lower().split()
260        tokens.update(f"w:{w}" for w in words if len(w) > 1)
261        # ChineseCharacterbigram
262        cjk_chars = []
263        for ch in text:
264            if '\u4e00' <= ch <= '\u9fff':
265                cjk_chars.append(ch)
266        for i in range(len(cjk_chars) - 1):
267            tokens.add(f"c:{cjk_chars[i]}{cjk_chars[i + 1]}")
268        return tokens
269
270    tokens_a = _tokenize(text_a)
271    tokens_b = _tokenize(text_b)
272
273    if not tokens_a or not tokens_b:
274        # return character-level comparison
275        return 1.0 if text_a == text_b else 0.0
276
277    intersection = tokens_a & tokens_b
278    union = tokens_a | tokens_b
279    return len(intersection) / len(union) if union else 0.0
280
281
282# =============================================================================
283# DoomLoopDetector - core detector
284# =============================================================================
285
286class DoomLoopDetector:
287    """
288    Agent loop/stagnation detector.
289
290    four types of detectors covering different loop patterns:
291    1. generic_repeat - generic repeat: same (tool, params) repeated calls
292    2. poll_no_progress - poll-no-progress: same-args-same-result polling
293    3. ping_pong - alternating-oscillation: A→B→A→B oscillation
294    4. cross_agent - cross-agent loop: A→B→A round-trip
295
296    Adaptive threshold: dynamically adjust warn/critical thresholds based on call frequency.
297    Auto-recovery strategy: provide specific recovery suggestions after detection and can execute callbacks.
298
299    Security fix:
300    - Bounded state: detector short-circuit logic, return on any CRITICAL
301    - Thread-safe: all state operations protected by threading.Lock
302    - State reset mechanism: clear detector internal state after recovery
303    - Detector result cache: same batch of records not recomputed
304
305    Usage:
306        detector = DoomLoopDetector(config)
307        # record before each tool call
308        detector.record_tool_call("read_file", {"path": "/tmp/a.txt"}, agent_id="main")
309        # record result after each tool call
310        detector.record_tool_outcome("read_file", {"path": "/tmp/a.txt"}, result_text)
311        # detectwhetherstuckLoop
312        result = detector.detect()
313        if result.stuck:
314            await detector.execute_recovery(result)
315    """
316
317    def __init__(self, config: Optional[DoomLoopConfig] = None):
318        self.config = config or DoomLoopConfig()
319        # sliding window: latest records on the right side
320        self._history: deque = deque(maxlen=self.config.history_size)
321        # callfrequencytrack(used foradaptivethreshold)
322        self._call_timestamps: deque = deque(maxlen=100)
323        # Recovery strategy map: default recovery action for each detector type
324        self._recovery_strategies: Dict[DetectorType, RecoveryAction] = {
325            DetectorType.GENERIC_REPEAT: RecoveryAction.SWITCH_TOOL,
326            DetectorType.POLL_NO_PROGRESS: RecoveryAction.FORCE_SUMMARIZE,
327            DetectorType.PING_PONG: RecoveryAction.ROLLBACK_STEPS,
328            DetectorType.CROSS_AGENT: RecoveryAction.DOWNGRADE_MODEL,
329        }
330        # custom recovery callback
331        self._recovery_callbacks: Dict[RecoveryAction, Optional[Callable]] = {
332            action: None for action in RecoveryAction
333        }
334        # [Security fix] thread safety lock
335        self._lock = threading.Lock()
336        # [Security fix] detector result cache
337        self._last_detect_hash: str = ""  # hash of history at last detection
338        self._cached_result: Optional[LoopDetectionResult] = None
339        # [Security fix] detector internal state (used for state reset)
340        self._detector_state: Dict[str, Any] = {}
341
342    def record_tool_call(
343        self,
344        tool_name: str,
345        args: Dict[str, Any],
346        agent_id: str = "",
347    ) -> None:
348        """Record a tool call (before call).
349
350        Args:
351            tool_name: Tool name
352            args: Tool parameters
353            agent_id: caller agent ID
354        """
355        if not self.config.enabled:
356            return
357
358        args_hash = _deterministic_hash(args)
359        fuzzy_hash = _fuzzy_args_hash(tool_name, args)
360        record = ToolCallRecord(
361            tool_name=tool_name,
362            args_hash=args_hash,
363            fuzzy_hash=fuzzy_hash,
364            agent_id=agent_id,
365        )
366
367        # [Security fix]threadsecurity
368        with self._lock:
369            self._history.append(record)
370            self._call_timestamps.append(time.time())
371            # after new record is added, cache invalid
372            self._last_detect_hash = ""
373            self._cached_result = None
374
375    def record_tool_outcome(
376        self,
377        tool_name: str,
378        args: Dict[str, Any],
379        result_text: str,
380        agent_id: str = "",
381    ) -> None:
382        """Record tool call result (backfill after call).
383
384        Find matching unfilled record in sliding window and backfill result hash.
385
386        Args:
387            tool_name: Tool name
388            args: Tool parameters
389            result_text: Tool execution result text
390            agent_id: caller agent ID
391        """
392        if not self.config.enabled:
393            return
394
395        args_hash = _deterministic_hash(args)
396        result_hash = _deterministic_hash(result_text)
397
398        # [Security fix]threadsecurity
399        with self._lock:
400            # find matching not-yet-backfilled record from tail
401            for record in reversed(self._history):
402                if (record.tool_name == tool_name
403                        and record.args_hash == args_hash
404                        and not record.result_hash
405                        and record.agent_id == agent_id):
406                    record.result_hash = result_hash
407                    record.result_text = result_text
408                    # after result update, cache is invalid
409                    self._last_detect_hash = ""
410                    self._cached_result = None
411                    return
412
413            # if no matching record found (history size may be too small), record a complete record
414            record = ToolCallRecord(
415                tool_name=tool_name,
416                args_hash=args_hash,
417                result_hash=result_hash,
418                result_text=result_text,
419                agent_id=agent_id,
420            )
421            self._history.append(record)
422            self._last_detect_hash = ""
423            self._cached_result = None
424
425    def detect(self) -> LoopDetectionResult:
426        """Run all detectors, return the earliest detected loop.
427
428        Detectors run by priority: generic_repeat → poll_no_progress → ping_pong → cross_agent.
429        [Security fix] short-circuit logic: any detector triggering CRITICAL immediately returns, no further detection.
430        [Security fix] Result cache: same batch of records not recomputed.
431
432        Returns:
433            LoopDetectionResult
434        """
435        if not self.config.enabled:
436            return LoopDetectionResult()
437
438        # [Security fix]threadsecurityread
439        with self._lock:
440            history_snapshot = list(self._history)
441            timestamps_snapshot = list(self._call_timestamps)
442
443        if len(history_snapshot) < 3:
444            return LoopDetectionResult()
445
446        # [Security fix]Detector result cache: same batch of records not recomputed
447        current_hash = _deterministic_hash([
448            (r.tool_name, r.args_hash, r.result_hash, r.agent_id)
449            for r in history_snapshot
450        ])
451        if current_hash == self._last_detect_hash and self._cached_result is not None:
452            return self._cached_result
453
454        # computeadaptivethreshold
455        warn_threshold, critical_threshold = self._adaptive_thresholds(
456            timestamps_snapshot
457        )
458
459        # [Security fix] sequentially run detectors, CRITICAL short-circuit
460        for detector_fn in [
461            self._detect_generic_repeat,
462            self._detect_poll_no_progress,
463            self._detect_ping_pong,
464        ]:
465            result = detector_fn(history_snapshot, warn_threshold, critical_threshold)
466            if result.stuck:
467                # CRITICAL short-circuit: immediately return, do not continue detection
468                if result.level == LoopLevel.CRITICAL:
469                    self._update_cache(current_hash, result)
470                    return result
471                # WARNING also returns, but may still need subsequent detection
472                self._update_cache(current_hash, result)
473                return result
474
475        # cross-agent loop detection
476        if self.config.cross_agent_enabled:
477            result = self._detect_cross_agent(
478                history_snapshot, warn_threshold, critical_threshold
479            )
480            if result.stuck:
481                self._update_cache(current_hash, result)
482                return result
483
484        result = LoopDetectionResult()
485        self._update_cache(current_hash, result)
486        return result
487
488    def register_recovery_callback(
489        self, action: RecoveryAction, callback: Callable
490    ) -> None:
491        """Register recovery strategy callback.
492
493        Args:
494            action: Recovery action type
495            callback: callback function, signature: async def callback(result: LoopDetectionResult) -> bool
496        """
497        self._recovery_callbacks[action] = callback
498
499    def set_recovery_strategy(
500        self, detector_type: DetectorType, action: RecoveryAction
501    ) -> None:
502        """Set the recovery strategy for a detector.
503
504        Args:
505            detector_type: detector type
506            action: Recovery action
507        """
508        self._recovery_strategies[detector_type] = action
509
510    async def execute_recovery(self, result: LoopDetectionResult) -> bool:
511        """Execute recovery strategy.
512
513        auto-execute the corresponding recovery strategy after detecting a loop,
514        Rather than only returning stuck status and letting upper layer process on its own.
515
516        [Security fix] after recovery, clear the corresponding detector internal state,
517        avoid residual state affecting subsequent detection.
518
519        Args:
520            result: detectResult
521
522        Returns:
523            True indicates recovery success, can continue executing
524        """
525        if not self.config.recovery_enabled:
526            logger.warning(
527                f"[DoomLoop] loop detected but recovery strategy not enabled: {result.message}"
528            )
529            return False
530
531        action = self._recovery_strategies.get(
532            result.detector, RecoveryAction.FORCE_SUMMARIZE
533        )
534        result.recovery = action
535
536        callback = self._recovery_callbacks.get(action)
537        if callback:
538            try:
539                success = await callback(result)
540                # [Security fix] after recovery success, clear corresponding detector internal state
541                if success:
542                    self._reset_detector_state(result.detector)
543                    # Post-recovery validation: re-detect to confirm recovery actually resolved the loop
544                    post_recovery = self.detect()
545                    if post_recovery.stuck:
546                        logger.warning(
547                            f"[DoomLoop] post-recovery validation FAILED - loop persists after {action.value}. "
548                            f"Escalating: {post_recovery.message}"
549                        )
550                        return False
551                return success
552            except Exception as e:
553                logger.error(f"[DoomLoop] Recovery strategyExecuteFailed: {e}")
554                return False
555        else:
556            logger.info(
557                f"[DoomLoop] recommended recovery strategy: {action.value}, but no callback registered."
558                f"detectResult: {result.message}"
559            )
560            return False
561
562    def reset(self) -> None:
563        """Reset detector state (call when processing a new message)."""
564        with self._lock:
565            self._history.clear()
566            self._call_timestamps.clear()
567            self._detector_state.clear()
568            self._last_detect_hash = ""
569            self._cached_result = None
570
571    # =========================================================================
572    # internal detectors
573    # =========================================================================
574
575    def _update_cache(self, current_hash: str, result: LoopDetectionResult) -> None:
576        """Update detection result cache."""
577        self._last_detect_hash = current_hash
578        self._cached_result = result
579
580    def _reset_detector_state(self, detector_type: Optional[DetectorType]) -> None:
581        """[Security fix] Reset specified detector internal state.
582
583        Called after recovery success, avoid residual state affecting subsequent detection.
584
585        Args:
586            detector_type: detector type to reset
587        """
588        if detector_type is None:
589            return
590
591        with self._lock:
592            state_key = detector_type.value
593            if state_key in self._detector_state:
594                self._detector_state[state_key] = {}
595                logger.debug(
596                    f"[DoomLoop] detector '{state_key}' internal state reset"
597                )
598
599    def _adaptive_thresholds(
600        self, timestamps: Optional[List[float]] = None
601    ) -> Tuple[int, int]:
602        """Compute adaptive threshold.
603
604        based on call frequency dynamic adjustment:
605        - high-freq calls (>1 per second) → reduce threshold (faster loop detection)
606        - low-freq calls (<0.1 per second) → increase threshold (avoid false alarms)
607
608        Args:
609            timestamps: timestamplist(threadsecuritysnapshot)
610
611        Returns:
612            (warn_threshold, critical_threshold)
613        """
614        if not self.config.adaptive_enabled:
615            return self.config.warn_threshold_base, self.config.critical_threshold_base
616
617        if timestamps is None:
618            with self._lock:
619                timestamps = list(self._call_timestamps)
620
621        if len(timestamps) < 2:
622            return self.config.warn_threshold_base, self.config.critical_threshold_base
623
624        now = time.time()
625        recent = [ts for ts in timestamps if now - ts < 60]
626        if len(recent) < 2:
627            return self.config.warn_threshold_base, self.config.critical_threshold_base
628
629        # per-secondcallcount
630        duration = recent[-1] - recent[0]
631        freq = len(recent) / duration if duration > 0 else len(recent)
632
633        warn = self.config.warn_threshold_base
634        critical = self.config.critical_threshold_base
635
636        if freq > 1.0:
637            # high-freq: reduce threshold, faster detection
638            warn = max(5, int(warn * 0.5))
639            critical = max(10, int(critical * 0.5))
640        elif freq > 0.5:
641            # medium-high freq: slightly lower
642            warn = max(6, int(warn * 0.7))
643            critical = max(12, int(critical * 0.7))
644        elif freq < 0.1:
645            # low-freq: increase threshold, avoid false alarms
646            warn = min(int(warn * 1.5), 50)  # Cap at 50 max
647            critical = min(int(critical * 1.5), 100)  # Cap at 100 max
648
649        return warn, critical
650
651    def _detect_generic_repeat(
652        self,
653        history: List[ToolCallRecord],
654        warn_threshold: int,
655        critical_threshold: int,
656    ) -> LoopDetectionResult:
657        """Detector 1: generic repeat - same (tool, params) repeated calls.
658
659        count calls within window where (tool_name, args_hash) are fully identical,
660        report if threshold exceeded.
661
662        Args:
663            history: historyrecordsnapshot
664            warn_threshold: warningthreshold
665            critical_threshold: criticalthreshold
666
667        Returns:
668            LoopDetectionResult
669        """
670        counts: Dict[Tuple[str, str], int] = {}
671        for record in history:
672            # For bash: use fuzzy hash to catch near-duplicates (diff timeout, same command)
673            fh = record.fuzzy_hash or record.args_hash
674            key = (record.tool_name, fh)
675            counts[key] = counts.get(key, 0) + 1
676
677        # check in descending order of repeat count
678        for (tool_name, args_hash), count in sorted(
679            counts.items(), key=lambda x: -x[1]
680        ):
681            if count >= critical_threshold:
682                return LoopDetectionResult(
683                    stuck=True,
684                    level=LoopLevel.CRITICAL,
685                    detector=DetectorType.GENERIC_REPEAT,
686                    count=count,
687                    message=(
688                        f"Tool '{tool_name}' repeated call with same params {count} times, "
689                        f"exceeds critical threshold {critical_threshold}. May be stuck in infinite loop."
690                    ),
691                    details={
692                        'tool_name': tool_name,
693                        'args_hash': args_hash[:16],
694                        'count': count,
695                        'threshold': critical_threshold,
696                    },
697                )
698            elif count >= warn_threshold:
699                return LoopDetectionResult(
700                    stuck=True,
701                    level=LoopLevel.WARNING,
702                    detector=DetectorType.GENERIC_REPEAT,
703                    count=count,
704                    message=(
705                        f"Tool '{tool_name}' repeated call with same params {count} times, "
706                        f"exceeds warning threshold {warn_threshold}. Suggest switching strategy."
707                    ),
708                    details={
709                        'tool_name': tool_name,
710                        'args_hash': args_hash[:16],
711                        'count': count,
712                        'threshold': warn_threshold,
713                    },
714                )
715
716        return LoopDetectionResult()
717
718    def _detect_poll_no_progress(
719        self,
720        history: List[ToolCallRecord],
721        warn_threshold: int,
722        critical_threshold: int,
723    ) -> LoopDetectionResult:
724        """Detector 2: poll-no-progress - same-args-same-result polling.
725
726        count consecutive "no-progress" from tail forward.
727        No need for keyword matching to judge poll tool (too coarse), rather look at whether result changes.
728
729        Semantic similarity enhancement: timestamp changed but content unchanged also counts as no-progress.
730
731        Args:
732            history: historyrecordsnapshot
733            warn_threshold: warningthreshold
734            critical_threshold: criticalthreshold
735
736        Returns:
737            LoopDetectionResult
738        """
739        no_progress_count = 0
740        last_args_hash = None
741        last_result_hash = None
742        last_result_text = None
743
744        for record in reversed(history):
745            if not record.result_hash:
746                continue
747
748            if record.args_hash == last_args_hash:
749                if record.result_hash == last_result_hash:
750                    # params and result fully same → no-progress
751                    no_progress_count += 1
752                elif (self.config.semantic_similarity_enabled
753                      and last_result_text is not None):
754                    # semantic similarity judgment: surface different but substance same
755                    sim = _text_similarity(record.result_text, last_result_text)
756                    if sim >= self.config.semantic_threshold:
757                        no_progress_count += 1
758                    else:
759                        break
760                else:
761                    break
762            elif last_args_hash is None:
763                # first record, initialization
764                last_args_hash = record.args_hash
765                last_result_hash = record.result_hash
766                last_result_text = record.result_text
767                no_progress_count = 1
768            else:
769                break
770
771        if no_progress_count >= critical_threshold:
772            return LoopDetectionResult(
773                stuck=True,
774                level=LoopLevel.CRITICAL,
775                detector=DetectorType.POLL_NO_PROGRESS,
776                count=no_progress_count,
777                message=(
778                    f"consecutive {no_progress_count} calls returned same result (no-progress), "
779                    f"exceeds critical threshold {critical_threshold}. Poll may be hanging."
780                ),
781                details={'count': no_progress_count, 'threshold': critical_threshold},
782            )
783        elif no_progress_count >= warn_threshold:
784            return LoopDetectionResult(
785                stuck=True,
786                level=LoopLevel.WARNING,
787                detector=DetectorType.POLL_NO_PROGRESS,
788                count=no_progress_count,
789                message=(
790                    f"consecutive {no_progress_count} calls returned same result, "
791                    f"exceeds warning threshold {warn_threshold}. Suggest summarizing current status."
792                ),
793                details={'count': no_progress_count, 'threshold': warn_threshold},
794            )
795
796        return LoopDetectionResult()
797
798    def _detect_ping_pong(
799        self,
800        history: List[ToolCallRecord],
801        warn_threshold: int,
802        critical_threshold: int,
803    ) -> LoopDetectionResult:
804        """Detector 3: alternating-oscillation - A→B→A→B oscillation.
805
806        detect if the window tail has two different call patterns strictly alternating.
807        not only detects param alternation, also detects tool name alternation.
808
809        Args:
810            history: historyrecordsnapshot
811            warn_threshold: warningthreshold
812            critical_threshold: criticalthreshold
813
814        Returns:
815            LoopDetectionResult
816        """
817        if len(history) < 4:
818            return LoopDetectionResult()
819
820        tail_records = list(history)[-20:]
821        if len(tail_records) < 4:
822            return LoopDetectionResult()
823
824        signatures = [(r.tool_name, r.args_hash) for r in tail_records]
825
826        # find two different signatures from tail
827        pattern_a = signatures[-1]
828        pattern_b = None
829        for sig in reversed(signatures[:-1]):
830            if sig != pattern_a:
831                pattern_b = sig
832                break
833
834        if pattern_b is None:
835            return LoopDetectionResult()
836
837        # Verifytailwhetherstrictalternate
838        alternating_count = 0
839        expected = pattern_a
840        for sig in reversed(signatures):
841            if sig == expected:
842                alternating_count += 1
843                expected = pattern_b if expected == pattern_a else pattern_a
844            else:
845                break
846
847        if alternating_count >= critical_threshold:
848            return LoopDetectionResult(
849                stuck=True,
850                level=LoopLevel.CRITICAL,
851                detector=DetectorType.PING_PONG,
852                count=alternating_count,
853                message=(
854                    f"detected alternating-oscillation pattern: "
855                    f"'{pattern_a[0]}' ↔ '{pattern_b[0]}', "
856                    f"alternated {alternating_count} times. "
857                    f"Agent oscillates back and forth between two states."
858                ),
859                details={
860                    'pattern_a': pattern_a[0],
861                    'pattern_b': pattern_b[0],
862                    'count': alternating_count,
863                },
864            )
865        elif alternating_count >= warn_threshold:
866            return LoopDetectionResult(
867                stuck=True,
868                level=LoopLevel.WARNING,
869                detector=DetectorType.PING_PONG,
870                count=alternating_count,
871                message=(
872                    f"may exist alternating-oscillation: "
873                    f"'{pattern_a[0]}' ↔ '{pattern_b[0]}', "
874                    f"alternated {alternating_count} times."
875                ),
876                details={
877                    'pattern_a': pattern_a[0],
878                    'pattern_b': pattern_b[0],
879                    'count': alternating_count,
880                },
881            )
882
883        return LoopDetectionResult()
884
885    def _detect_cross_agent(
886        self,
887        history: List[ToolCallRecord],
888        warn_threshold: int,
889        critical_threshold: int,
890    ) -> LoopDetectionResult:
891        """Detector 4: cross-agent loop - A→B→A round-trip.
892
893        In multi-agent scenario, Agent A calls Agent B, B calls back A, forming a cross-agent loop.
894        detection method: count alternating patterns of agent_id in adjacent records.
895
896        Args:
897            history: historyrecordsnapshot
898            warn_threshold: warningthreshold
899            critical_threshold: criticalthreshold
900
901        Returns:
902            LoopDetectionResult
903        """
904        if len(history) < 4:
905            return LoopDetectionResult()
906
907        records_with_agent = [r for r in history if r.agent_id]
908        if len(records_with_agent) < 4:
909            return LoopDetectionResult()
910
911        tail_agents = [r.agent_id for r in records_with_agent[-20:]]
912        if len(set(tail_agents)) < 2:
913            return LoopDetectionResult()
914
915        agent_a = tail_agents[-1]
916        agent_b = None
917        for aid in reversed(tail_agents[:-1]):
918            if aid != agent_a:
919                agent_b = aid
920                break
921
922        if agent_b is None:
923            return LoopDetectionResult()
924
925        # VerifyalternateMode
926        alternating_count = 0
927        expected = agent_a
928        for aid in reversed(tail_agents):
929            if aid == expected:
930                alternating_count += 1
931                expected = agent_b if expected == agent_a else agent_a
932            else:
933                break
934
935        if alternating_count >= critical_threshold:
936            return LoopDetectionResult(
937                stuck=True,
938                level=LoopLevel.CRITICAL,
939                detector=DetectorType.CROSS_AGENT,
940                count=alternating_count,
941                message=(
942                    f"detected cross-agent loop: "
943                    f"'{agent_a}' ↔ '{agent_b}', "
944                    f"alternated {alternating_count} times."
945                ),
946                details={
947                    'agent_a': agent_a,
948                    'agent_b': agent_b,
949                    'count': alternating_count,
950                },
951            )
952        elif alternating_count >= warn_threshold:
953            return LoopDetectionResult(
954                stuck=True,
955                level=LoopLevel.WARNING,
956                detector=DetectorType.CROSS_AGENT,
957                count=alternating_count,
958                message=(
959                    f"may exist cross-agent loop: "
960                    f"'{agent_a}' ↔ '{agent_b}', "
961                    f"alternated {alternating_count} times."
962                ),
963                details={
964                    'agent_a': agent_a,
965                    'agent_b': agent_b,
966                    'count': alternating_count,
967                },
968            )
969
970        return LoopDetectionResult()
971
972    # =========================================================================
973    # Helper method
974    # =========================================================================
975
976    def get_stats(self) -> Dict:
977        """Get detector statistics info."""
978        with self._lock:
979            return {
980                'enabled': self.config.enabled,
981                'history_size': len(self._history),
982                'max_history': self.config.history_size,
983                'call_timestamps': len(self._call_timestamps),
984                'adaptive_enabled': self.config.adaptive_enabled,
985                'cross_agent_enabled': self.config.cross_agent_enabled,
986            }
987