stall_detector.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/stall_detector.py
 
"""Pure, turn-scoped detection of repeated tool-call stalls.
 
Purpose
-------
Detect a genuine runaway tool loop from observations the ReAct loop already
owns. The detector has no event-bus, presenter, session or dispatch dependency;
it records completed calls and evaluates the next proposed call.
 
Responsibilities
----------------
- Canonicalise tool arguments so key order and string whitespace are irrelevant.
- Deny the N-th identical proposal after N-1 matching calls completed.
- Detect identical-error streaks and iterations containing no new read or write.
- Carry short structured evidence that the loop can render for the operator.
 
Contracts
---------
- All state is turn-scoped and :meth:`reset` removes it completely.
- :meth:`observe` is called exactly once per completed call.
- A write-class call or any new call shape resets the no-progress streak.
- Thresholds are positive integers; configuration resolution lives elsewhere.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- json (parsing tool arguments supplied as JSON text and dumping every argument set as sorted, compact canonical JSON)
- dataclasses (the frozen StallEvidence and Stall records)
- oc_chat.core.defaults (lazy; ceiling - the three stall thresholds for from_defaults)
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L2 - a stall is detected
    L3 - state resets
    L4 - module load
"""
from __future__ import annotations
 
import json
import re
from dataclasses import dataclass
from typing import Any, Mapping
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.CORE", "module", "loaded", 4)
 
 
@dataclass(frozen=True)
class StallEvidence:
    """The bounded facts needed to explain one detected stall."""
 
    tool: str
    completed_count: int
    proposed_count: int | None = None
    last_error_summary: str | None = None
    iteration_start: int | None = None
    iteration_end: int | None = None
 
 
@dataclass(frozen=True)
class Stall:
    """A detected signal and the evidence that tripped it."""
 
    kind: str
    evidence: StallEvidence
 
 
def _normalise(value: Any) -> Any:
    """Return a JSON-compatible value with whitespace-normalised strings."""
    if isinstance(value, str):
        return re.sub(r"\s+", " ", value).strip()
    if isinstance(value, Mapping):
        return {str(key): _normalise(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_normalise(item) for item in value]
    if value is None or isinstance(value, (bool, int, float)):
        return value
    return repr(value)
 
 
def canonical_arguments(arguments: Any) -> str:
    """Return stable JSON for arguments supplied as JSON text or Python values."""
    parsed = arguments
    if isinstance(arguments, str):
        try:
            parsed = json.loads(arguments)
        except (json.JSONDecodeError, TypeError):
            parsed = arguments
    return json.dumps(
        _normalise(parsed),
        ensure_ascii=False,
        separators=(",", ":"),
        sort_keys=True,
    )
 
 
class StallDetector:
    """Hold the three stall signals for one turn."""
 
    def __init__(
        self,
        *,
        repeat_calls: int,
        error_streak: int,
        no_progress_iterations: int,
    ) -> None:
        thresholds = (repeat_calls, error_streak, no_progress_iterations)
        if any(isinstance(value, bool) or not isinstance(value, int) or value <= 0
               for value in thresholds):
            raise ValueError("stall thresholds must be positive integers")
        self.repeat_calls = repeat_calls
        self.error_streak = error_streak
        self.no_progress_iterations = no_progress_iterations
        self.reset()
 
    @classmethod
    def from_defaults(cls) -> "StallDetector":
        """Construct a detector from the three shared ceiling resolutions."""
        from oc_chat.core.defaults import ceiling
 
        return cls(
            repeat_calls=ceiling("stall_repeat_calls"),
            error_streak=ceiling("stall_error_streak"),
            no_progress_iterations=ceiling("stall_no_progress_iterations"),
        )
 
    @staticmethod
    def _shape(tool: str, arguments: Any) -> tuple[str, str]:
        return str(tool), canonical_arguments(arguments)
 
    def reset(self) -> None:
        """Clear every observation at the start of a turn."""
        func = "StallDetector.reset"
        self._shape_counts: dict[tuple[str, str], int] = {}
        self._shape_last_error: dict[tuple[str, str], str | None] = {}
        self._error_summary: str | None = None
        self._error_count = 0
        self._error_start: int | None = None
        self._no_progress_count = 0
        self._no_progress_start: int | None = None
        self._iteration: int | None = None
        self._iteration_has_calls = False
        self._iteration_all_repeated = True
        self._iteration_has_write = False
        self._iteration_tool = ""
        self._pending: Stall | None = None
        _dbg("OC_CHAT.CORE", func, "turn state cleared", 3)
 
    def reset_shape(self, tool: str, arguments: Any) -> None:
        """Allow a human-approved call shape to start a fresh count."""
        func = "StallDetector.reset_shape"
        shape = self._shape(tool, arguments)
        self._shape_counts.pop(shape, None)
        self._shape_last_error.pop(shape, None)
        self._error_summary = None
        self._error_count = 0
        self._error_start = None
        self._no_progress_count = 0
        self._no_progress_start = None
        self._pending = None
        self._reset_iteration()
        _dbg("OC_CHAT.CORE", func, f"shape reset for tool={tool}", 3)
 
    def check_proposal(self, tool: str, arguments: Any) -> Stall | None:
        """Return a pending signal or deny the N-th matching proposal."""
        func = "StallDetector.check_proposal"
        if self._pending is not None:
            return self._pending
        shape = self._shape(tool, arguments)
        completed = self._shape_counts.get(shape, 0)
        proposed = completed + 1
        if proposed < self.repeat_calls:
            return None
        stall = Stall(
            kind="repeat",
            evidence=StallEvidence(
                tool=str(tool),
                completed_count=completed,
                proposed_count=proposed,
                last_error_summary=self._shape_last_error.get(shape),
            ),
        )
        _dbg("OC_CHAT.CORE", func,
             f"stall=repeat tool={tool} completed={completed} proposed={proposed}", 2)
        return stall
 
    def observe(
        self,
        *,
        tool: str,
        arguments: Any,
        outcome: str,
        error_summary: str | None,
        is_write: bool,
        iteration: int,
        iteration_complete: bool,
    ) -> Stall | None:
        """Record one completed call and return a newly reached signal, if any."""
        if self._iteration is not None and iteration != self._iteration:
            self._finish_iteration()
        if self._iteration is None:
            self._iteration = iteration
 
        shape = self._shape(tool, arguments)
        repeated = self._shape_counts.get(shape, 0) > 0
        self._shape_counts[shape] = self._shape_counts.get(shape, 0) + 1
        self._shape_last_error[shape] = error_summary
 
        self._iteration_has_calls = True
        self._iteration_all_repeated = self._iteration_all_repeated and repeated
        self._iteration_has_write = self._iteration_has_write or bool(is_write)
        self._iteration_tool = str(tool)
 
        detected = self._observe_error(
            tool=str(tool),
            outcome=outcome,
            error_summary=error_summary,
            iteration=iteration,
        )
        if iteration_complete:
            detected = detected or self._finish_iteration()
        return detected
 
    def _observe_error(
        self,
        *,
        tool: str,
        outcome: str,
        error_summary: str | None,
        iteration: int,
    ) -> Stall | None:
        """Advance or reset the identical-error streak."""
        func = "StallDetector._observe_error"
        if outcome != "error" or not error_summary:
            self._error_summary = None
            self._error_count = 0
            self._error_start = None
            return None
        if error_summary == self._error_summary:
            self._error_count += 1
        else:
            self._error_summary = error_summary
            self._error_count = 1
            self._error_start = iteration
        if self._error_count < self.error_streak:
            return None
        stall = Stall(
            kind="error_streak",
            evidence=StallEvidence(
                tool=tool,
                completed_count=self._error_count,
                last_error_summary=error_summary,
                iteration_start=self._error_start,
                iteration_end=iteration,
            ),
        )
        self._pending = self._pending or stall
        _dbg("OC_CHAT.CORE", func,
             f"stall=error_streak tool={tool} count={self._error_count}", 2)
        return stall
 
    def _finish_iteration(self) -> Stall | None:
        """Fold the current iteration into the consecutive no-progress signal."""
        func = "StallDetector._finish_iteration"
        if self._iteration is None:
            return None
        iteration = self._iteration
        qualifies = (
            self._iteration_has_calls
            and self._iteration_all_repeated
            and not self._iteration_has_write
        )
        if qualifies:
            self._no_progress_count += 1
            if self._no_progress_count == 1:
                self._no_progress_start = iteration
        else:
            self._no_progress_count = 0
            self._no_progress_start = None
 
        stall = None
        if self._no_progress_count >= self.no_progress_iterations:
            stall = Stall(
                kind="no_progress",
                evidence=StallEvidence(
                    tool=self._iteration_tool,
                    completed_count=self._no_progress_count,
                    iteration_start=self._no_progress_start,
                    iteration_end=iteration,
                ),
            )
            self._pending = self._pending or stall
            _dbg("OC_CHAT.CORE", func,
                 f"stall=no_progress iterations={self._no_progress_count}", 2)
        self._reset_iteration()
        return stall
 
    def _reset_iteration(self) -> None:
        self._iteration = None
        self._iteration_has_calls = False
        self._iteration_all_repeated = True
        self._iteration_has_write = False
        self._iteration_tool = ""
 
 
def stall_sentence(stall: Stall) -> str:
    """Render one bounded sentence naming the loop and its evidence."""
    evidence = stall.evidence
    if stall.kind == "repeat":
        error = (
            f"; the last error was {evidence.last_error_summary}"
            if evidence.last_error_summary else ""
        )
        return (
            f"Stall detected: proposed call {evidence.proposed_count} to "
            f"`{evidence.tool}` repeats identical arguments after "
            f"{evidence.completed_count} completed calls{error}; the turn is halted."
        )
    if stall.kind == "error_streak":
        return (
            f"Stall detected: {evidence.completed_count} consecutive calls through "
            f"`{evidence.tool}` failed with the same error "
            f"{evidence.last_error_summary}; the turn is halted."
        )
    return (
        f"Stall detected: iterations {evidence.iteration_start}-"
        f"{evidence.iteration_end} made no progress across "
        f"{evidence.completed_count} repeated read-only iterations, ending with "
        f"`{evidence.tool}`; the turn is halted."
    )
 
 
__all__ = [
    "Stall",
    "StallDetector",
    "StallEvidence",
    "canonical_arguments",
    "stall_sentence",
]