# Agent Stall Resilience > Detects stalls and repetition loops in AI agent sessions and implements auto-respawn, circuit breakers, and graceful recovery to maintain session continuity under prolonged inactivity or infinite message cycles. - Skill: `paulpas/agent-stall-resilience` (Agent Skill, multi-file: 2 files) - Install (CLI): `npx skillmds@latest add paulpas/agent-stall-resilience` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/agent-stall-resilience/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/agent-stall-resilience --- # Agent Stall Resilience Detects stalls and infinite repetition loops in AI agent sessions and implements auto-respawn, circuit breakers, and graceful recovery. This skill makes the model actively monitor session health — tracking elapsed time since last activity, comparing recent messages for repetition, and injecting recovery instructions when a stall or loop is detected — while also providing router-level timeout protection that wraps every HTTP handler with deadline enforcement. Stall resilience is not a single detection rule. It is a layered defense: **time-based monitoring** catches long pauses, **similarity-based loop detection** catches rapid repetition, **auto-respawn** recovers the session by pruning loops and re-injecting context, and **router-level timeouts** prevent individual requests from blocking the entire system indefinitely. ## TL;DR Checklist - [ ] SessionStallDetector configured with warning (60s) and critical (120s) thresholds - [ ] RepetitionCircuitBreaker initialized with sliding window of 5 messages and threshold of 3 repeats - [ ] AutoRespawnOrchestrator registered as fallback handler for all stalled sessions - [ ] Universal loop-guard instruction (light or aggressive variant) injected into every session's system prompt - [ ] Self-correction protocol active when model detects repetition - [ ] Output conciseness rule enforced after stall recovery - [ ] Router /route and /execute handlers wrapped with Promise.race timeout protection - [ ] All error paths use specific exception types — no bare except clauses - [ ] Auto-respawn preserves conversation context (minus the looping segment) --- ## When to Use Use this skill when: - An OpenCode agent appears stuck in a repetition loop, re-asking the same question or regenerating identical responses - A session has been idle for more than 60 seconds without tool calls or new messages - The skill router API layer returns 504-like timeouts and you need deadline enforcement on all HTTP handlers - You are building resilience into an agent orchestration system that must auto-recover from model hangs - Debugging a session where the model enters an infinite reasoning or tool-call cycle ## When NOT to Use Avoid this skill for: - **Simple debugging** of non-stall issues — use `agent-debugging` (stacktrace root cause) instead - **One-off error recovery** after a single exception — use `exception-handling-recovery` instead - **Long-running batch processing** where stalls are expected (e.g., training jobs) — use timeout-aware batching strategies instead - **UI/frontend sessions** where user input is inherently slow — auto-respawn would destroy user progress --- ## Core Workflow 1. **Initialize Stall Monitoring** — Create a `SessionStallDetector` instance with warning threshold at 60 seconds and critical threshold at 120 seconds. Start the monitor by recording the current timestamp and the last activity timestamp (defaulting to now). Register it as a background check that runs before every agent response generation. **Checkpoint:** Verify thresholds are between 30s and 300s — below 30s causes false positives during complex reasoning, above 300s allows unbounded hangs. 2. **Inject Universal Loop-Guard Instruction** — Append one of the two pre-configured loop-guard system prompt templates (see section below) to every session's instructions. Choose the **light guard** variant (threshold=3, one response tolerance) for routine tasks and the **aggressive guard** variant (threshold=5, zero-tolerance) for complex or high-risk tasks. Both variants include built-in self-correction protocols, stall-awareness clauses, and output conciseness rules — no manual configuration needed. **Checkpoint:** Verify the injected instruction references itself as a "persistent system directive" and includes all five recovery sub-sections (repetition check, self-correction protocol, stall awareness, completion behavior, conciseness rule). 3. **Start Repetition Monitoring** — Initialize a `RepetitionCircuitBreaker` with a sliding window of 5 recent agent messages and a repetition threshold of 3. After each model response, add it to the circuit breaker's window using the message digest hash. If the same digest appears ≥ 3 times within the window, trigger the circuit breaker immediately — do not wait for the time-based stall detector. **Checkpoint:** Verify the `reset()` method is registered as the recovery callback that clears the sliding window after respawn completes. 4. **Evaluate Detection State** — Before generating each new response, check: (a) time since last tool call or message exceeds warning threshold → log a warning and reduce response verbosity to be more concise; (b) time exceeds critical threshold OR circuit breaker is tripped → invoke `AutoRespawnOrchestrator.snapshot_and_respawn()` to prune the looping segment, inject recovery instruction, and continue with fresh context. **Checkpoint:** Verify that the respawning session preserves the original conversation history minus only the messages within the last window of repeated responses (typically the last 3–5 messages). 5. **Execute Router-Level Timeout Protection** — Wrap every `/route` and `/execute` HTTP handler in the skill router with a `Promise.race` timeout pattern using 60-second deadlines. If a request exceeds the deadline, return a structured error (`StallTimeoutError`) with session context and trigger auto-respawn on the client side. **Checkpoint:** Verify that timeout responses include enough metadata (session_id, elapsed_ms, last_activity_type) for diagnostic tracing. 6. **Verify Recovery Completeness** — After an auto-respawn completes, confirm the new session can generate a novel response (i.e., its first output does not match any message in the pre-respawn sliding window). If it does repeat within 2 additional cycles, escalate to manual intervention and log the stall pattern for post-mortem analysis. **Checkpoint:** Post-recovery novelty check passes — the new response is semantically distinct from all messages in the last `window_size` entries before respawn. --- ## Implementation Patterns ### Pattern 1: SessionStallDetector Monitors elapsed time since last agent activity (tool call or response). Returns a detection state of `idle`, `warning`, or `stalled`. Configurable thresholds allow tuning for different session types. ```python """Session-level stall detection for OpenCode agent sessions.""" from __future__ import annotations import time from dataclasses import dataclass, field from enum import Enum from typing import Callable class StallState(str, Enum): """Detection states for an agent session's stall status.""" IDLE = "idle" # Within normal operating range WARNING = "warning" # Approaching stall threshold; reduce verbosity STALLED = "stalled" # Critical threshold exceeded; respawn required @dataclass(frozen=True) class StallEvent: """Immutable event emitted when the stall state changes.""" timestamp: float # time.time() when the state changed from_state: StallState to_state: StallState elapsed_seconds: float # Total time since last activity reason: str # Human-readable explanation @dataclass class SessionStallDetector: """Monitors elapsed time since last agent activity in a session. Tracks the wall-clock time between consecutive tool calls and responses. When the elapsed time crosses the warning threshold, returns `warning` state. When it crosses the critical threshold, returns `stalled` state. Usage: detector = SessionStallDetector(warning_threshold=60, critical_threshold=120) detector.record_activity() # Call after every tool call or response for _ in range(30): state = detector.check() if state == StallState.STALLED: handle_stall(detector.last_event) time.sleep(1) Args: warning_threshold: Seconds before the critical threshold to emit a warning. Must be between 5 and (critical_threshold - 1). critical_threshold: Seconds after which the session is considered stalled. Must be >= 30 to avoid false positives during complex reasoning. session_id: Optional identifier for logging and tracing purposes. Raises: ValueError: If thresholds are misconfigured or invalid. """ warning_threshold: float = 60.0 critical_threshold: float = 120.0 session_id: str = field(default="default") _last_activity: float = field(default_factory=time.time) _event_history: list[StallEvent] = field(default_factory=list, repr=False) _callbacks: list[Callable[[StallEvent], None]] = field(default_factory=list, repr=False) def __post_init__(self) -> None: if self.critical_threshold < 30.0: raise ValueError( f"critical_threshold must be >= 30 seconds " f"(got {self.critical_threshold}). Lower values cause false positives " "during complex reasoning or tool-call-heavy sessions." ) if self.warning_threshold >= self.critical_threshold: raise ValueError( f"warning_threshold ({self.warning_threshold}) must be less than " f"critical_threshold ({self.critical_threshold})" ) def record_activity(self) -> None: """Mark that the agent has performed an activity (tool call or response). Call this method after every tool invocation and after every model response to reset the stall timer. This is the single point of truth for session liveness. """ self._last_activity = time.time() def check(self) -> StallState: """Evaluate the current stall state based on elapsed time since last activity. Returns: StallState.IDLE when elapsed time is below the warning threshold. StallState.WARNING when elapsed time is between warning and critical thresholds. StallState.STALLED when elapsed time exceeds the critical threshold. Side effects: Records a StallEvent if the state has changed. Emits all registered callbacks with the new event. """ now = time.time() elapsed = now - self._last_activity previous_state = self._determine_state(elapsed) new_state = self._resolve_state(elapsed) if new_state != previous_state: # State has changed — record the transition and fire callbacks event = StallEvent( timestamp=now, from_state=previous_state, to_state=new_state, elapsed_seconds=elapsed, reason=self._event_reason(new_state, elapsed), ) self._last_emitted_state = new_state self._event_history.append(event) for cb in self._callbacks: try: cb(event) except Exception as e: # Callback failure must not break stall detection itself print(f"[stall-detector] callback error: {e}") # pragma: no cover return new_state _last_emitted_state: StallState = field(default=StallState.IDLE, init=False) def _determine_state(self, elapsed: float) -> StallState: """Pure function mapping elapsed time to the previous state value.""" if elapsed >= self.critical_threshold: return StallState.STALLED if elapsed >= self.warning_threshold: return StallState.WARNING return StallState.IDLE def _resolve_state(self, elapsed: float) -> StallState: """Determine the current state, auto-resolving to IDLE if activity occurred.""" if elapsed < self.critical_threshold: return StallState.IDLE # Session recovered after crossing threshold return self._last_emitted_state @staticmethod def _event_reason(state: StallState, elapsed: float) -> str: """Generate a human-readable explanation for the state transition.""" if state == StallState.STALLED: return ( f"Session idle for {elapsed:.1f}s " f"(critical threshold: {SessionStallDetector().critical_threshold:.0f}s)" ) return f"Session idle for {elapsed:.1f}s (warning threshold: 60s)" def register_callback(self, callback: Callable[[StallEvent], None]) -> None: """Register a callback to receive stall state change events. Args: callback: A function that takes a single StallEvent argument. Called synchronously on every state transition. """ self._callbacks.append(callback) def get_status(self) -> dict: """Return a serializable snapshot of the detector's current status. Returns: Dict with keys: session_id, last_activity_ago_seconds, state, warning_threshold, critical_threshold, event_count, recent_events. """ elapsed = time.time() - self._last_activity return { "session_id": self.session_id, "last_activity_ago_seconds": round(elapsed, 2), "state": self.check().value, "warning_threshold": self.warning_threshold, "critical_threshold": self.critical_threshold, "event_count": len(self._event_history), "recent_events": [ { "timestamp": e.timestamp, "from": e.from_state.value, "to": e.to_state.value, "elapsed": round(e.elapsed_seconds, 2), "reason": e.reason, } for e in self._event_history[-5:] ], } def reset(self) -> None: """Reset the detector to initial idle state. Clears all event history and resets the last activity timestamp to now. Useful when a session has been respawned or recovered. """ self._last_activity = time.time() self._event_history.clear() self._last_emitted_state = StallState.IDLE # --- Usage Example --- def demo_stall_detection() -> None: """Demonstrate stall detection with simulated time progression.""" detector = SessionStallDetector( warning_threshold=5, # 5 seconds for demo (normally 60) critical_threshold=10, # 10 seconds for demo (normally 120) session_id="demo-session-1", ) state_history: list[tuple[float, StallState]] = [] def on_state_change(event: StallEvent) -> None: print(f"[{event.timestamp:.1f}] {event.from_state.value} → {event.to_state.value}: {event.reason}") state_history.append((event.timestamp, event.to_state)) detector.register_callback(on_state_change) # Simulate a session with varying activity patterns for second in range(15): detector.record_activity() if second % 8 == 0 else None # Activity every 8 seconds state = detector.check() elapsed = time.time() - detector._last_activity print(f" sec={second:2d} | state={state.value:7s} | elapsed={elapsed:.1f}s") if not state_history: print(" (no state transitions occurred during this run)") if __name__ == "__main__": demo_stall_detection() ``` ### Pattern 2: RepetitionCircuitBreaker Sliding window of N recent agent messages with similarity-based detection. When the same message digest appears ≥ threshold times, triggers a circuit breaker with a fallback response. The `reset()` method clears the window after recovery. ```python """Repetition loop detection with sliding-window circuit breaker for agent sessions.""" from __future__ import annotations import hashlib import re from collections import deque from dataclasses import dataclass, field from typing import Callable @dataclass class RepetitionEvent: """Event emitted when the repetition circuit breaker trips. Attributes: digest: The repeated message hash. occurrences: Number of times this exact message appeared in the window. window_size: Total messages currently in the sliding window. threshold: The configured repetition threshold that was exceeded. """ digest: str occurrences: int window_size: int threshold: int class RepetitionCircuitBreaker: """Sliding-window circuit breaker for detecting message repetition loops. Maintains a deque of recent agent messages, each hashed via SHA-256 on a normalized version of the text (lowercased, whitespace-collapsed). When any digest repeats at or above the configured threshold within the window, the circuit breaker trips and emits a RepetitionEvent. Normalization removes trailing punctuation and strips numbers to catch variations of repeated prompts like "Try again" vs "Try again." vs "Try step 2". Usage: breaker = RepetitionCircuitBreaker(window_size=5, threshold=3) breaker.record("What is the capital of France?") breaker.record("What is the capital of France?") if breaker.is_tripped(): print(f"Repetition detected! Digest: {breaker.last_event.digest}") Args: window_size: Number of recent messages to track. Messages older than this are evicted when the window fills. threshold: Number of repetitions required to trip the breaker. A value of 3 means the same message must appear 3 times in the window. normalization_mode: How to normalize messages before hashing. Options: - "fuzzy": collapse whitespace, lowercase, strip trailing punctuation - "exact": use raw text (preserves casing and punctuation differences) Raises: ValueError: If window_size < 2 or threshold < 2. """ def __init__( self, window_size: int = 5, threshold: int = 3, normalization_mode: str = "fuzzy", ) -> None: if window_size < 2: raise ValueError("window_size must be >= 2") if threshold < 2: raise ValueError("threshold must be >= 2") self.window_size: int = window_size self.threshold: int = threshold self.normalization_mode: str = normalization_mode self._messages: deque[str] = deque(maxlen=window_size) self._digests: deque[str] = deque(maxlen=window_size) self._last_event: RepetitionEvent | None = None self._tripped: bool = False self._on_trip_callbacks: list[Callable[[RepetitionEvent], None]] = [] @staticmethod def normalize(text: str, mode: str = "fuzzy") -> str: """Normalize message text for consistent hashing. Args: text: The raw message string to normalize. mode: Normalization strategy ("fuzzy" or "exact"). Returns: Normalized string suitable for comparison and hashing. """ if mode == "exact": return text # Fuzzy normalization: lowercased, whitespace-collapsed, trailing punctuation stripped normalized = text.lower().strip() normalized = re.sub(r"\s+", " ", normalized) # collapse whitespace normalized = normalized.rstrip(".,;:!?)\"'") # strip trailing punctuation return normalized def digest(self, text: str) -> str: """Compute SHA-256 digest of a normalized message. Args: text: The raw message to hash. Returns: Hex-encoded SHA-256 hash of the normalized message text. """ normalized = self.normalize(text, self.normalization_mode) return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:16] # 16-char short hash def record(self, message: str) -> None: """Record a new agent message into the sliding window. Normalizes the message, computes its digest, and appends it to the sliding window. After adding, checks if any digest has reached the repetition threshold. If so, trips the circuit breaker. Args: message: The raw message text from the agent model. """ msg_digest = self.digest(message) # Evict oldest if window is full (deque maxlen handles this automatically) self._messages.append(message) self._digests.append(msg_digest) # Count occurrences of each digest in the current window digest_counts: dict[str, int] = {} for d in self._digests: digest_counts[d] = digest_counts.get(d, 0) + 1 # Check if any digest exceeds the threshold repeated_digests = {d: c for d, c in digest_counts.items() if c >= self.threshold} if repeated_digests and not self._tripped: most_common_digest, most_common_count = max( repeated_digests.items(), key=lambda item: item[1] ) event = RepetitionEvent( digest=most_common_digest, occurrences=most_common_count, window_size=len(self._messages), threshold=self.threshold, ) self._tripped = True self._last_event = event for cb in self._on_trip_callbacks: try: cb(event) except Exception as e: # pragma: no cover print(f"[repetition-breaker] callback error: {e}") def is_tripped(self) -> bool: """Check if the circuit breaker is currently in a tripped state. Returns True if the threshold was exceeded and reset() has not been called. A tripped breaker must be explicitly cleared via reset() before monitoring resumes — this prevents silent recovery from a genuine infinite loop. """ return self._tripped def reset(self) -> None: """Clear the sliding window and untrip the circuit breaker. Called after auto-respawn completes to start fresh monitoring. Empties both the message deque and digest deque, resets the tripped flag, and clears the last event record. """ self._messages.clear() self._digests.clear() self._tripped = False self._last_event = None def get_window(self) -> list[str]: """Return a copy of the current sliding window contents (oldest first). Returns: List of messages currently tracked in the sliding window. """ return list(self._messages) def get_status(self) -> dict: """Return a serializable status snapshot. Returns: Dict with keys: tripped, window_size, threshold, message_count, last_event (or None), most_common_digest_counts. """ digest_counts: dict[str, int] = {} for d in self._digests: digest_counts[d] = digest_counts.get(d, 0) + 1 top_counts = sorted(digest_counts.items(), key=lambda x: x[1], reverse=True)[:3] return { "tripped": self._tripped, "window_size": self.window_size, "threshold": self.threshold, "message_count": len(self._messages), "last_event": ( { "digest": self._last_event.digest if self._last_event else None, "occurrences": self._last_event.occurrences if self._last_event else None, } if self._last_event else None ), "most_common_digest_counts": [ {"digest": d[:8] + "...", "count": c} for d, c in top_counts ], } def on_trip(self, callback: Callable[[RepetitionEvent], None]) -> None: """Register a callback to fire when the circuit breaker trips. Args: callback: Function that receives the RepetitionEvent when tripped. """ self._on_trip_callbacks.append(callback) # --- Usage Example --- def demo_repetition_detection() -> None: """Demonstrate repetition detection with a simulated loop.""" breaker = RepetitionCircuitBreaker(window_size=5, threshold=3) # Simulate a normal conversation print("=== Normal conversation ===") for msg in [ "What is 2+2?", "The answer is 4.", "And what is 3*3?", "The answer is 9.", ]: breaker.record(msg) status = breaker.get_status() print(f"Window: {status['message_count']} messages | Tripped: {status['tripped']}") # Simulate a loop — same message repeated print("\n=== Repetition loop detected ===") for _ in range(4): breaker.record("I don't know. Let me think about this again.") if breaker.is_tripped(): event = breaker.last_event print(f"TRIpped! Digest: {event.digest} | Occurrences: {event.occurrences}") breaker.reset() print(f"After reset — Tripped: {breaker.is_tripped()}") if __name__ == "__main__": demo_repetition_detection() ``` ### Pattern 3: AutoRespawnOrchestrator Combines stall detection + loop detection to manage full session recovery. On trigger, it snapshots the current session context, creates a respawn instruction, and returns a new prompt that preserves conversation history minus the looping segment. ```python """Auto-respawn orchestrator for recovering stalled or looping agent sessions. This module provides SessionContext (serializable session snapshot), AutoRespawnOrchestrator (combines stall and loop detection into recovery), and utility functions for pruing repeated segments from conversation history. """ from __future__ import annotations import time from dataclasses import dataclass, field from typing import Any @dataclass class SessionContext: """Serializable snapshot of an agent session's state at a point in time. Stores the complete conversation history up to the respawn trigger, along with metadata about stall and repetition detection status. This context is used to reconstruct the session after respawn — preserving useful history while discarding the looping segment. Attributes: session_id: Unique identifier for this session. conversation_history: List of role/content dicts representing the full conversation up to the respawn point. stall_detector_state: Serializable state from SessionStallDetector. repetition_breaker_state: Serializable state from RepetitionCircuitBreaker. spawn_timestamp: Wall-clock time when this snapshot was taken. pruned_message_count: Number of messages removed during pruning (0 if none). """ session_id: str conversation_history: list[dict] # [{"role": "user"/"assistant", "content": "..."}, ...] stall_detector_state: dict = field(default_factory=dict) repetition_breaker_state: dict = field(default_factory=dict) spawn_timestamp: float = field(default_factory=time.time) pruned_message_count: int = 0 def to_dict(self) -> dict[str, Any]: """Serialize the full context to a flat dictionary. Returns: Dict suitable for JSON serialization or storage in a session database. """ return { "session_id": self.session_id, "conversation_history": self.conversation_history, "stall_detector_state": self.stall_detector_state, "repetition_breaker_state": self.repetition_breaker_state, "spawn_timestamp": self.spawn_timestamp, "pruned_message_count": self.pruned_message_count, } @classmethod def from_dict(cls, data: dict[str, Any]) -> SessionContext: """Deserialize a SessionContext from a flat dictionary. Args: data: The serialized context dict (output of to_dict). Returns: A new SessionContext with all fields restored. """ return cls( session_id=data["session_id"], conversation_history=data["conversation_history"], stall_detector_state=data.get("stall_detector_state", {}), repetition_breaker_state=data.get("repetition_breaker_state", {}), spawn_timestamp=data.get("spawn_timestamp", time.time()), pruned_message_count=data.get("pruned_message_count", 0), ) @dataclass class RespawnInstruction: """A recovery instruction to inject into a freshly respawned session. This is the prompt fragment that tells the model it has just been restored from a stall/loop and should proceed with a clean, focused approach. Attributes: preamble: Human-readable explanation of what happened. instructions: Numbered list of specific actions for the model to take now. preserved_context_summary: Brief description of retained conversation history. pruned_segment_description: Description of what was removed and why. """ preamble: str instructions: list[str] = field(default_factory=list) preserved_context_summary: str = "" pruned_segment_description: str = "" def format_as_prompt_fragment(self) -> str: """Format this instruction as a system message fragment for injection. Returns: A markdown-formatted string ready to be prepended to the model's system prompt on respawn. """ lines = [ "---", f"**SESSION RECOVERED:** {self.preamble}", ] if self.preserved_context_summary: lines.append(f"Retained context: {self.preserved_context_summary}") if self.pruned_segment_description: lines.append(f"Removed (repeated): {self.pruned_segment_description}") if self.instructions: lines.append("**Proceed with these steps:**") for i, instruction in enumerate(self.instructions, 1): lines.append(f"{i}. {instruction}") lines.append("---") return "\n".join(lines) class AutoRespawnOrchestrator: """Combines stall detection + loop detection to manage full session recovery. Monitors a session for stalls and repetition loops. When either trigger fires, it prunes the problematic messages from conversation history, preserves all useful prior context, generates a respawn instruction, and returns a new prompt that the model can use to continue with a clean slate. The orchestration follows the principle of minimal disruption: preserve everything up to the point where the loop began, remove only the repeated messages, and inject a clear recovery signal so the model knows it must break its current pattern. Args: max_prune_depth: Maximum number of trailing messages to consider removing. Defaults to 5 — this prevents accidentally deleting valid conversation history. fallback_response: Message to return when no context can be preserved (e.g., session was entirely looping). If None, a generic recovery message is used. Raises: ValueError: If max_prune_depth < 1. """ _DEFAULT_FALLBACK = "Session was recovered from a stall or infinite loop. Proceeding with a focused, single-step approach." def __init__( self, max_prune_depth: int = 5, fallback_response: str | None = None, ) -> None: if max_prune_depth < 1: raise ValueError("max_prune_depth must be >= 1") self.max_prune_depth: int = max_prune_depth self.fallback_response: str = fallback_response or self._DEFAULT_FALLBACK def snapshot_and_respawn( self, conversation_history: list[dict], stall_state: str = "idle", repetition_tripped: bool = False, repitition_digest: str = "", ) -> tuple[SessionContext, RespawnInstruction]: """Snapshot the session and generate a respawn plan. Prunes repeated messages from the end of conversation history, preserves all prior context, and produces a SessionContext + RespawnInstruction pair. This is the primary entry point called when either stall or repetition detection fires. It handles both types of stall uniformly by treating them as "prune the last N messages and continue with fresh instructions." Args: conversation_history: Full conversation list of {"role": ..., "content": ...} dicts. stall_state: Current stall detector state ("idle", "warning", or "stalled"). repetition_tripped: Whether the repetition circuit breaker is currently tripped. repitition_digest: The repeated message digest (if known). Returns: A tuple of (SessionContext, RespawnInstruction) ready for injection. Raises: ValueError: If conversation_history is empty and no fallback context exists. """ if not conversation_history: raise ValueError( "Cannot respawn with empty conversation history. " "Provide at least one prior message or set a fallback_response." ) # --- Determine pruning scope --- prune_from_index = self._determine_prune_point(conversation_history) preserved_messages = conversation_history[:prune_from_index] pruned_messages = conversation_history[prune_from_index:] # --- Build the session context snapshot --- context = SessionContext( session_id="session-" + str(int(time.time())), conversation_history=preserved_messages, pruned_message_count=len(pruned_messages), ) # --- Build the respawn instruction --- trigger_reasons: list[str] = [] if stall_state == "stalled": trigger_reasons.append(f"session stalled (idle for critical duration)") elif stall_state == "warning": trigger_reasons.append(f"session in warning state ({stall_state})") if repetition_tripped: trigger_reasons.append( f"repetition loop detected (digest={repitition_digest[:8] if repitition_digest else 'N/A'})" ) instruction = RespawnInstruction( preamble=f"Session was triggered for recovery: {'; '.join(trigger_reasons)}. " f"{len(pruned_messages)} trailing message(s) removed to break the loop.", instructions=self._generate_recovery_instructions(len(preserved_messages), len(pruned_messages)), preserved_context_summary=( f"{len(preserved_messages)} prior messages retained" if preserved_messages else "no prior context (using fallback)" ), pruned_segment_description=( f"The last {len(pruned_messages)} message(s) were removed because they represented a repetition or stall pattern." if pruned_messages else "no messages pruned — continuing from current position" ), ) return context, instruction def _determine_prune_point(self, history: list[dict]) -> int: """Determine the index at which to start pruning (exclusive). Examines the last N messages (up to max_prune_depth) for repetition patterns. Returns the index after the last unique message before the loop began. If no clear loop is detected, prunes the last 2 messages as a conservative reset. Args: history: Full conversation history list. Returns: Index in the history list to begin pruning from (exclusive). """ if len(history) == 0: return 0 # Look at the last max_prune_depth messages for repeated content depth = min(self.max_prune_depth, len(history)) recent = history[-depth:] if depth < 2: # Not enough messages to detect a loop — prune just the last one return max(0, len(history) - 1) # Check for identical or near-identical consecutive messages digests: list[str] = [] for msg in recent: content = msg.get("content", "") if isinstance(msg, dict) else str(msg) normalized = RepetitionCircuitBreaker.normalize(content).lower() digest = hashlib_sha256_short(normalized) digests.append(digest) # Find the longest suffix of identical digests loop_end = len(digests) for i in range(len(digests) - 1, 0, -1): if digests[i] != digests[i - 1]: break loop_end = i # The prune point is: len(history) - depth + loop_end return len(history) - depth + loop_end def _generate_recovery_instructions( self, preserved_count: int, pruned_count: int ) -> list[str]: """Generate specific recovery instructions based on what was preserved/removed. Args: preserved_count: Number of messages retained in the new context. pruned_count: Number of messages removed to break the loop. Returns: A list of numbered instruction strings for the RespawnInstruction. """ if preserved_count == 0 and pruned_count > 0: return [ "Start fresh with a single, focused question.", "If unsure about the user's intent, ask one clarifying question.", ] instructions = [ "Acknowledge that the previous attempts did not progress, then proceed directly to solving the retained problem.", ] if pruned_count >= 3: instructions.append( "The loop involved repeated responses — provide a concise answer in at most one paragraph." ) else: instructions.append( "A brief stall was detected — continue with focused execution on the last retained task." ) instructions.append("Do not re-explain concepts already covered in the preserved history.") return instructions def hashlib_sha256_short(text: str) -> str: """Compute a short (8-char hex) SHA-256 hash for efficient comparison. Args: text: The string to hash. Returns: First 8 characters of the hex-encoded SHA-256 digest. """ import hashlib return hashlib.sha256(text.encode("utf-8")).hexdigest()[:8] # --- Usage Example --- def demo_auto_respawn() -> None: """Demonstrate full respawn flow with simulated conversation.""" orchestrator = AutoRespawnOrchestrator(max_prune_depth=5) # Simulate a conversation that enters a loop history: list[dict] = [ {"role": "user", "content": "What is the meaning of life?"}, {"role": "assistant", "content": "According to philosophy, 42."}, {"role": "user", "content": "I need more detail."}, # --- Loop begins --- {"role": "assistant", "content": "I don't know. Let me think about this again."}, {"role": "assistant", "content": "I don't know. Let me think about this again."}, {"role": "assistant", "content": "I don't know. Let me think about this again."}, ] print(f"Conversation before respawn: {len(history)} messages") context, instruction = orchestrator.snapshot_and_respawn( conversation_history=history, stall_state="stalled", repetition_tripped=True, repitition_digest="a1b2c3d4", ) print(f"\nContext preserved: {len(context.conversation_history)} messages") print(f"Pruned: {context.pruned_message_count} messages") print(f"\nRespawn instruction:\n{instruction.format_as_prompt_fragment()}") if __name__ == "__main__": demo_auto_respawn() ``` ### Pattern 4: RouterTimeoutWrapper (TypeScript) Wraps the skill router's `/route` and `/execute` HTTP handlers with a `Promise.race` timeout pattern. Matches the existing codebase style in `src/index.ts`. ```typescript /** * RouterTimeo …(truncated)