Rename loop detector to token-window variant

Author: Mateusz

src/core/app/lifecycle.py

@@ -1,126 +1,134 @@
-from __future__ import annotations
-
-import asyncio
-import logging
-from typing import Any
-
-from fastapi import FastAPI
-
-from src.core.interfaces.session_service_interface import ISessionService
-
-logger = logging.getLogger(__name__)
-
-
-class AppLifecycle:
-    """Handles application lifecycle events.
-
-    This class manages startup and shutdown tasks for the application.
-    """
-
-    def __init__(self, app: FastAPI, config: dict[str, Any]) -> None:
-        """Initialize the lifecycle manager.
-
-        Args:
-            app: The FastAPI application
-            config: The application configuration
-        """
-        self.app = app
-        self.config = config
-        self._background_tasks: list[asyncio.Task] = []
-
-    async def startup(self) -> None:
-        """Perform startup tasks.
-
-        This method is called during application startup.
-        """
-        logger.info("Starting application lifecycle...")
-
-        # Start background tasks
-        self._start_background_tasks()
-
-    async def shutdown(self) -> None:
-        """Perform shutdown tasks.
-
-        This method is called during application shutdown.
-        """
-        logger.info("Shutting down application lifecycle...")
-
-        # Stop background tasks
-        await self._stop_background_tasks()
-
-        # Close any remaining connections
-        await self._close_connections()
-
-    def _start_background_tasks(self) -> None:
-        """Start background tasks."""
-        # Start session cleanup task
-        if self.config.get("session_cleanup_enabled", False):
-            interval = self.config.get(
-                "session_cleanup_interval", 3600
-            )  # 1 hour default
-            max_age = self.config.get("session_max_age", 86400)  # 1 day default
-
-            task = asyncio.create_task(
-                self._session_cleanup_task(interval, max_age), name="session_cleanup"
-            )
-            self._background_tasks.append(task)
-            logger.info(
-                f"Started session cleanup task (interval: {interval}s, max age: {max_age}s)"
-            )
-
-    async def _stop_background_tasks(self) -> None:
-        """Stop background tasks."""
-        for task in self._background_tasks:
-            if not task.done():
-                task.cancel()
-                try:
-                    await task
-                except asyncio.CancelledError:
-                    logger.info(f"Cancelled background task: {task.get_name()}")
-
-    async def _close_connections(self) -> None:
-        """Close any remaining connections."""
-        # Any connection cleanup code would go here
-
-    async def _session_cleanup_task(self, interval: int, max_age: int) -> None:
-        """Background task for cleaning up expired sessions.
-
-        Args:
-            interval: The interval in seconds between cleanup runs
-            max_age: The maximum age in seconds for sessions
-        """
-        try:
-            while True:
-                await asyncio.sleep(interval)
-
-                try:
-                    # Get service provider
-                    provider = self.app.state.service_provider
-                    if not provider:
-                        logger.warning(
-                            "Service provider not available for session cleanup"
-                        )
-                        continue
-
-                    # Get session service
-                    session_service = provider.get_service(ISessionService)
-                    if not session_service:
-                        logger.warning("Session service not available for cleanup")
-                        continue
-
-                    # Perform cleanup
-                    deleted_count = 0
-                    if hasattr(session_service, "cleanup_expired_sessions"):
-                        deleted_count = await session_service.cleanup_expired_sessions(
-                            max_age
-                        )
-
-                    if deleted_count > 0 and logger.isEnabledFor(logging.INFO):
-                        logger.info(f"Cleaned up {deleted_count} expired sessions")
-
-                except Exception as e:
-                    logger.error(f"Error during session cleanup: {e!s}")
-
-        except asyncio.CancelledError:
-            logger.debug("Session cleanup task cancelled")
-            raise
+from __future__ import annotations
+
+import asyncio
+import logging
+from contextlib import suppress
+from typing import Any
+
+from fastapi import FastAPI
+
+from src.core.interfaces.session_service_interface import ISessionService
+from src.core.interfaces.wire_capture_interface import IWireCapture
+
+logger = logging.getLogger(__name__)
+
+
+class AppLifecycle:
+    """Handles application lifecycle events.
+
+    This class manages startup and shutdown tasks for the application.
+    """
+
+    def __init__(self, app: FastAPI, config: dict[str, Any]) -> None:
+        """Initialize the lifecycle manager.
+
+        Args:
+            app: The FastAPI application
+            config: The application configuration
+        """
+        self.app = app
+        self.config = config
+        self._background_tasks: list[asyncio.Task] = []
+
+    async def startup(self) -> None:
+        """Perform startup tasks.
+
+        This method is called during application startup.
+        """
+        logger.info("Starting application lifecycle...")
+
+        # Start background tasks
+        self._start_background_tasks()
+
+    async def shutdown(self) -> None:
+        """Perform shutdown tasks.
+
+        This method is called during application shutdown.
+        """
+        logger.info("Shutting down application lifecycle...")
+
+        # Stop background tasks
+        await self._stop_background_tasks()
+
+        # Close any remaining connections
+        await self._close_connections()
+
+    def _start_background_tasks(self) -> None:
+        """Start background tasks."""
+        # Start session cleanup task
+        if self.config.get("session_cleanup_enabled", False):
+            interval = self.config.get(
+                "session_cleanup_interval", 3600
+            )  # 1 hour default
+            max_age = self.config.get("session_max_age", 86400)  # 1 day default
+
+            task = asyncio.create_task(
+                self._session_cleanup_task(interval, max_age), name="session_cleanup"
+            )
+            self._background_tasks.append(task)
+            logger.info(
+                f"Started session cleanup task (interval: {interval}s, max age: {max_age}s)"
+            )
+
+    async def _stop_background_tasks(self) -> None:
+        """Stop background tasks."""
+        for task in self._background_tasks:
+            if not task.done():
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    logger.info(f"Cancelled background task: {task.get_name()}")
+
+    async def _close_connections(self) -> None:
+        """Close any remaining connections."""
+        # Get service provider
+        provider = getattr(self.app.state, "service_provider", None)
+        if not provider:
+            return
+
+        # Get wire capture service and shut it down
+        wire_capture_service = provider.get_service(IWireCapture)
+        if wire_capture_service and hasattr(wire_capture_service, "shutdown"):
+            await wire_capture_service.shutdown()
+
+    async def _session_cleanup_task(self, interval: int, max_age: int) -> None:
+        """Background task for cleaning up expired sessions.
+
+        Args:
+            interval: The interval in seconds between cleanup runs
+            max_age: The maximum age in seconds for sessions
+        """
+        try:
+            while True:
+                await asyncio.sleep(interval)
+
+                try:
+                    # Get service provider
+                    provider = self.app.state.service_provider
+                    if not provider:
+                        logger.warning(
+                            "Service provider not available for session cleanup"
+                        )
+                        continue
+
+                    # Get session service
+                    session_service = provider.get_service(ISessionService)
+                    if not session_service:
+                        logger.warning("Session service not available for cleanup")
+                        continue
+
+                    # Perform cleanup
+                    deleted_count = 0
+                    with suppress(AttributeError):
+                        deleted_count = await session_service.cleanup_expired(max_age)
+
+                    if deleted_count > 0 and logger.isEnabledFor(logging.INFO):
+                        logger.info(f"Cleaned up {deleted_count} expired sessions")
+
+                except Exception as e:
+                    logger.error(f"Error during session cleanup: {e!s}")
+
+        except asyncio.CancelledError:
+            logger.debug("Session cleanup task cancelled")
+            raise

src/core/services/buffered_wire_capture_service.py

@@ -856,9 +856,19 @@ def force_shutdown_sync(self) -> None:
         if self._flush_task and not self._flush_task.done():
             with contextlib.suppress(Exception):
                 task = self._flush_task
+                # Suppress the 'task was destroyed but it is pending!' message
                 if hasattr(task, "_log_destroy_pending"):
                     cast(Any, task)._log_destroy_pending = False
                 loop = task.get_loop()
-                if not loop.is_closed():
-                    task.cancel()
+                if loop.is_running() and not loop.is_closed():
+                    loop.call_soon_threadsafe(task.cancel)
+                elif not loop.is_closed():
+                    # This is a fallback and might not always work if the loop is closing
+                    loop.run_until_complete(self.shutdown())
+
         self._flush_task = None
+
+    def __del__(self) -> None:
+        """Ensure cleanup is attempted on garbage collection."""
+        if self.enabled():
+            self.force_shutdown_sync()

src/loop_detection/hybrid_detector.py

@@ -19,7 +19,7 @@
     LoopDetectionResult,
 )
 from src.loop_detection.event import LoopDetectionEvent
-from src.loop_detection.gemini_cli_detector import GeminiCliLoopDetector
+from src.loop_detection.token_window_loop_detector import TokenWindowLoopDetector
 
 logger = logging.getLogger(__name__)
 
@@ -208,7 +208,7 @@ def __init__(
         """
         # Initialize short pattern detector (gemini-cli)
         short_config = short_detector_config or {}
-        self.short_detector = GeminiCliLoopDetector(
+        self.short_detector = TokenWindowLoopDetector(
             content_loop_threshold=short_config.get("content_loop_threshold", 10),
             content_chunk_size=short_config.get("content_chunk_size", 50),
             max_history_length=short_config.get("max_history_length", 1000),

src/loop_detection/gemini_cli_detector.py renamed to src/loop_detection/token_window_loop_detector.py

@@ -1,5 +1,7 @@
 """
-Loop detection implementation ported from Google's gemini-cli.
+Sliding-window loop detector originally ported from the Gemini CLI.
+
+This detector is backend-agnostic and can be used for any streaming LLM output.
 
 This is a direct port of the loop detection algorithm from:
 https://github.com/google/generative-ai-docs/tree/main/gemini-cli
@@ -45,9 +47,9 @@
 )
 
 
-class GeminiCliLoopDetector(ILoopDetector):
+class TokenWindowLoopDetector(ILoopDetector):
     """
-    Loop detector ported from Google's gemini-cli.
+    Backend-agnostic loop detector using a sliding token window with markdown-aware resets.
 
     This implementation uses a sliding window approach with intelligent context tracking
     to detect repetitive patterns in LLM responses without manual parameter tuning.
@@ -83,7 +85,7 @@ def __init__(
         self._loop_events: list[LoopDetectionEvent] = []
 
         logger.debug(
-            "GeminiCliLoopDetector initialized: chunk_size=%d, threshold=%d, max_history=%d",
+            "TokenWindowLoopDetector initialized: chunk_size=%d, threshold=%d, max_history=%d",
             self.content_chunk_size,
             self.content_loop_threshold,
             self.max_history_length,