diff --git a/apps/backend/core/progress.py b/apps/backend/core/progress.py index 1bbe647a..ffe74da7 100644 --- a/apps/backend/core/progress.py +++ b/apps/backend/core/progress.py @@ -9,8 +9,11 @@ Enhanced with colored output, icons, and better visual formatting. """ import json +import logging from pathlib import Path +logger = logging.getLogger(__name__) + from core.plan_normalization import normalize_subtask_aliases from ui import ( Icons, @@ -230,8 +233,8 @@ def print_progress_summary(spec_dir: Path, show_next: bool = True) -> None: f" {icon(Icons.ARROW_RIGHT)} Next: {highlight(next_id)} - {next_desc}" ) - except (OSError, json.JSONDecodeError, UnicodeDecodeError): - pass # Ignore corrupted/unreadable progress files + except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e: + logger.debug(f"Failed to load plan file for phase summary: {e}") else: print() print_status("No implementation subtasks yet - planner needs to run", "pending") @@ -404,6 +407,8 @@ def get_next_subtask(spec_dir: Path) -> dict | None: """ Find the next subtask to work on, respecting phase dependencies. + Skips subtasks that are marked as stuck in the recovery manager's attempt history. + Args: spec_dir: Directory containing implementation_plan.json @@ -415,6 +420,23 @@ def get_next_subtask(spec_dir: Path) -> dict | None: if not plan_file.exists(): return None + # Load stuck subtasks from recovery manager's attempt history + stuck_subtask_ids = set() + attempt_history_file = spec_dir / "memory" / "attempt_history.json" + if attempt_history_file.exists(): + try: + with open(attempt_history_file, encoding="utf-8") as f: + attempt_history = json.load(f) + # Collect IDs of subtasks marked as stuck + stuck_subtask_ids = { + entry["subtask_id"] + for entry in attempt_history.get("stuck_subtasks", []) + if "subtask_id" in entry + } + except (OSError, json.JSONDecodeError, UnicodeDecodeError): + # If we can't read the file, continue without stuck checking + pass + try: with open(plan_file, encoding="utf-8") as f: plan = json.load(f) @@ -455,9 +477,15 @@ def get_next_subtask(spec_dir: Path) -> dict | None: if not deps_satisfied: continue - # Find first pending subtask in this phase + # Find first pending subtask in this phase (skip stuck subtasks) for subtask in phase.get("subtasks", phase.get("chunks", [])): status = subtask.get("status", "pending") + subtask_id = subtask.get("id") + + # Skip stuck subtasks + if subtask_id in stuck_subtask_ids: + continue + if status in {"pending", "not_started", "not started"}: subtask_out, _changed = normalize_subtask_aliases(subtask) subtask_out["status"] = "pending" diff --git a/apps/backend/integrations/graphiti/queries_pkg/client.py b/apps/backend/integrations/graphiti/queries_pkg/client.py index 3852e05b..e362ee98 100644 --- a/apps/backend/integrations/graphiti/queries_pkg/client.py +++ b/apps/backend/integrations/graphiti/queries_pkg/client.py @@ -5,7 +5,9 @@ Handles database connection, initialization, and lifecycle management. Uses LadybugDB as the embedded graph database (no Docker required, Python 3.12+). """ +import asyncio import logging +import random import sys from datetime import datetime, timezone @@ -14,6 +16,27 @@ from graphiti_config import GraphitiConfig, GraphitiState logger = logging.getLogger(__name__) +# Retry configuration for LadybugDB lock contention +MAX_LOCK_RETRIES = 5 +INITIAL_BACKOFF_SECONDS = 0.5 +MAX_BACKOFF_SECONDS = 8.0 +JITTER_PERCENT = 0.2 + + +def _is_lock_error(error: Exception) -> bool: + """Check if an error indicates database lock contention.""" + error_msg = str(error).lower() + return "could not set lock" in error_msg or ( + "lock" in error_msg and ("file" in error_msg or "database" in error_msg) + ) + + +def _backoff_with_jitter(attempt: int) -> float: + """Calculate exponential backoff with jitter for retry delays.""" + backoff = min(INITIAL_BACKOFF_SECONDS * (2**attempt), MAX_BACKOFF_SECONDS) + jitter = backoff * JITTER_PERCENT * (2 * random.random() - 1) + return max(0.01, backoff + jitter) + def _apply_ladybug_monkeypatch() -> bool: """ @@ -196,32 +219,36 @@ class GraphitiClient: ) db_path = self.config.get_db_path() - try: - self._driver = create_patched_kuzu_driver(db=str(db_path)) - except (OSError, PermissionError) as e: - logger.warning( - f"Failed to initialize LadybugDB driver at {db_path}: {e}" - ) - capture_exception( - e, - error_type=type(e).__name__, - db_path=str(db_path), - llm_provider=self.config.llm_provider, - embedder_provider=self.config.embedder_provider, - ) - return False - except Exception as e: - logger.warning( - f"Unexpected error initializing LadybugDB driver at {db_path}: {e}" - ) - capture_exception( - e, - error_type=type(e).__name__, - db_path=str(db_path), - llm_provider=self.config.llm_provider, - embedder_provider=self.config.embedder_provider, - ) - return False + + # Retry with exponential backoff for lock contention + for attempt in range(MAX_LOCK_RETRIES + 1): + try: + self._driver = create_patched_kuzu_driver(db=str(db_path)) + if attempt > 0: + logger.info( + f"LadybugDB lock acquired after {attempt} retries" + ) + break # Success + except Exception as e: + if _is_lock_error(e) and attempt < MAX_LOCK_RETRIES: + wait_time = _backoff_with_jitter(attempt) + logger.debug( + f"LadybugDB lock contention (attempt {attempt + 1}/{MAX_LOCK_RETRIES}), retrying in {wait_time:.2f}s" + ) + await asyncio.sleep(wait_time) + continue + logger.warning( + f"Failed to initialize LadybugDB driver at {db_path}: {e}" + ) + capture_exception( + e, + error_type=type(e).__name__, + db_path=str(db_path), + llm_provider=self.config.llm_provider, + embedder_provider=self.config.embedder_provider, + ) + return False + logger.info(f"Initialized LadybugDB driver (patched) at: {db_path}") except ImportError as e: logger.warning(f"KuzuDriver not available: {e}") diff --git a/apps/backend/services/recovery.py b/apps/backend/services/recovery.py index 100e44f0..af9eb6f7 100644 --- a/apps/backend/services/recovery.py +++ b/apps/backend/services/recovery.py @@ -14,12 +14,19 @@ Key Features: """ import json +import logging import subprocess from dataclasses import dataclass -from datetime import datetime +from datetime import datetime, timedelta, timezone from enum import Enum from pathlib import Path +# Recovery manager configuration +ATTEMPT_WINDOW_SECONDS = 7200 # Only count attempts within last 2 hours +MAX_ATTEMPT_HISTORY_PER_SUBTASK = 50 # Cap stored attempts per subtask + +logger = logging.getLogger(__name__) + class FailureType(Enum): """Types of failures that can occur during autonomous builds.""" @@ -82,8 +89,8 @@ class RecoveryManager: "subtasks": {}, "stuck_subtasks": [], "metadata": { - "created_at": datetime.now().isoformat(), - "last_updated": datetime.now().isoformat(), + "created_at": datetime.now(timezone.utc).isoformat(), + "last_updated": datetime.now(timezone.utc).isoformat(), }, } with open(self.attempt_history_file, "w", encoding="utf-8") as f: @@ -95,8 +102,8 @@ class RecoveryManager: "commits": [], "last_good_commit": None, "metadata": { - "created_at": datetime.now().isoformat(), - "last_updated": datetime.now().isoformat(), + "created_at": datetime.now(timezone.utc).isoformat(), + "last_updated": datetime.now(timezone.utc).isoformat(), }, } with open(self.build_commits_file, "w", encoding="utf-8") as f: @@ -114,7 +121,7 @@ class RecoveryManager: def _save_attempt_history(self, data: dict) -> None: """Save attempt history to JSON file.""" - data["metadata"]["last_updated"] = datetime.now().isoformat() + data["metadata"]["last_updated"] = datetime.now(timezone.utc).isoformat() with open(self.attempt_history_file, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) @@ -130,7 +137,7 @@ class RecoveryManager: def _save_build_commits(self, data: dict) -> None: """Save build commits to JSON file.""" - data["metadata"]["last_updated"] = datetime.now().isoformat() + data["metadata"]["last_updated"] = datetime.now(timezone.utc).isoformat() with open(self.build_commits_file, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) @@ -185,17 +192,44 @@ class RecoveryManager: def get_attempt_count(self, subtask_id: str) -> int: """ - Get how many times this subtask has been attempted. + Get how many times this subtask has been attempted within the time window. + + Only counts attempts within ATTEMPT_WINDOW_SECONDS (default: 2 hours). + This prevents unbounded accumulation across crash/restart cycles. Args: subtask_id: ID of the subtask Returns: - Number of attempts + Number of attempts within the time window """ history = self._load_attempt_history() subtask_data = history["subtasks"].get(subtask_id, {}) - return len(subtask_data.get("attempts", [])) + attempts = subtask_data.get("attempts", []) + + # Calculate cutoff time for the window + cutoff_time = datetime.now(timezone.utc) - timedelta( + seconds=ATTEMPT_WINDOW_SECONDS + ) + # For backward compatibility with naive timestamps, also create naive cutoff + cutoff_time_naive = datetime.now() - timedelta(seconds=ATTEMPT_WINDOW_SECONDS) + + # Count only attempts within the time window + recent_count = 0 + for attempt in attempts: + try: + attempt_time = datetime.fromisoformat(attempt["timestamp"]) + # Use appropriate cutoff based on whether timestamp is naive or aware + cutoff = ( + cutoff_time_naive if attempt_time.tzinfo is None else cutoff_time + ) + if attempt_time >= cutoff: + recent_count += 1 + except (KeyError, ValueError): + # If timestamp is missing or invalid, count it (backward compatibility) + recent_count += 1 + + return recent_count def record_attempt( self, @@ -208,6 +242,8 @@ class RecoveryManager: """ Record an attempt at a subtask. + Automatically trims old attempts if the history exceeds MAX_ATTEMPT_HISTORY_PER_SUBTASK. + Args: subtask_id: ID of the subtask session: Session number @@ -224,13 +260,24 @@ class RecoveryManager: # Add the attempt attempt = { "session": session, - "timestamp": datetime.now().isoformat(), + "timestamp": datetime.now(timezone.utc).isoformat(), "approach": approach, "success": success, "error": error, } history["subtasks"][subtask_id]["attempts"].append(attempt) + # Hard cap: trim oldest attempts if we exceed the maximum + attempts = history["subtasks"][subtask_id]["attempts"] + if len(attempts) > MAX_ATTEMPT_HISTORY_PER_SUBTASK: + trimmed_count = len(attempts) - MAX_ATTEMPT_HISTORY_PER_SUBTASK + history["subtasks"][subtask_id]["attempts"] = attempts[ + -MAX_ATTEMPT_HISTORY_PER_SUBTASK: + ] + logger.debug( + f"Trimmed {trimmed_count} old attempts for subtask {subtask_id} (cap: {MAX_ATTEMPT_HISTORY_PER_SUBTASK})" + ) + # Update status if success: history["subtasks"][subtask_id]["status"] = "completed" @@ -405,7 +452,7 @@ class RecoveryManager: commit_record = { "hash": commit_hash, "subtask_id": subtask_id, - "timestamp": datetime.now().isoformat(), + "timestamp": datetime.now(timezone.utc).isoformat(), } commits["commits"].append(commit_record) @@ -450,7 +497,7 @@ class RecoveryManager: stuck_entry = { "subtask_id": subtask_id, "reason": reason, - "escalated_at": datetime.now().isoformat(), + "escalated_at": datetime.now(timezone.utc).isoformat(), "attempt_count": self.get_attempt_count(subtask_id), } diff --git a/apps/frontend/src/main/index.ts b/apps/frontend/src/main/index.ts index 4e438940..f98725d3 100644 --- a/apps/frontend/src/main/index.ts +++ b/apps/frontend/src/main/index.ts @@ -330,6 +330,10 @@ function createWindow(): void { // Clean up on close mainWindow.on('closed', () => { + // Kill all agents when window closes (prevents orphaned processes) + agentManager?.killAll?.()?.catch((err: unknown) => { + console.warn('[main] Error killing agents on window close:', err); + }); mainWindow = null; }); } diff --git a/apps/frontend/src/main/ipc-handlers/index.ts b/apps/frontend/src/main/ipc-handlers/index.ts index db04f7af..fdd7c5b7 100644 --- a/apps/frontend/src/main/ipc-handlers/index.ts +++ b/apps/frontend/src/main/ipc-handlers/index.ts @@ -35,6 +35,7 @@ import { registerProfileHandlers } from './profile-handlers'; import { registerScreenshotHandlers } from './screenshot-handlers'; import { registerTerminalWorktreeIpcHandlers } from './terminal'; import { notificationService } from '../notification-service'; +import { setAgentManagerRef } from './utils'; /** * Setup all IPC handlers across all domains @@ -53,6 +54,9 @@ export function setupIpcHandlers( // Initialize notification service notificationService.initialize(getMainWindow); + // Wire up agent manager for circuit breaker cleanup + setAgentManagerRef(agentManager); + // Project handlers (including Python environment setup) registerProjectHandlers(pythonEnvManager, agentManager, getMainWindow); diff --git a/apps/frontend/src/main/ipc-handlers/utils.ts b/apps/frontend/src/main/ipc-handlers/utils.ts index cb113da1..e33049bf 100644 --- a/apps/frontend/src/main/ipc-handlers/utils.ts +++ b/apps/frontend/src/main/ipc-handlers/utils.ts @@ -12,6 +12,17 @@ import type { BrowserWindow } from "electron"; const warnTimestamps = new Map(); const WARN_COOLDOWN_MS = 5000; // 5 seconds between warnings per channel +/** Circuit breaker: kill agents after consecutive renderer disposal errors */ +const MAX_CONSECUTIVE_DISPOSAL_ERRORS = 10; +let consecutiveDisposalErrors = 0; +let agentManagerRef: { killAll: () => void | Promise } | null = null; +let circuitBreakerTriggered = false; + +/** Set agent manager reference for circuit breaker cleanup */ +export function setAgentManagerRef(manager: { killAll: () => void | Promise }): void { + agentManagerRef = manager; +} + /** * Check if a channel is within the warning cooldown period. * @returns true if within cooldown (should skip warning), false if cooldown expired @@ -108,6 +119,9 @@ export function safeSendToRenderer( // All checks passed - safe to send mainWindow.webContents.send(channel, ...args); + // On successful send, reset circuit breaker state (allow re-trigger after recovery) + consecutiveDisposalErrors = 0; + circuitBreakerTriggered = false; return true; } catch (error) { // Catch any disposal errors that might occur between our checks and the actual send @@ -115,6 +129,16 @@ export function safeSendToRenderer( // Only log disposal errors once per channel to avoid log spam if (errorMessage.includes("disposed") || errorMessage.includes("destroyed")) { + // Circuit breaker: track consecutive disposal errors + consecutiveDisposalErrors++; + if (consecutiveDisposalErrors >= MAX_CONSECUTIVE_DISPOSAL_ERRORS && !circuitBreakerTriggered && agentManagerRef) { + circuitBreakerTriggered = true; + console.error('[safeSendToRenderer] Circuit breaker triggered: killing all agents after renderer death'); + Promise.resolve(agentManagerRef.killAll()).catch((err) => { + console.error('[safeSendToRenderer] Error killing agents:', err); + }); + } + if (!isWithinCooldown(channel)) { console.warn(`[safeSendToRenderer] Frame disposed, skipping send: ${channel}`); recordWarning(channel); diff --git a/apps/frontend/src/renderer/hooks/useIpc.ts b/apps/frontend/src/renderer/hooks/useIpc.ts index 14a71a45..1dab10bc 100644 --- a/apps/frontend/src/renderer/hooks/useIpc.ts +++ b/apps/frontend/src/renderer/hooks/useIpc.ts @@ -7,6 +7,9 @@ import { useAuthFailureStore } from '../stores/auth-failure-store'; import { useProjectStore } from '../stores/project-store'; import type { ImplementationPlan, TaskStatus, RoadmapGenerationStatus, Roadmap, ExecutionProgress, RateLimitInfo, SDKRateLimitInfo, AuthFailureInfo } from '../../shared/types'; +/** Maximum log entries to buffer in the batch queue between flushes (OOM prevention) */ +const MAX_BATCH_QUEUE_LOGS = 100; + /** * Batched update queue for IPC events. * Collects updates within a 16ms window (one frame) and flushes them together. @@ -124,6 +127,10 @@ function queueUpdate(taskId: string, update: BatchedUpdate): void { let mergedLogs = existing.logs; if (update.logs) { mergedLogs = [...(existing.logs || []), ...update.logs]; + // Cap batch queue logs to prevent OOM when logs arrive faster than flush interval + if (mergedLogs.length > MAX_BATCH_QUEUE_LOGS) { + mergedLogs = mergedLogs.slice(-MAX_BATCH_QUEUE_LOGS); + } } batchQueue.set(taskId, { diff --git a/apps/frontend/src/renderer/stores/task-store.ts b/apps/frontend/src/renderer/stores/task-store.ts index 6b28e88c..b0ef7206 100644 --- a/apps/frontend/src/renderer/stores/task-store.ts +++ b/apps/frontend/src/renderer/stores/task-store.ts @@ -7,6 +7,9 @@ import { useProjectStore } from './project-store'; /** Default max parallel tasks when no project setting is configured */ export const DEFAULT_MAX_PARALLEL_TASKS = 3; +/** Maximum log entries stored per task to prevent renderer OOM */ +export const MAX_LOG_ENTRIES = 5000; + interface TaskState { tasks: Task[]; selectedTaskId: string | null; @@ -475,7 +478,7 @@ export const useTaskStore = create((set, get) => ({ return { tasks: updateTaskAtIndex(state.tasks, index, (t) => ({ ...t, - logs: [...(t.logs || []), log] + logs: [...(t.logs || []), log].slice(-MAX_LOG_ENTRIES) })) }; }), @@ -508,7 +511,7 @@ export const useTaskStore = create((set, get) => ({ return { tasks: updateTaskAtIndex(state.tasks, index, (t) => ({ ...t, - logs: [...(t.logs || []), ...logs] + logs: [...(t.logs || []), ...logs].slice(-MAX_LOG_ENTRIES) })) }; }); diff --git a/tests/test_graphiti.py b/tests/test_graphiti.py index d753750d..396aca15 100644 --- a/tests/test_graphiti.py +++ b/tests/test_graphiti.py @@ -1,4 +1,5 @@ """Tests for Graphiti memory integration.""" +import asyncio import os import pytest from pathlib import Path @@ -529,3 +530,252 @@ class TestGraphitiState: assert len(state.error_log) == 10 assert "Error 5" in state.error_log[0]["error"] assert "Error 14" in state.error_log[-1]["error"] + + +# ============================================================================= +# LADYBUGDB LOCK RETRY LOGIC TESTS +# ============================================================================= + + +class TestIsLockError: + """Tests for _is_lock_error lock detection function.""" + + def test_lock_file_error_detected(self): + """Detects lock + file pattern in error messages.""" + from integrations.graphiti.queries_pkg.client import _is_lock_error + + assert _is_lock_error(Exception("Could not set lock on file")) is True + + def test_lock_database_error_detected(self): + """Detects lock + database pattern in error messages.""" + from integrations.graphiti.queries_pkg.client import _is_lock_error + + assert _is_lock_error(Exception("Database lock contention detected")) is True + + def test_could_not_set_lock_detected(self): + """Detects 'could not set lock' pattern.""" + from integrations.graphiti.queries_pkg.client import _is_lock_error + + assert _is_lock_error(Exception("could not set lock")) is True + + def test_non_lock_error_not_detected(self): + """Non-lock errors are not detected as lock errors.""" + from integrations.graphiti.queries_pkg.client import _is_lock_error + + assert _is_lock_error(Exception("Connection refused")) is False + assert _is_lock_error(Exception("Timeout error")) is False + assert _is_lock_error(Exception("Permission denied")) is False + + def test_lock_without_file_or_database_not_detected(self): + """'lock' alone without 'file' or 'database' is not detected.""" + from integrations.graphiti.queries_pkg.client import _is_lock_error + + # 'lock' without 'file' or 'database' and no 'could not set lock' + assert _is_lock_error(Exception("Object is locked by user")) is False + + +class TestBackoffWithJitter: + """Tests for _backoff_with_jitter calculation.""" + + def test_backoff_increases_with_attempt(self): + """Backoff time increases with attempt number.""" + from integrations.graphiti.queries_pkg.client import _backoff_with_jitter + + # Run multiple times to account for jitter + attempt_0_values = [_backoff_with_jitter(0) for _ in range(20)] + attempt_3_values = [_backoff_with_jitter(3) for _ in range(20)] + + avg_0 = sum(attempt_0_values) / len(attempt_0_values) + avg_3 = sum(attempt_3_values) / len(attempt_3_values) + + assert avg_3 > avg_0, "Higher attempts should have higher average backoff" + + def test_backoff_is_positive(self): + """Backoff is always positive.""" + from integrations.graphiti.queries_pkg.client import _backoff_with_jitter + + for attempt in range(10): + for _ in range(10): + assert _backoff_with_jitter(attempt) > 0 + + def test_backoff_capped_at_max(self): + """Backoff should not exceed MAX_BACKOFF_SECONDS + jitter.""" + from integrations.graphiti.queries_pkg.client import ( + JITTER_PERCENT, + MAX_BACKOFF_SECONDS, + _backoff_with_jitter, + ) + + max_possible = MAX_BACKOFF_SECONDS * (1 + JITTER_PERCENT) + for _ in range(50): + val = _backoff_with_jitter(100) # Very high attempt + assert val <= max_possible + 0.01, f"Backoff {val} exceeded max {max_possible}" + + +class TestGraphitiClientRetryLogic: + """Tests for LadybugDB lock retry logic in GraphitiClient.initialize(). + + These tests exercise the retry loop behavior by mocking the modules + that are imported locally inside initialize(). We patch at the source + module level since the imports are local to the method. + """ + + def _make_config(self): + """Create a mock GraphitiConfig for testing.""" + config = MagicMock() + config.llm_provider = "openai" + config.embedder_provider = "openai" + config.get_db_path.return_value = Path("/tmp/test-db") + config.get_provider_summary.return_value = "openai/openai" + return config + + def _make_mock_providers(self): + """Create mock graphiti_providers module.""" + mock_providers = MagicMock() + mock_providers.create_llm_client = MagicMock(return_value=MagicMock()) + mock_providers.create_embedder = MagicMock(return_value=MagicMock()) + mock_providers.ProviderError = type("ProviderError", (Exception,), {}) + mock_providers.ProviderNotInstalled = type( + "ProviderNotInstalled", (mock_providers.ProviderError,), {} + ) + return mock_providers + + def _make_noop_sleep(self): + """Create an async no-op replacement for asyncio.sleep.""" + async def _noop_sleep(_delay): + return + + return _noop_sleep + + @pytest.mark.asyncio + async def test_successful_retry_after_lock_error(self): + """Client retries and succeeds after transient lock error.""" + from integrations.graphiti.queries_pkg.client import GraphitiClient + + config = self._make_config() + client = GraphitiClient(config) + + call_count = 0 + + def mock_create_driver(db=""): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise OSError("Could not set lock on file /tmp/test-db") + return MagicMock() + + mock_graphiti_instance = MagicMock() + + async def mock_build_indices(): + pass + + mock_graphiti_instance.build_indices_and_constraints = mock_build_indices + + mock_graphiti_cls = MagicMock(return_value=mock_graphiti_instance) + mock_graphiti_core = MagicMock() + mock_graphiti_core.Graphiti = mock_graphiti_cls + + mock_kuzu_driver = MagicMock() + mock_kuzu_driver.create_patched_kuzu_driver = mock_create_driver + + with ( + patch.dict(sys.modules, { + "graphiti_core": mock_graphiti_core, + "graphiti_providers": self._make_mock_providers(), + "integrations.graphiti.queries_pkg.kuzu_driver_patched": mock_kuzu_driver, + }), + patch( + "integrations.graphiti.queries_pkg.client._apply_ladybug_monkeypatch", + return_value=True, + ), + patch( + "integrations.graphiti.queries_pkg.client.asyncio.sleep", + side_effect=self._make_noop_sleep(), + ), + ): + result = await client.initialize() + + assert call_count == 2, "Should have retried once after lock error" + assert result is True, "Should succeed after retry" + + @pytest.mark.asyncio + async def test_exhausted_retries_returns_false(self): + """Client returns False after exhausting all retries on lock errors.""" + from integrations.graphiti.queries_pkg.client import ( + MAX_LOCK_RETRIES, + GraphitiClient, + ) + + config = self._make_config() + client = GraphitiClient(config) + + call_count = 0 + + def always_lock_error(db=""): + nonlocal call_count + call_count += 1 + raise OSError("Could not set lock on database file") + + mock_kuzu_driver = MagicMock() + mock_kuzu_driver.create_patched_kuzu_driver = always_lock_error + + with ( + patch.dict(sys.modules, { + "graphiti_core": MagicMock(), + "graphiti_providers": self._make_mock_providers(), + "integrations.graphiti.queries_pkg.kuzu_driver_patched": mock_kuzu_driver, + }), + patch( + "integrations.graphiti.queries_pkg.client._apply_ladybug_monkeypatch", + return_value=True, + ), + patch( + "integrations.graphiti.queries_pkg.client.capture_exception", + ), + patch( + "integrations.graphiti.queries_pkg.client.asyncio.sleep", + side_effect=self._make_noop_sleep(), + ), + ): + result = await client.initialize() + + assert result is False, "Should return False after exhausting retries" + # Should attempt MAX_LOCK_RETRIES + 1 times (initial + retries) + assert call_count == MAX_LOCK_RETRIES + 1 + + @pytest.mark.asyncio + async def test_non_lock_error_fails_immediately(self): + """Non-lock errors cause immediate failure without retry.""" + from integrations.graphiti.queries_pkg.client import GraphitiClient + + config = self._make_config() + client = GraphitiClient(config) + + call_count = 0 + + def connection_error(db=""): + nonlocal call_count + call_count += 1 + raise RuntimeError("Connection refused - server not running") + + mock_kuzu_driver = MagicMock() + mock_kuzu_driver.create_patched_kuzu_driver = connection_error + + with ( + patch.dict(sys.modules, { + "graphiti_core": MagicMock(), + "graphiti_providers": self._make_mock_providers(), + "integrations.graphiti.queries_pkg.kuzu_driver_patched": mock_kuzu_driver, + }), + patch( + "integrations.graphiti.queries_pkg.client._apply_ladybug_monkeypatch", + return_value=True, + ), + patch( + "integrations.graphiti.queries_pkg.client.capture_exception", + ), + ): + result = await client.initialize() + + assert call_count == 1, "Non-lock errors should not trigger retries" + assert result is False diff --git a/tests/test_implementation_plan.py b/tests/test_implementation_plan.py index e57963be..a059642d 100644 --- a/tests/test_implementation_plan.py +++ b/tests/test_implementation_plan.py @@ -1630,3 +1630,144 @@ class TestEdgeCaseStateTransitions: # All subtasks blocked = effectively pending state = backlog assert plan.status == "backlog" assert plan.planStatus == "pending" + + +# ============================================================================= +# STUCK SUBTASK SKIPPING TESTS (progress.py get_next_subtask) +# ============================================================================= + +class TestStuckSubtaskSkipping: + """Tests for stuck subtask skipping in progress.get_next_subtask().""" + + def _make_plan(self, subtasks): + """Helper to create a minimal implementation_plan.json dict.""" + return { + "feature": "Test", + "workflow_type": "feature", + "phases": [ + { + "phase": 1, + "name": "Phase 1", + "depends_on": [], + "subtasks": subtasks, + } + ], + } + + def _make_attempt_history(self, stuck_ids): + """Helper to create attempt_history.json with stuck subtasks.""" + return { + "subtasks": {}, + "stuck_subtasks": [ + {"subtask_id": sid, "reason": "stuck", "escalated_at": "2024-01-01T00:00:00"} + for sid in stuck_ids + ], + "metadata": {"created_at": "2024-01-01T00:00:00", "last_updated": "2024-01-01T00:00:00"}, + } + + def test_stuck_subtask_is_skipped(self, temp_dir): + """Stuck subtasks are skipped when selecting the next subtask.""" + from progress import get_next_subtask + + spec_dir = temp_dir / "spec" + spec_dir.mkdir(parents=True) + + # Create plan with two pending subtasks + plan = self._make_plan([ + {"id": "stuck-1", "description": "Stuck task", "status": "pending"}, + {"id": "good-1", "description": "Normal task", "status": "pending"}, + ]) + (spec_dir / "implementation_plan.json").write_text(json.dumps(plan)) + + # Mark stuck-1 as stuck + memory_dir = spec_dir / "memory" + memory_dir.mkdir(parents=True) + history = self._make_attempt_history(["stuck-1"]) + (memory_dir / "attempt_history.json").write_text(json.dumps(history)) + + result = get_next_subtask(spec_dir) + assert result is not None + assert result["id"] == "good-1", "Should skip stuck-1 and select good-1" + + def test_normal_subtask_selected_when_stuck_exist(self, temp_dir): + """Normal pending subtasks are selected even when stuck ones exist.""" + from progress import get_next_subtask + + spec_dir = temp_dir / "spec" + spec_dir.mkdir(parents=True) + + plan = self._make_plan([ + {"id": "stuck-a", "description": "Stuck A", "status": "pending"}, + {"id": "stuck-b", "description": "Stuck B", "status": "pending"}, + {"id": "normal-c", "description": "Normal C", "status": "pending"}, + ]) + (spec_dir / "implementation_plan.json").write_text(json.dumps(plan)) + + memory_dir = spec_dir / "memory" + memory_dir.mkdir(parents=True) + history = self._make_attempt_history(["stuck-a", "stuck-b"]) + (memory_dir / "attempt_history.json").write_text(json.dumps(history)) + + result = get_next_subtask(spec_dir) + assert result is not None + assert result["id"] == "normal-c" + + def test_no_attempt_history_file(self, temp_dir): + """When attempt_history.json doesn't exist, normal selection proceeds.""" + from progress import get_next_subtask + + spec_dir = temp_dir / "spec" + spec_dir.mkdir(parents=True) + + plan = self._make_plan([ + {"id": "task-1", "description": "Task 1", "status": "pending"}, + ]) + (spec_dir / "implementation_plan.json").write_text(json.dumps(plan)) + + # No memory directory or attempt_history.json + + result = get_next_subtask(spec_dir) + assert result is not None + assert result["id"] == "task-1" + + def test_corrupted_attempt_history_json(self, temp_dir): + """When attempt_history.json is corrupted, normal selection proceeds.""" + from progress import get_next_subtask + + spec_dir = temp_dir / "spec" + spec_dir.mkdir(parents=True) + + plan = self._make_plan([ + {"id": "task-1", "description": "Task 1", "status": "pending"}, + ]) + (spec_dir / "implementation_plan.json").write_text(json.dumps(plan)) + + memory_dir = spec_dir / "memory" + memory_dir.mkdir(parents=True) + (memory_dir / "attempt_history.json").write_text("{invalid json!!!") + + result = get_next_subtask(spec_dir) + assert result is not None + assert result["id"] == "task-1", "Should still select task when JSON is corrupted" + + def test_all_pending_subtasks_stuck_returns_none(self, temp_dir): + """When ALL pending subtasks are stuck, returns None.""" + from progress import get_next_subtask + + spec_dir = temp_dir / "spec" + spec_dir.mkdir(parents=True) + + plan = self._make_plan([ + {"id": "stuck-1", "description": "Stuck 1", "status": "pending"}, + {"id": "stuck-2", "description": "Stuck 2", "status": "pending"}, + {"id": "done-1", "description": "Done 1", "status": "completed"}, + ]) + (spec_dir / "implementation_plan.json").write_text(json.dumps(plan)) + + memory_dir = spec_dir / "memory" + memory_dir.mkdir(parents=True) + history = self._make_attempt_history(["stuck-1", "stuck-2"]) + (memory_dir / "attempt_history.json").write_text(json.dumps(history)) + + result = get_next_subtask(spec_dir) + assert result is None, "Should return None when all pending subtasks are stuck" diff --git a/tests/test_recovery.py b/tests/test_recovery.py index b147dc95..a3c61ed9 100755 --- a/tests/test_recovery.py +++ b/tests/test_recovery.py @@ -13,6 +13,7 @@ Tests the recovery system functionality including: import json import subprocess import sys +from datetime import datetime from pathlib import Path import pytest @@ -533,6 +534,322 @@ def test_checkpoint_clear_and_reset(test_env): assert manager2.get_attempt_count("subtask-1") == 2, "subtask-1 history lost" +# ============================================================================= +# TIME-WINDOW FILTERING TESTS (get_attempt_count) +# ============================================================================= + +def test_get_attempt_count_time_window_filtering(test_env): + """Test that get_attempt_count only counts attempts within the 2-hour window.""" + from datetime import timedelta + + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + old_time = (datetime.now() - timedelta(hours=3)).isoformat() + recent_time = (datetime.now() - timedelta(minutes=30)).isoformat() + + history = manager._load_attempt_history() + history["subtasks"]["test-1"] = { + "attempts": [ + {"timestamp": old_time, "approach": "old approach", "success": False}, + {"timestamp": recent_time, "approach": "recent approach", "success": False}, + ], + "status": "failed", + } + manager._save_attempt_history(history) + + count = manager.get_attempt_count("test-1") + assert count == 1, "Should only count the recent attempt within 2-hour window" + + +def test_get_attempt_count_boundary_just_inside_and_outside(test_env): + """Test attempts just inside and outside the 2-hour cutoff boundary.""" + from datetime import timedelta + + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + # 1 second inside the window (1h 59m 59s ago) - should be included + inside_time = (datetime.now() - timedelta(seconds=7199)).isoformat() + # 10 seconds outside the window (2h 10s ago) - should be excluded + outside_time = (datetime.now() - timedelta(seconds=7210)).isoformat() + + history = manager._load_attempt_history() + history["subtasks"]["test-boundary"] = { + "attempts": [ + {"timestamp": inside_time, "approach": "inside window", "success": False}, + {"timestamp": outside_time, "approach": "outside window", "success": False}, + ], + "status": "failed", + } + manager._save_attempt_history(history) + + count = manager.get_attempt_count("test-boundary") + assert count == 1, "Attempt inside window should be counted, outside should not" + + +def test_get_attempt_count_all_outside_window(test_env): + """Test that all attempts outside the time window returns 0.""" + from datetime import timedelta + + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + old_time_1 = (datetime.now() - timedelta(hours=5)).isoformat() + old_time_2 = (datetime.now() - timedelta(hours=4)).isoformat() + old_time_3 = (datetime.now() - timedelta(hours=3)).isoformat() + + history = manager._load_attempt_history() + history["subtasks"]["test-old"] = { + "attempts": [ + {"timestamp": old_time_1, "approach": "old 1", "success": False}, + {"timestamp": old_time_2, "approach": "old 2", "success": False}, + {"timestamp": old_time_3, "approach": "old 3", "success": False}, + ], + "status": "failed", + } + manager._save_attempt_history(history) + + count = manager.get_attempt_count("test-old") + assert count == 0, "All attempts outside window should result in count of 0" + + +def test_get_attempt_count_all_recent(test_env): + """Test that all recent attempts are counted.""" + from datetime import timedelta + + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + times = [ + (datetime.now() - timedelta(minutes=10)).isoformat(), + (datetime.now() - timedelta(minutes=30)).isoformat(), + (datetime.now() - timedelta(minutes=90)).isoformat(), + ] + + history = manager._load_attempt_history() + history["subtasks"]["test-recent"] = { + "attempts": [ + {"timestamp": times[0], "approach": "a1", "success": False}, + {"timestamp": times[1], "approach": "a2", "success": False}, + {"timestamp": times[2], "approach": "a3", "success": False}, + ], + "status": "failed", + } + manager._save_attempt_history(history) + + count = manager.get_attempt_count("test-recent") + assert count == 3, "All recent attempts should be counted" + + +def test_get_attempt_count_missing_timestamp_backward_compat(test_env): + """Test backward compatibility: attempts without timestamps are counted as recent.""" + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + history = manager._load_attempt_history() + history["subtasks"]["test-no-ts"] = { + "attempts": [ + {"approach": "no timestamp", "success": False}, + {"approach": "also no timestamp", "success": False}, + ], + "status": "failed", + } + manager._save_attempt_history(history) + + count = manager.get_attempt_count("test-no-ts") + assert count == 2, "Attempts without timestamps should be counted (backward compat)" + + +def test_get_attempt_count_invalid_timestamp_backward_compat(test_env): + """Test backward compatibility: attempts with invalid timestamps are counted as recent.""" + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + history = manager._load_attempt_history() + history["subtasks"]["test-bad-ts"] = { + "attempts": [ + {"timestamp": "not-a-date", "approach": "bad ts", "success": False}, + {"timestamp": "2024-13-99T99:99:99", "approach": "invalid ts", "success": False}, + ], + "status": "failed", + } + manager._save_attempt_history(history) + + count = manager.get_attempt_count("test-bad-ts") + assert count == 2, "Attempts with invalid timestamps should be counted (backward compat)" + + +def test_get_attempt_count_mixed_timestamps(test_env): + """Test mixed scenario: some attempts with timestamps, some without.""" + from datetime import timedelta + + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + old_time = (datetime.now() - timedelta(hours=5)).isoformat() + recent_time = (datetime.now() - timedelta(minutes=10)).isoformat() + + history = manager._load_attempt_history() + history["subtasks"]["test-mixed"] = { + "attempts": [ + {"timestamp": old_time, "approach": "old", "success": False}, + {"timestamp": recent_time, "approach": "recent", "success": False}, + {"approach": "no timestamp", "success": False}, + {"timestamp": "garbage", "approach": "bad timestamp", "success": False}, + ], + "status": "failed", + } + manager._save_attempt_history(history) + + # old_time: excluded (outside window) + # recent_time: included (within window) + # no timestamp: included (backward compat) + # bad timestamp: included (backward compat) + count = manager.get_attempt_count("test-mixed") + assert count == 3, "Should count recent + missing/invalid timestamps, exclude old" + + +# ============================================================================= +# ATTEMPT HISTORY TRIMMING TESTS (record_attempt) +# ============================================================================= + +def test_record_attempt_trimming_at_51(test_env): + """Test that recording the 51st attempt triggers trimming to 50.""" + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + # Manually inject 50 attempts + history = manager._load_attempt_history() + history["subtasks"]["trim-test"] = { + "attempts": [ + { + "session": i, + "timestamp": datetime.now().isoformat(), + "approach": f"approach-{i}", + "success": False, + "error": None, + } + for i in range(50) + ], + "status": "failed", + } + manager._save_attempt_history(history) + + # Record the 51st attempt + manager.record_attempt("trim-test", 51, False, "approach-50", "error") + + history = manager._load_attempt_history() + attempts = history["subtasks"]["trim-test"]["attempts"] + assert len(attempts) == 50, "Should trim to 50 after exceeding cap" + + +def test_record_attempt_trimming_keeps_newest(test_env): + """Test that trimming keeps the newest 50 attempts, not the oldest.""" + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + # Inject 50 attempts with identifiable approaches + history = manager._load_attempt_history() + history["subtasks"]["trim-order"] = { + "attempts": [ + { + "session": i, + "timestamp": datetime.now().isoformat(), + "approach": f"old-approach-{i}", + "success": False, + "error": None, + } + for i in range(50) + ], + "status": "failed", + } + manager._save_attempt_history(history) + + # Record new attempt (triggers trim) + manager.record_attempt("trim-order", 99, False, "newest-approach", "error") + + history = manager._load_attempt_history() + attempts = history["subtasks"]["trim-order"]["attempts"] + assert len(attempts) == 50 + + # The oldest attempt (old-approach-0) should be gone + approaches = [a["approach"] for a in attempts] + assert "old-approach-0" not in approaches, "Oldest attempt should be trimmed" + # The newest attempt should be present + assert "newest-approach" in approaches, "Newest attempt should be kept" + # old-approach-1 should be the oldest remaining + assert "old-approach-1" in approaches, "Second oldest should now be first" + + +def test_record_attempt_no_trimming_at_exactly_50(test_env): + """Test that exactly 50 attempts does not trigger trimming.""" + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + # Inject 49 attempts + history = manager._load_attempt_history() + history["subtasks"]["no-trim"] = { + "attempts": [ + { + "session": i, + "timestamp": datetime.now().isoformat(), + "approach": f"approach-{i}", + "success": False, + "error": None, + } + for i in range(49) + ], + "status": "failed", + } + manager._save_attempt_history(history) + + # Record the 50th attempt (should NOT trigger trimming) + manager.record_attempt("no-trim", 50, False, "approach-49", "error") + + history = manager._load_attempt_history() + attempts = history["subtasks"]["no-trim"]["attempts"] + assert len(attempts) == 50, "Exactly 50 should not trigger trimming" + # First attempt should still be present + assert attempts[0]["approach"] == "approach-0", "No attempts should be removed" + + +def test_record_attempt_trimming_from_100(test_env): + """Test trimming from 100 attempts keeps exactly 50.""" + temp_dir, spec_dir, project_dir = test_env + manager = RecoveryManager(spec_dir, project_dir) + + # Inject 100 attempts + history = manager._load_attempt_history() + history["subtasks"]["big-trim"] = { + "attempts": [ + { + "session": i, + "timestamp": datetime.now().isoformat(), + "approach": f"approach-{i}", + "success": False, + "error": None, + } + for i in range(100) + ], + "status": "failed", + } + manager._save_attempt_history(history) + + # Record attempt 101 (triggers trim from 101 -> 50) + manager.record_attempt("big-trim", 101, False, "approach-100", "error") + + history = manager._load_attempt_history() + attempts = history["subtasks"]["big-trim"]["attempts"] + assert len(attempts) == 50, "Should trim to exactly 50" + + # Verify newest are kept + approaches = [a["approach"] for a in attempts] + assert "approach-100" in approaches, "Newest attempt should be kept" + assert "approach-0" not in approaches, "Oldest attempts should be trimmed" + assert "approach-50" not in approaches, "Mid-range old attempts should be trimmed" + + def run_all_tests(): """Run all tests.""" print("=" * 70)