fix: prevent OOM, orphaned agents, and unbounded growth during overnight builds (#1813)
* fix(stability): prevent OOM, orphaned agents, and unbounded growth during overnight builds Address multiple crash/stability issues observed during long-running autonomous builds: Backend: - Skip stuck subtasks in get_next_subtask() using attempt_history.json - Add retry with exponential backoff + jitter for LadybugDB lock contention - Time-window filter attempt counts (2h window) to prevent unbounded accumulation - Trim attempt history per subtask (cap at 50) to bound file size - Use timezone-aware UTC datetimes throughout recovery manager Frontend: - Kill all agents on window close to prevent orphaned processes - Circuit breaker: kill agents after 10 consecutive renderer disposal errors - Cap batch queue logs at 100 entries (OOM prevention in IPC batching) - Cap task log entries at 5000 per task (OOM prevention in store) Tests: - Add lock retry logic tests (lock detection, backoff, retry exhaustion) - Add stuck subtask skipping tests (skip, corrupt JSON, all-stuck) - Add time-window filtering and attempt trimming tests Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix(stability): address PR review findings for crash stability - Add .catch() to async killAll() calls to prevent unhandled promise rejections (index.ts on window close, utils.ts circuit breaker) - Reset circuitBreakerTriggered on successful send so it can re-trigger after renderer recovery followed by a second crash - Fix agentManagerRef type to reflect async killAll() signature - Switch %-format logging to f-strings for consistency with codebase convention Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Andy
@@ -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"
|
||||
|
||||
@@ -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}")
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -12,6 +12,17 @@ import type { BrowserWindow } from "electron";
|
||||
const warnTimestamps = new Map<string, number>();
|
||||
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<void> } | null = null;
|
||||
let circuitBreakerTriggered = false;
|
||||
|
||||
/** Set agent manager reference for circuit breaker cleanup */
|
||||
export function setAgentManagerRef(manager: { killAll: () => void | Promise<void> }): 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);
|
||||
|
||||
@@ -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, {
|
||||
|
||||
@@ -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<TaskState>((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<TaskState>((set, get) => ({
|
||||
return {
|
||||
tasks: updateTaskAtIndex(state.tasks, index, (t) => ({
|
||||
...t,
|
||||
logs: [...(t.logs || []), ...logs]
|
||||
logs: [...(t.logs || []), ...logs].slice(-MAX_LOG_ENTRIES)
|
||||
}))
|
||||
};
|
||||
});
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user