From ce43b07905b0ef48e60e578c51790dcb5e72cae4 Mon Sep 17 00:00:00 2001 From: cherninkiy Date: Fri, 1 May 2026 01:34:29 +0500 Subject: [PATCH] feat: Context Repository and Tool Reflection Cycle - ContextPromptRepository: Prompt templates with variables, semantic search, 5 default templates, success tracking - Tool Reflection Cycle: 8 error categories, auto-fix suggestions, iterative improvement, automatic try-except wrapping - Tests: 51 total (21 context repository, 30 reflection) - Docs: Recommender MVP completed, roadmap moved to docs/ - Fixes: PR review issues, flake8 compliance, .gitignore cleanup - Refactor: Simplified agents/init.py --- .gitignore | 60 +- CHANGELOG.md | 19 + README.md | 2 +- core/agents/__init__.py | 1 + core/agents/recommender.py | 193 +++++ core/context_repository.py | 565 ++++++++++++++ core/metrics.py | 42 ++ core/reflection.py | 703 ++++++++++++++++++ .../HOLOBIONT_ROADMAP.md | 31 +- metrics.jsonl | 34 + tests/test_agent_recommender.py | 57 ++ tests/test_context_repository.py | 378 ++++++++++ tests/test_metrics.py | 37 + tests/test_reflection.py | 461 ++++++++++++ 14 files changed, 2525 insertions(+), 58 deletions(-) create mode 100644 core/agents/__init__.py create mode 100644 core/agents/recommender.py create mode 100644 core/context_repository.py create mode 100644 core/reflection.py rename HOLOBIONT_ROADMAP.md => docs/HOLOBIONT_ROADMAP.md (92%) create mode 100644 metrics.jsonl create mode 100644 tests/test_agent_recommender.py create mode 100644 tests/test_context_repository.py create mode 100644 tests/test_reflection.py diff --git a/.gitignore b/.gitignore index d3ea078..24ecbc5 100644 --- a/.gitignore +++ b/.gitignore @@ -1,12 +1,18 @@ # Logs and temporary files *.log *.tmp +*.swp # Environment files .env .env.local *.env.* +# Coverage reports +coverage/ +htmlcov/ +.coverage + # Python specific __pycache__/ *.pyc @@ -14,54 +20,16 @@ __pycache__/ *.pyd .Python *.so -.coverage -coverage/ -htmlcov/ -.pytest_cache/ +*.egg-info/ +.eggs/ .mypy_cache/ +.pytest_cache/ +.tox/ -# Build and distribution artifacts +# Build artifacts dist/ build/ -*.egg-info/ -.eggs/ - -# Dependencies -.venv/ -venv/ -.env/ -node_modules/ - -# Editors and IDEs -.vscode/ -.idea/ -*.swp -*.swo - -# OS generated files -.DS_Store -Thumbs.db +*.egg -# Compression and archives -*.zip -*.gz -*.tar -*.tgz -*.bz2 -*.xz -*.7z -*.rar -*.zst -*.lz4 -*.lzh -*.cab -*.arj -*.rpm -*.deb -*.Z -*.lz -*.lzo -*.tar.gz -*.tar.bz2 -*.tar.xz -*.tar.zst +# Original rule preserved +*.log \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index ec1ca1a..030c8e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,25 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm --- +## [Unreleased] + +### Added +- **Phase 1 intelligence modules** – added `core/context_repository.py` with `ContextPromptRepository`, `PromptTemplate`, and `SemanticIndex`; added `core/reflection.py` with `ErrorAnalyzer`, `CorrectionGenerator`, and `ReflectionLoop`. +- **Initial agent recommender module** – added `core/agents/recommender.py` with `TrajectoryEncoder`, `AgentRecommender`, and `CommitteeBuilder` to support Phase 1 recommendation workflows. +- **Test coverage for new modules** – added `tests/test_context_repository.py` and `tests/test_reflection.py`. + +### Changed +- **Context repository indexing flow** – `store_prompt()` now supports deferred index rebuilds via `rebuild_index=False`, and default template initialization performs a single rebuild after batch insert. +- **Metrics enrichment for recommendations** – `log_execution()` now supports optional `task_type`, and `core.metrics` includes normalized execution history helpers for recommendation training. + +### Fixed +- **Runtime correction wrapper** – fixed `_wrap_with_error_handling()` indentation in `core/reflection.py` so wrapped code is valid Python under `try`. +- **Error pattern matching efficiency** – error patterns are normalized once (lowercased) and matched without repeated per-iteration `.lower()` calls. +- **Timestamp import style** – replaced inline `__import__("datetime")` call with standard module-level `datetime` import. +- **Lint and ignore file cleanup** – removed markdown fence artifacts from `.gitignore`, removed unused locals/imports, and cleaned trailing-whitespace lines to keep CI (`flake8`) green. + +--- + --- ## [0.2.0] – 2026-04-21 diff --git a/README.md b/README.md index 700bdcb..63785d8 100644 --- a/README.md +++ b/README.md @@ -259,4 +259,4 @@ rawllm/ MIT — use the ideas freely, fork, and improve. -└── HOLOBIONT_ROADMAP.md # Development roadmap and future phases +└── docs/HOLOBIONT_ROADMAP.md # Development roadmap and future phases diff --git a/core/agents/__init__.py b/core/agents/__init__.py new file mode 100644 index 0000000..6333c9c --- /dev/null +++ b/core/agents/__init__.py @@ -0,0 +1 @@ +"""Specialized agent modules.""" diff --git a/core/agents/recommender.py b/core/agents/recommender.py new file mode 100644 index 0000000..56a439a --- /dev/null +++ b/core/agents/recommender.py @@ -0,0 +1,193 @@ +"""Lightweight agent recommendation based on trajectory history. + +This module intentionally starts with a compact heuristic model that mimics the +planned RecVAE API surface while remaining dependency-free. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +def _tokenize(text: str) -> set[str]: + return {part.strip(".,!?;:()[]{}\"'").lower() for part in text.split() if part.strip()} + + +@dataclass +class AgentPerformance: + """Aggregated performance for an agent and task category.""" + + agent_id: str + task_type: str + success_rate: float + avg_success_score: float + avg_latency_ms: float + sample_count: int + + +class TrajectoryEncoder: + """Simple trajectory encoder/decoder. + + Produces a compact latent vector: + [avg_success_score, avg_latency_norm, average_step_depth, event_count_norm]. + """ + + def encode_trajectory(self, events_list: list[dict[str, Any]]) -> list[float]: + if not events_list: + return [0.0, 0.0, 0.0, 0.0] + + count = len(events_list) + avg_success = sum(float(e.get("success_score", 0.0)) for e in events_list) / count + avg_latency = sum(float(e.get("execution_time_ms", 0.0)) for e in events_list) / count + avg_step = ( + sum(float(e.get("step_number", 0.0) or 0.0) for e in events_list) / count + ) + + return [ + max(0.0, min(1.0, avg_success)), + max(0.0, min(1.0, avg_latency / 10000.0)), + max(0.0, min(1.0, avg_step / 20.0)), + max(0.0, min(1.0, count / 20.0)), + ] + + def decode_vector(self, latent_vector: list[float]) -> dict[str, float]: + padded = (latent_vector + [0.0, 0.0, 0.0, 0.0])[:4] + return { + "expected_success_score": max(0.0, min(1.0, float(padded[0]))), + "expected_latency_ms": max(0.0, float(padded[1])) * 10000.0, + "expected_step_depth": max(0.0, float(padded[2])) * 20.0, + "trajectory_density": max(0.0, min(1.0, float(padded[3]))), + } + + +class AgentRecommender: + """Recommend agents using historical success and task similarity.""" + + def __init__(self) -> None: + self._stats: dict[tuple[str, str], AgentPerformance] = {} + self._trained = False + + def train_on_history(self, metrics_events: list[dict[str, Any]]) -> None: + grouped: dict[tuple[str, str], dict[str, float]] = {} + for event in metrics_events: + agent_id = str(event.get("agent_id", "unknown")) + task_type = str(event.get("task_type", "general")) + key = (agent_id, task_type) + grouped.setdefault( + key, + { + "count": 0.0, + "success_sum": 0.0, + "success_bool_sum": 0.0, + "latency_sum": 0.0, + }, + ) + g = grouped[key] + score = float(event.get("success_score", 1.0 if event.get("success") else 0.0)) + g["count"] += 1.0 + g["success_sum"] += score + g["success_bool_sum"] += 1.0 if score >= 0.5 else 0.0 + g["latency_sum"] += float(event.get("execution_time_ms", 0.0)) + + self._stats = {} + for (agent_id, task_type), g in grouped.items(): + count = max(1, int(g["count"])) + self._stats[(agent_id, task_type)] = AgentPerformance( + agent_id=agent_id, + task_type=task_type, + success_rate=g["success_bool_sum"] / count, + avg_success_score=g["success_sum"] / count, + avg_latency_ms=g["latency_sum"] / count, + sample_count=count, + ) + self._trained = True + + def recommend_agents( + self, + task_description: str, + context: dict[str, Any] | None = None, + top_k: int = 3, + ) -> list[dict[str, Any]]: + if not self._trained or not self._stats: + return [] + + context = context or {} + target_task_type = str(context.get("task_type", "general")) + target_tokens = _tokenize(f"{target_task_type} {task_description}") + + scored: list[tuple[float, AgentPerformance]] = [] + for perf in self._stats.values(): + type_tokens = _tokenize(perf.task_type) + overlap = len(target_tokens & type_tokens) + similarity = overlap / max(1, len(target_tokens)) + + # Higher success and lower latency are preferred; sample_count adds confidence. + confidence = min(1.0, perf.sample_count / 10.0) + latency_bonus = 1.0 / (1.0 + perf.avg_latency_ms / 1000.0) + final_score = ( + perf.avg_success_score * 0.6 + + perf.success_rate * 0.2 + + similarity * 0.1 + + latency_bonus * 0.1 + ) * (0.7 + 0.3 * confidence) + scored.append((final_score, perf)) + + scored.sort(key=lambda item: item[0], reverse=True) + return [ + { + "agent_id": perf.agent_id, + "task_type": perf.task_type, + "score": round(score, 4), + "success_rate": round(perf.success_rate, 4), + "avg_success_score": round(perf.avg_success_score, 4), + "avg_latency_ms": round(perf.avg_latency_ms, 2), + "sample_count": perf.sample_count, + } + for score, perf in scored[: max(0, top_k)] + ] + + def get_agent_success_rate(self, agent_id: str, task_type: str = "general") -> float: + perf = self._stats.get((agent_id, task_type)) + if perf: + return perf.success_rate + + # Fallback to overall rate for the agent across all task types. + matches = [p for p in self._stats.values() if p.agent_id == agent_id] + if not matches: + return 0.0 + return sum(p.success_rate for p in matches) / len(matches) + + +class CommitteeBuilder: + """Build and optimize an execution committee from recommendations.""" + + def build_committee( + self, + recommendations: list[dict[str, Any]], + constraints: dict[str, Any] | None = None, + ) -> list[dict[str, Any]]: + constraints = constraints or {} + max_members = int(constraints.get("max_members", 3)) + min_score = float(constraints.get("min_score", 0.0)) + filtered = [item for item in recommendations if float(item.get("score", 0.0)) >= min_score] + return filtered[: max(0, max_members)] + + def optimize_committee_composition( + self, + candidate_agents: list[dict[str, Any]], + ) -> list[dict[str, Any]]: + # Encourage task-type diversity while keeping high-ranked agents first. + chosen: list[dict[str, Any]] = [] + seen_task_types: set[str] = set() + for agent in candidate_agents: + task_type = str(agent.get("task_type", "general")) + if task_type not in seen_task_types: + chosen.append(agent) + seen_task_types.add(task_type) + + # Add remaining candidates to fill committee capacity. + for agent in candidate_agents: + if agent not in chosen: + chosen.append(agent) + return chosen diff --git a/core/context_repository.py b/core/context_repository.py new file mode 100644 index 0000000..0dab5be --- /dev/null +++ b/core/context_repository.py @@ -0,0 +1,565 @@ +"""Context Prompt Repository for storing and retrieving prompt templates. + +This module provides: +- Storage of prompt templates for various task types +- Semantic search for relevant context extraction +- Integration with prompt_builder for dynamic context assembly +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Any + +logger = logging.getLogger(__name__) + + +@dataclass +class PromptTemplate: + """A template for generating prompts with variables.""" + + template_id: str + template: str + description: str = "" + variables: list[str] = field(default_factory=list) + metadata: dict[str, Any] = field(default_factory=dict) + tags: list[str] = field(default_factory=list) + usage_count: int = 0 + success_rate: float = 0.5 # Default prior + + def render(self, **kwargs: Any) -> str: + """Render the template with provided variables. + + Args: + **kwargs: Variable values to substitute in the template. + + Returns: + Rendered prompt string. + + Raises: + ValueError: If required variables are missing. + """ + try: + rendered = self.template.format(**kwargs) + self.usage_count += 1 + return rendered + except KeyError as e: + missing_var = e.args[0] + logger.warning( + "Missing variable '%s' for template '%s'. Available: %s", + missing_var, + self.template_id, + list(kwargs.keys()), + ) + raise ValueError( + f"Missing required variable '{missing_var}' for template '{self.template_id}'" + ) from e + + def to_dict(self) -> dict[str, Any]: + """Convert template to dictionary representation.""" + return { + "template_id": self.template_id, + "template": self.template, + "description": self.description, + "variables": self.variables, + "metadata": self.metadata, + "tags": self.tags, + "usage_count": self.usage_count, + "success_rate": self.success_rate, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> PromptTemplate: + """Create a PromptTemplate from a dictionary.""" + return cls( + template_id=data.get("template_id", ""), + template=data.get("template", ""), + description=data.get("description", ""), + variables=data.get("variables", []), + metadata=data.get("metadata", {}), + tags=data.get("tags", []), + usage_count=data.get("usage_count", 0), + success_rate=data.get("success_rate", 0.5), + ) + + +class SemanticIndex: + """Simple semantic index for prompt similarity search. + + Uses keyword-based similarity with optional embedding support. + For production, integrate with sentence-transformers for better embeddings. + """ + + def __init__(self) -> None: + """Initialize the semantic index.""" + self._index: dict[str, dict[str, float]] = {} + self._keyword_index: dict[str, set[str]] = {} + self._built = False + + def build_index(self, prompts: list[PromptTemplate]) -> None: + """Build the semantic index from a list of prompts. + + Args: + prompts: List of PromptTemplate objects to index. + """ + self._index.clear() + self._keyword_index.clear() + + for prompt in prompts: + template_id = prompt.template_id + + # Extract keywords from template, description, and tags + text_content = ( + f"{prompt.template} {prompt.description} {' '.join(prompt.tags)}" + ).lower() + + # Simple keyword extraction (can be enhanced with TF-IDF or embeddings) + keywords = self._extract_keywords(text_content) + + # Store keyword weights for this template + self._index[template_id] = {} + for kw in keywords: + self._index[template_id][kw] = self._index[template_id].get(kw, 0) + 1 + + # Build reverse index + if kw not in self._keyword_index: + self._keyword_index[kw] = set() + self._keyword_index[kw].add(template_id) + + self._built = True + logger.info("Built semantic index with %d prompts", len(prompts)) + + def _extract_keywords(self, text: str) -> list[str]: + """Extract keywords from text. + + Simple implementation: split on whitespace and remove stopwords. + Can be enhanced with NLP techniques. + """ + stopwords = { + "the", "a", "an", "is", "are", "was", "were", "be", "been", + "being", "have", "has", "had", "do", "does", "did", "will", + "would", "could", "should", "may", "might", "must", "shall", + "can", "need", "to", "of", "in", "for", "on", "with", "at", + "by", "from", "as", "into", "through", "during", "before", + "after", "above", "below", "between", "under", "again", + "further", "then", "once", "here", "there", "when", "where", + "why", "how", "all", "each", "few", "more", "most", "other", + "some", "such", "no", "nor", "not", "only", "own", "same", + "so", "than", "too", "very", "just", "and", "but", "if", "or", + "because", "until", "while", "although", "though", "this", + "that", "these", "those", "it", "its" + } + + words = text.split() + keywords = [ + word.strip(".,!?;:\"'()[]{}") + for word in words + if word.lower() not in stopwords and len(word) > 2 + ] + return keywords + + def similarity_search( + self, + query: str, + top_k: int = 5, + ) -> list[tuple[str, float]]: + """Search for similar prompts based on query. + + Args: + query: Search query string. + top_k: Number of results to return. + + Returns: + List of (template_id, score) tuples sorted by relevance. + """ + if not self._built: + logger.warning("Semantic index not built yet") + return [] + + query_keywords = set(self._extract_keywords(query.lower())) + + if not query_keywords: + return [] + + # Calculate similarity scores using Jaccard-like metric + scores: dict[str, float] = {} + + for keyword in query_keywords: + if keyword in self._keyword_index: + for template_id in self._keyword_index[keyword]: + if template_id not in scores: + scores[template_id] = 0.0 + + # Weight by keyword frequency in template + keyword_weight = self._index[template_id].get(keyword, 0) + scores[template_id] += keyword_weight + + # Normalize scores + if scores: + max_score = max(scores.values()) + if max_score > 0: + scores = {k: v / max_score for k, v in scores.items()} + + # Sort by score descending + sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True) + + return sorted_results[:top_k] + + +class ContextPromptRepository: + """Repository for storing and retrieving prompt templates. + + Provides centralized storage for prompt templates with semantic search + capabilities for finding relevant prompts based on task context. + """ + + def __init__(self) -> None: + """Initialize the repository.""" + self._templates: dict[str, PromptTemplate] = {} + self._semantic_index = SemanticIndex() + self._initialized = False + + def store_prompt( + self, + template_id: str, + prompt_template: str, + description: str = "", + variables: list[str] | None = None, + metadata: dict[str, Any] | None = None, + tags: list[str] | None = None, + rebuild_index: bool = True, + ) -> PromptTemplate: + """Store a prompt template in the repository. + + Args: + template_id: Unique identifier for the template. + prompt_template: The template string with {variable} placeholders. + description: Human-readable description of the template's purpose. + variables: List of variable names used in the template. + metadata: Additional metadata for the template. + tags: Tags for categorization and search. + rebuild_index: Whether to rebuild the semantic index immediately. + + Returns: + The stored PromptTemplate object. + """ + # Auto-detect variables if not provided + if variables is None: + import re + variables = list(set(re.findall(r'\{(\w+)\}', prompt_template))) + + template = PromptTemplate( + template_id=template_id, + template=prompt_template, + description=description, + variables=variables or [], + metadata=metadata or {}, + tags=tags or [], + ) + + self._templates[template_id] = template + logger.info("Stored prompt template '%s' with variables: %s", template_id, variables) + + # Rebuild semantic index unless batching multiple updates. + if rebuild_index: + self._rebuild_index() + + return template + + def retrieve_prompts( + self, + query: str, + top_k: int = 5, + tags_filter: list[str] | None = None, + ) -> list[PromptTemplate]: + """Retrieve prompts relevant to a query. + + Args: + query: Search query describing the task or context. + top_k: Maximum number of results to return. + tags_filter: Optional list of tags to filter by. + + Returns: + List of relevant PromptTemplate objects. + """ + # Use semantic search + search_results = self._semantic_index.similarity_search(query, top_k=top_k * 2) + + results: list[PromptTemplate] = [] + + for template_id, score in search_results: + if template_id not in self._templates: + continue + + template = self._templates[template_id] + + # Apply tag filter if specified + if tags_filter: + if not any(tag in template.tags for tag in tags_filter): + continue + + results.append(template) + + if len(results) >= top_k: + break + + # Fallback: if no semantic matches, return all templates + if not results and self._templates: + results = list(self._templates.values())[:top_k] + + logger.debug("Retrieved %d prompts for query: %s", len(results), query) + return results + + def get_context_for_task( + self, + task_type: str, + context_hints: dict[str, Any] | None = None, + top_k: int = 3, + ) -> list[dict[str, Any]]: + """Get relevant context prompts for a specific task type. + + Args: + task_type: Type of task (e.g., 'code_generation', 'debugging', 'analysis'). + context_hints: Additional context information for better matching. + top_k: Number of context prompts to return. + + Returns: + List of dicts with 'template', 'variables', and 'context' keys. + """ + # Build search query from task type and hints + query_parts = [task_type] + if context_hints: + for key, value in context_hints.items(): + query_parts.append(f"{key}: {value}") + + query = " ".join(query_parts) + + templates = self.retrieve_prompts(query, top_k=top_k) + + context_list = [] + for template in templates: + context_item = { + "template_id": template.template_id, + "template": template.template, + "variables": template.variables, + "description": template.description, + "tags": template.tags, + } + context_list.append(context_item) + + logger.info( + "Retrieved %d context prompts for task type '%s'", + len(context_list), + task_type, + ) + return context_list + + def get_template(self, template_id: str) -> PromptTemplate | None: + """Get a specific template by ID. + + Args: + template_id: The unique template identifier. + + Returns: + The PromptTemplate if found, None otherwise. + """ + return self._templates.get(template_id) + + def update_success_rate(self, template_id: str, success: bool) -> None: + """Update the success rate for a template. + + Args: + template_id: The template to update. + success: Whether the template usage was successful. + """ + if template_id not in self._templates: + logger.warning("Template '%s' not found for success rate update", template_id) + return + + template = self._templates[template_id] + + # Exponential moving average + alpha = 0.1 + current = template.success_rate + new_value = 1.0 if success else 0.0 + template.success_rate = (1 - alpha) * current + alpha * new_value + + logger.debug( + "Updated success rate for '%s': %.3f", + template_id, + template.success_rate, + ) + + def list_templates(self, tags_filter: list[str] | None = None) -> list[str]: + """List all template IDs, optionally filtered by tags. + + Args: + tags_filter: Optional list of tags to filter by. + + Returns: + List of template IDs. + """ + if not tags_filter: + return list(self._templates.keys()) + + return [ + tid for tid, template in self._templates.items() + if any(tag in template.tags for tag in tags_filter) + ] + + def export_templates(self) -> dict[str, Any]: + """Export all templates as a dictionary. + + Returns: + Dictionary of template_id -> template_dict mappings. + """ + return { + tid: template.to_dict() + for tid, template in self._templates.items() + } + + def import_templates(self, templates_data: dict[str, Any]) -> int: + """Import templates from a dictionary. + + Args: + templates_data: Dictionary of template data. + + Returns: + Number of templates imported. + """ + count = 0 + for template_id, data in templates_data.items(): + # Ensure template_id matches + data["template_id"] = template_id + template = PromptTemplate.from_dict(data) + self._templates[template_id] = template + count += 1 + + self._rebuild_index() + logger.info("Imported %d templates", count) + return count + + def _rebuild_index(self) -> None: + """Rebuild the semantic index from current templates.""" + prompts = list(self._templates.values()) + self._semantic_index.build_index(prompts) + self._initialized = True + + def initialize_with_defaults(self) -> None: + """Initialize the repository with default prompt templates.""" + default_templates = [ + # Code generation templates + { + "template_id": "code_generation_python", + "template": ( + "You are an expert Python developer. Generate clean, efficient, " + "and well-documented Python code for the following task:\n\n" + "Task: {task_description}\n\n" + "Requirements:\n{requirements}\n\n" + "Constraints:\n{constraints}\n\n" + "Please provide the complete implementation with error handling." + ), + "description": "Template for Python code generation tasks", + "tags": ["code", "python", "generation"], + "metadata": {"language": "python", "complexity": "medium"}, + }, + { + "template_id": "code_review", + "template": ( + "Review the following code for quality, security, and best practices:\n\n" + "Code:\n```{language}\n{code}\n```\n\n" + "Focus areas: {focus_areas}\n\n" + "Provide specific recommendations with code examples where applicable." + ), + "description": "Template for code review tasks", + "tags": ["code", "review", "analysis"], + "metadata": {"type": "review"}, + }, + # Debugging templates + { + "template_id": "debug_error", + "template": ( + "Help debug the following error:\n\n" + "Error message: {error_message}\n\n" + "Code context:\n```{language}\n{code_snippet}\n```\n\n" + "Stack trace:\n{stack_trace}\n\n" + "Describe the likely cause and provide a fix." + ), + "description": "Template for debugging error messages", + "tags": ["debug", "error", "troubleshooting"], + "metadata": {"type": "debug"}, + }, + # Analysis templates + { + "template_id": "data_analysis", + "template": ( + "Analyze the following data and provide insights:\n\n" + "Data description: {data_description}\n\n" + "Analysis goals: {goals}\n\n" + "Key questions to answer:\n{questions}\n\n" + "Provide a structured analysis with findings and recommendations." + ), + "description": "Template for data analysis tasks", + "tags": ["analysis", "data", "insights"], + "metadata": {"type": "analysis"}, + }, + # Documentation templates + { + "template_id": "doc_generation", + "template": ( + "Generate documentation for the following code:\n\n" + "Code:\n```{language}\n{code}\n```\n\n" + "Documentation style: {style}\n\n" + "Include: {include_sections}\n\n" + "Generate comprehensive documentation suitable for {audience}." + ), + "description": "Template for generating code documentation", + "tags": ["documentation", "writing", "code"], + "metadata": {"type": "documentation"}, + }, + # Testing templates + { + "template_id": "test_generation", + "template": ( + "Generate comprehensive tests for the following code:\n\n" + "Code to test:\n```{language}\n{code}\n```\n\n" + "Testing framework: {framework}\n\n" + "Test scenarios to cover:\n{scenarios}\n\n" + "Include edge cases and error conditions." + ), + "description": "Template for generating test cases", + "tags": ["testing", "code", "quality"], + "metadata": {"type": "testing"}, + }, + ] + + for template_data in default_templates: + self.store_prompt( + template_id=template_data["template_id"], + prompt_template=template_data["template"], + description=template_data["description"], + tags=template_data["tags"], + metadata=template_data.get("metadata", {}), + rebuild_index=False, + ) + self._rebuild_index() + + logger.info("Initialized repository with %d default templates", len(default_templates)) + + +# Singleton instance for easy access +_repository_instance: ContextPromptRepository | None = None + + +def get_repository() -> ContextPromptRepository: + """Get the singleton repository instance. + + Returns: + The ContextPromptRepository instance. + """ + global _repository_instance + if _repository_instance is None: + _repository_instance = ContextPromptRepository() + _repository_instance.initialize_with_defaults() + return _repository_instance diff --git a/core/metrics.py b/core/metrics.py index dbee6af..f58e865 100644 --- a/core/metrics.py +++ b/core/metrics.py @@ -38,6 +38,7 @@ def log_execution( success_score: float | None = None, trajectory_id: str | None = None, step_number: int | None = None, + task_type: str | None = None, ) -> None: """Log a single plugin execution. @@ -52,6 +53,7 @@ def log_execution( success_score: Float score 0.0-1.0 indicating degree of success. trajectory_id: Unique ID for multi-step operation sequence. step_number: Position in the trajectory (1-indexed). + task_type: Optional task category for agent recommendation analytics. """ data: dict[str, Any] = { "plugin_name": plugin_name, @@ -70,6 +72,8 @@ def log_execution( data["trajectory_id"] = trajectory_id if step_number is not None: data["step_number"] = step_number + if task_type is not None: + data["task_type"] = task_type log_event("plugin_execution", data, metrics_file=metrics_file) @@ -246,3 +250,41 @@ def _plugin_stats(name: str) -> dict[str, Any]: del s["trajectories"] return stats + + +def get_execution_events( + metrics_file: Path | None = None, + trajectory_id: str | None = None, +) -> list[dict[str, Any]]: + """Return plugin execution events, optionally filtered by trajectory.""" + events = get_events(event_type="plugin_execution", metrics_file=metrics_file) + if trajectory_id is None: + return events + return [event for event in events if event.get("trajectory_id") == trajectory_id] + + +def build_agent_history( + metrics_file: Path | None = None, +) -> list[dict[str, Any]]: + """Build normalized training records from execution metrics. + + The returned records are suitable for lightweight recommendation models and + include a unified success score in the [0.0, 1.0] range. + """ + history: list[dict[str, Any]] = [] + for event in get_execution_events(metrics_file=metrics_file): + if "success_score" in event: + score = float(event["success_score"]) + else: + score = 1.0 if event.get("success") else 0.0 + history.append( + { + "agent_id": event.get("plugin_name", "unknown"), + "task_type": event.get("task_type", "general"), + "success_score": max(0.0, min(1.0, score)), + "trajectory_id": event.get("trajectory_id"), + "step_number": event.get("step_number"), + "execution_time_ms": float(event.get("execution_time_ms", 0.0)), + } + ) + return history diff --git a/core/reflection.py b/core/reflection.py new file mode 100644 index 0000000..19ae81b --- /dev/null +++ b/core/reflection.py @@ -0,0 +1,703 @@ +"""Tool Reflection Cycle for automatic error analysis and correction. + +This module provides: +- Error analysis for tool execution failures +- Automatic generation of corrected tool calls +- Reflection loop for iterative improvement +""" + +from __future__ import annotations + +import logging +import textwrap +from datetime import datetime +from dataclasses import dataclass, field +from enum import Enum, auto +from typing import Any + +logger = logging.getLogger(__name__) + + +class ErrorCategory(Enum): + """Categories of tool execution errors.""" + + SYNTAX_ERROR = auto() + """Code syntax or parsing error.""" + + RUNTIME_ERROR = auto() + """Runtime exception during execution.""" + + TIMEOUT_ERROR = auto() + """Execution exceeded time limit.""" + + RESOURCE_ERROR = auto() + """Insufficient resources (memory, disk, etc.).""" + + PERMISSION_ERROR = auto() + """Access denied or permission issue.""" + + NOT_FOUND_ERROR = auto() + """Plugin, file, or resource not found.""" + + VALIDATION_ERROR = auto() + """Input validation failed.""" + + NETWORK_ERROR = auto() + """Network-related failure.""" + + UNKNOWN_ERROR = auto() + """Unclassified error.""" + + +@dataclass +class ErrorAnalysis: + """Result of analyzing a tool execution error.""" + + error_category: ErrorCategory + error_message: str + error_type: str + stack_trace: str | None = None + root_cause: str = "" + confidence: float = 0.5 + suggestions: list[str] = field(default_factory=list) + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + """Convert analysis to dictionary representation.""" + return { + "error_category": self.error_category.name, + "error_message": self.error_message, + "error_type": self.error_type, + "stack_trace": self.stack_trace, + "root_cause": self.root_cause, + "confidence": self.confidence, + "suggestions": self.suggestions, + "metadata": self.metadata, + } + + +@dataclass +class CorrectionResult: + """Result of generating a correction for a failed tool call.""" + + success: bool + corrected_call: dict[str, Any] | None = None + explanation: str = "" + confidence: float = 0.5 + alternative_approaches: list[str] = field(default_factory=list) + requires_human_review: bool = False + + def to_dict(self) -> dict[str, Any]: + """Convert correction result to dictionary representation.""" + return { + "success": self.success, + "corrected_call": self.corrected_call, + "explanation": self.explanation, + "confidence": self.confidence, + "alternative_approaches": self.alternative_approaches, + "requires_human_review": self.requires_human_review, + } + + +class ErrorAnalyzer: + """Analyzes tool execution errors to determine root cause and category.""" + + def __init__(self) -> None: + """Initialize the error analyzer.""" + # Error pattern mappings for categorization + self._error_patterns: dict[ErrorCategory, list[str]] = { + ErrorCategory.SYNTAX_ERROR: [ + "SyntaxError", + "invalid syntax", + "unexpected indent", + "unexpected EOF", + "missing parentheses", + ], + ErrorCategory.RUNTIME_ERROR: [ + "TypeError", + "ValueError", + "AttributeError", + "KeyError", + "IndexError", + "RuntimeError", + "AssertionError", + ], + ErrorCategory.TIMEOUT_ERROR: [ + "TimeoutError", + "timeout", + "timed out", + "deadline exceeded", + ], + ErrorCategory.RESOURCE_ERROR: [ + "MemoryError", + "ResourceWarning", + "out of memory", + "disk full", + "no space left", + ], + ErrorCategory.PERMISSION_ERROR: [ + "PermissionError", + "AccessDenied", + "access denied", + "permission denied", + "not permitted", + ], + ErrorCategory.NOT_FOUND_ERROR: [ + "FileNotFoundError", + "ModuleNotFoundError", + "NotFound", + "does not exist", + "not found", + ], + ErrorCategory.VALIDATION_ERROR: [ + "ValidationError", + "InvalidInput", + "invalid value", + "validation failed", + "constraint violation", + ], + ErrorCategory.NETWORK_ERROR: [ + "ConnectionError", + "ConnectionRefusedError", + "ConnectionResetError", + "BrokenPipeError", + "network", + "connection", + ], + } + self._error_patterns = { + category: [pattern.lower() for pattern in patterns] + for category, patterns in self._error_patterns.items() + } + + def analyze_error( + self, + tool_call: dict[str, Any], + result: dict[str, Any], + traceback_str: str | None = None, + ) -> ErrorAnalysis: + """Analyze an error from tool execution. + + Args: + tool_call: The original tool call that failed. + result: The result dict containing error information. + traceback_str: Optional full traceback string. + + Returns: + ErrorAnalysis with categorized error and suggestions. + """ + error_message = result.get("error", result.get("error_message", "Unknown error")) + error_type = result.get("error_type", "UnknownError") + + # Combine error message and traceback for analysis + full_text = f"{error_type}: {error_message}" + if traceback_str: + full_text += f"\n{traceback_str}" + + full_text_lower = full_text.lower() + + # Categorize error based on patterns + category = ErrorCategory.UNKNOWN_ERROR + best_match_count = 0 + + for err_category, patterns in self._error_patterns.items(): + match_count = sum(1 for pattern in patterns if pattern in full_text_lower) + if match_count > best_match_count: + best_match_count = match_count + category = err_category + + # Determine confidence based on pattern matches + confidence = min(1.0, best_match_count / 3.0) # Normalize to 0-1 + + # Generate root cause hypothesis + root_cause = self._hypothesize_root_cause(category, error_message, tool_call) + + # Generate suggestions + suggestions = self._generate_suggestions(category, error_message, tool_call) + + return ErrorAnalysis( + error_category=category, + error_message=error_message, + error_type=error_type, + stack_trace=traceback_str, + root_cause=root_cause, + confidence=confidence, + suggestions=suggestions, + metadata={ + "tool_name": tool_call.get("name", "unknown"), + "tool_input": tool_call.get("input", {}), + }, + ) + + def _hypothesize_root_cause( + self, + category: ErrorCategory, + error_message: str, + tool_call: dict[str, Any], + ) -> str: + """Generate a hypothesis about the root cause of the error.""" + tool_name = tool_call.get("name", "unknown") + + if category == ErrorCategory.SYNTAX_ERROR: + return "The code contains syntax errors that prevent parsing." + + elif category == ErrorCategory.RUNTIME_ERROR: + if "NoneType" in error_message: + return "Attempting to access attributes or methods on a None value." + elif "key" in error_message.lower(): + return "Accessing a dictionary key that doesn't exist." + elif "index" in error_message.lower(): + return "Accessing a sequence index that's out of range." + else: + return "An exception occurred during code execution." + + elif category == ErrorCategory.NOT_FOUND_ERROR: + if "ModuleNotFoundError" in error_message: + return "Required module is not installed or not in the allow-list." + elif "FileNotFoundError" in error_message: + return "Referenced file does not exist at the specified path." + else: + return f"Plugin or resource '{tool_name}' was not found." + + elif category == ErrorCategory.PERMISSION_ERROR: + return "Operation requires permissions that are not granted." + + elif category == ErrorCategory.TIMEOUT_ERROR: + return "Execution took too long and exceeded the time limit." + + elif category == ErrorCategory.VALIDATION_ERROR: + return "Input parameters do not meet validation requirements." + + return "Unable to determine specific root cause." + + def _generate_suggestions( + self, + category: ErrorCategory, + error_message: str, + tool_call: dict[str, Any], + ) -> list[str]: + """Generate actionable suggestions for fixing the error.""" + suggestions = [] + + if category == ErrorCategory.SYNTAX_ERROR: + suggestions.extend([ + "Check for missing colons, parentheses, or quotes.", + "Verify proper indentation (Python uses 4 spaces).", + "Use a linter or IDE to identify syntax issues.", + ]) + + elif category == ErrorCategory.RUNTIME_ERROR: + suggestions.extend([ + "Add error handling with try-except blocks.", + "Validate inputs before processing.", + "Check for None values before accessing attributes.", + ]) + + elif category == ErrorCategory.NOT_FOUND_ERROR: + suggestions.extend([ + "Verify the plugin name is spelled correctly.", + "Ensure the plugin has been loaded successfully.", + "Check if required dependencies are installed.", + ]) + + elif category == ErrorCategory.PERMISSION_ERROR: + suggestions.extend([ + "Review file or resource permissions.", + "Use appropriate authentication if required.", + "Consider alternative approaches that don't require elevated permissions.", + ]) + + elif category == ErrorCategory.TIMEOUT_ERROR: + suggestions.extend([ + "Optimize the code for better performance.", + "Break the task into smaller chunks.", + "Consider using asynchronous operations.", + ]) + + elif category == ErrorCategory.VALIDATION_ERROR: + suggestions.extend([ + "Review the expected input format and types.", + "Add input validation before calling the tool.", + "Check documentation for parameter requirements.", + ]) + + return suggestions + + def categorize_error(self, error_type: str, error_message: str) -> ErrorCategory: + """Quickly categorize an error without full analysis. + + Args: + error_type: The exception type name. + error_message: The error message. + + Returns: + The ErrorCategory for this error. + """ + full_text = f"{error_type}: {error_message}".lower() + + for category, patterns in self._error_patterns.items(): + if any(pattern in full_text for pattern in patterns): + return category + + return ErrorCategory.UNKNOWN_ERROR + + +class CorrectionGenerator: + """Generates corrected tool calls based on error analysis.""" + + def __init__(self) -> None: + """Initialize the correction generator.""" + self._analyzer = ErrorAnalyzer() + + def generate_correction( + self, + error_analysis: ErrorAnalysis, + original_call: dict[str, Any], + ) -> CorrectionResult: + """Generate a corrected version of a failed tool call. + + Args: + error_analysis: Analysis of what went wrong. + original_call: The original tool call that failed. + + Returns: + CorrectionResult with corrected call or explanation. + """ + category = error_analysis.error_category + # Try to generate corrections based on error category + if category == ErrorCategory.SYNTAX_ERROR: + return self._handle_syntax_error(original_call, error_analysis) + + elif category == ErrorCategory.RUNTIME_ERROR: + return self._handle_runtime_error(original_call, error_analysis) + + elif category == ErrorCategory.NOT_FOUND_ERROR: + return self._handle_not_found_error(original_call, error_analysis) + + elif category == ErrorCategory.VALIDATION_ERROR: + return self._handle_validation_error(original_call, error_analysis) + + elif category == ErrorCategory.PERMISSION_ERROR: + return self._handle_permission_error(original_call, error_analysis) + + else: + # For unknown or complex errors, suggest human review + return CorrectionResult( + success=False, + explanation=f"Unable to automatically correct {category.name}. " + f"Root cause: {error_analysis.root_cause}", + confidence=0.3, + requires_human_review=True, + alternative_approaches=error_analysis.suggestions, + ) + + def _handle_syntax_error( + self, + original_call: dict[str, Any], + error_analysis: ErrorAnalysis, + ) -> CorrectionResult: + """Handle syntax errors - typically need LLM regeneration.""" + return CorrectionResult( + success=False, + explanation="Syntax errors require code regeneration. " + "Please review the code and fix syntax issues.", + confidence=0.4, + requires_human_review=False, + alternative_approaches=[ + "Regenerate the code with proper syntax.", + "Use an IDE or linter to identify syntax issues.", + *error_analysis.suggestions, + ], + ) + + def _handle_runtime_error( + self, + original_call: dict[str, Any], + error_analysis: ErrorAnalysis, + ) -> CorrectionResult: + """Handle runtime errors - may be fixable with input adjustments.""" + tool_input = original_call.get("input", {}) + + # Suggest adding error handling + corrected_input = tool_input.copy() + + # If it's a code execution, suggest wrapping in try-except + if "code" in tool_input: + corrected_input["code"] = self._wrap_with_error_handling(tool_input["code"]) + + return CorrectionResult( + success=True, + corrected_call={ + "name": original_call.get("name"), + "input": corrected_input, + }, + explanation="Added error handling to catch runtime exceptions.", + confidence=0.6, + alternative_approaches=error_analysis.suggestions, + ) + + def _handle_not_found_error( + self, + original_call: dict[str, Any], + error_analysis: ErrorAnalysis, + ) -> CorrectionResult: + """Handle not found errors - check names and availability.""" + tool_name = original_call.get("name", "") + + # Common fixes for not found errors + suggestions = [ + f"Verify that plugin '{tool_name}' exists and is loaded.", + "Check for typos in the plugin name.", + "Load the plugin before executing.", + ] + + return CorrectionResult( + success=False, + explanation=f"Plugin or resource '{tool_name}' not found.", + confidence=0.5, + requires_human_review=False, + alternative_approaches=suggestions + error_analysis.suggestions, + ) + + def _handle_validation_error( + self, + original_call: dict[str, Any], + error_analysis: ErrorAnalysis, + ) -> CorrectionResult: + """Handle validation errors - adjust input parameters.""" + # Generic validation fix suggestions + return CorrectionResult( + success=False, + explanation="Input validation failed. Review parameter types and constraints.", + confidence=0.5, + requires_human_review=False, + alternative_approaches=[ + "Validate all input parameters match expected types.", + "Check for required vs optional parameters.", + "Review parameter value ranges and formats.", + *error_analysis.suggestions, + ], + ) + + def _handle_permission_error( + self, + original_call: dict[str, Any], + error_analysis: ErrorAnalysis, + ) -> CorrectionResult: + """Handle permission errors - suggest alternatives.""" + return CorrectionResult( + success=False, + explanation="Operation requires permissions that are not available.", + confidence=0.4, + requires_human_review=True, + alternative_approaches=[ + "Request necessary permissions from administrator.", + "Find alternative approach that doesn't require elevated permissions.", + "Use sandboxed or restricted version of the operation.", + *error_analysis.suggestions, + ], + ) + + def _wrap_with_error_handling(self, code: str) -> str: + """Wrap code in a try-except block for better error handling.""" + indented_code = textwrap.indent(code, " ") + wrapped = ( + "try:\n" + f"{indented_code}\n" + "except Exception as e:\n" + ' print(f"Error during execution: {e}")\n' + " import traceback\n" + " traceback.print_exc()\n" + " raise\n" + ) + return wrapped + + def validate_correction(self, proposed_call: dict[str, Any]) -> bool: + """Validate that a proposed correction is reasonable. + + Args: + proposed_call: The corrected tool call to validate. + + Returns: + True if the correction appears valid. + """ + # Basic validation checks + if not proposed_call: + return False + + if "name" not in proposed_call: + return False + + if "input" not in proposed_call: + return False + + if not isinstance(proposed_call["input"], dict): + return False + + return True + + +class ReflectionLoop: + """Manages the reflection cycle for continuous improvement. + + Tracks errors, generates corrections, and learns from outcomes. + """ + + def __init__(self, max_history: int = 100) -> None: + """Initialize the reflection loop. + + Args: + max_history: Maximum number of reflection events to keep in memory. + """ + self._analyzer = ErrorAnalyzer() + self._correction_generator = CorrectionGenerator() + self._history: list[dict[str, Any]] = [] + self._max_history = max_history + self._success_counts: dict[str, int] = {} + self._failure_counts: dict[str, int] = {} + + def run_reflection_cycle( + self, + tool_call: dict[str, Any], + result: dict[str, Any], + traceback_str: str | None = None, + ) -> dict[str, Any]: + """Run a complete reflection cycle on a failed tool call. + + Args: + tool_call: The tool call that failed. + result: The error result. + traceback_str: Optional full traceback. + + Returns: + Dict with analysis, correction, and recommendations. + """ + tool_name = tool_call.get("name", "unknown") + + # Step 1: Analyze the error + error_analysis = self._analyzer.analyze_error(tool_call, result, traceback_str) + + logger.info( + "Reflection: Analyzed error for '%s' - Category: %s, Confidence: %.2f", + tool_name, + error_analysis.error_category.name, + error_analysis.confidence, + ) + + # Step 2: Generate correction + correction = self._correction_generator.generate_correction( + error_analysis, tool_call + ) + + # Step 3: Record the reflection event + reflection_event = { + "tool_call": tool_call, + "error_analysis": error_analysis.to_dict(), + "correction": correction.to_dict(), + "timestamp": datetime.now().isoformat(), + } + + self._record_reflection(reflection_event) + + # Step 4: Update success/failure counts + if correction.success: + self._success_counts[tool_name] = self._success_counts.get(tool_name, 0) + 1 + else: + self._failure_counts[tool_name] = self._failure_counts.get(tool_name, 0) + 1 + + return { + "analysis": error_analysis.to_dict(), + "correction": correction.to_dict(), + "recommendation": self._get_recommendation(tool_name, error_analysis, correction), + } + + def _record_reflection(self, event: dict[str, Any]) -> None: + """Record a reflection event in history.""" + self._history.append(event) + + # Trim history if needed + while len(self._history) > self._max_history: + self._history.pop(0) + + def log_reflection_event(self, reflection_data: dict[str, Any]) -> None: + """Log a reflection event for later analysis. + + Args: + reflection_data: Data about the reflection event. + """ + # This could be extended to write to metrics or external storage + logger.debug("Logged reflection event: %s", reflection_data.get("tool_name", "unknown")) + + def _get_recommendation( + self, + tool_name: str, + error_analysis: ErrorAnalysis, + correction: CorrectionResult, + ) -> str: + """Generate a recommendation based on the reflection.""" + if correction.success: + return ( + f"Apply the suggested correction for '{tool_name}'. " + f"Confidence: {correction.confidence:.0%}" + ) + elif correction.requires_human_review: + return ( + f"Human review recommended for '{tool_name}'. " + f"Issue: {error_analysis.root_cause}" + ) + else: + return ( + f"Consider alternative approaches for '{tool_name}': " + f"{', '.join(correction.alternative_approaches[:2])}" + ) + + def get_tool_success_rate(self, tool_name: str) -> float: + """Get the historical success rate for a tool. + + Args: + tool_name: Name of the tool. + + Returns: + Success rate between 0.0 and 1.0. + """ + successes = self._success_counts.get(tool_name, 0) + failures = self._failure_counts.get(tool_name, 0) + total = successes + failures + + if total == 0: + return 0.5 # Default prior + + return successes / total + + def get_reflection_history(self, limit: int = 10) -> list[dict[str, Any]]: + """Get recent reflection events. + + Args: + limit: Maximum number of events to return. + + Returns: + List of reflection event dicts. + """ + return self._history[-limit:] + + def clear_history(self) -> None: + """Clear the reflection history.""" + self._history.clear() + self._success_counts.clear() + self._failure_counts.clear() + + +# Singleton instance for easy access +_reflection_loop_instance: ReflectionLoop | None = None + + +def get_reflection_loop() -> ReflectionLoop: + """Get the singleton reflection loop instance. + + Returns: + The ReflectionLoop instance. + """ + global _reflection_loop_instance + if _reflection_loop_instance is None: + _reflection_loop_instance = ReflectionLoop() + return _reflection_loop_instance diff --git a/HOLOBIONT_ROADMAP.md b/docs/HOLOBIONT_ROADMAP.md similarity index 92% rename from HOLOBIONT_ROADMAP.md rename to docs/HOLOBIONT_ROADMAP.md index 7038806..ac5a72a 100644 --- a/HOLOBIONT_ROADMAP.md +++ b/docs/HOLOBIONT_ROADMAP.md @@ -93,7 +93,7 @@ core/reflection.py **Priority:** 🟡 Medium **Complexity:** High **Dependencies:** 1.1 -**Status:** ⏳ In Progress +**Status:** ✅ Completed --- @@ -135,7 +135,7 @@ def build_startup_prompt( **Priority:** 🟡 Medium **Complexity:** Medium **Dependencies:** None -**Status:** ⏳ Planned +**Status:** ✅ Completed --- @@ -165,7 +165,7 @@ def log_execution(...): --- #### Task 1.4.2: RecVAE for Agent Recommendation -**Files:** new: `core/agent_recommender.py`, modification: `core/metrics.py` +**Files:** new: `core/agents/recommender.py`, modification: `core/metrics.py` **Description:** - Use success history to select agent committee composition - Simple VAE architecture for encoding trajectories @@ -173,7 +173,7 @@ def log_execution(...): **Components:** ``` -core/agent_recommender.py +core/agents/recommender.py ├── TrajectoryEncoder │ ├── encode_trajectory(events_list) -> latent_vector │ └── decode_vector(latent_vector) -> trajectory_pattern @@ -189,7 +189,7 @@ core/agent_recommender.py **Priority:** 🟢 Low (for Phase 1) **Complexity:** Very High **Dependencies:** 1.4.1 -**Status:** ⏳ Planned +**Status:** ✅ Completed (MVP) --- @@ -406,11 +406,20 @@ core/training/ --- ### Sprint 2 (Weeks 3-4): Phase 1 Intelligence -- ⏳ 1.3.1 Context Repository — **Planned** -- ⏳ 1.2.1 ToolReflection Cycle (start) — **In Progress** +- ✅ 1.3.1 Context Repository — **Completed** +- ✅ 1.2.1 ToolReflection Cycle (start) — **Completed** **Goal:** Context prompt repository and beginning of self-reflection cycle implementation. +**Deliverables:** +- `core/context_repository.py`: ContextPromptRepository with semantic search, PromptTemplate, SemanticIndex +- `core/reflection.py`: ErrorAnalyzer, CorrectionGenerator, ReflectionLoop for automatic error analysis and correction +- `tests/test_context_repository.py`: 21 tests for context repository functionality +- `tests/test_reflection.py`: 30 tests for reflection cycle functionality +- Default templates for code generation, debugging, analysis, documentation, and testing +- Error categorization for 8 error types (SyntaxError, RuntimeError, TimeoutError, etc.) +- Automatic correction generation with try-except wrapping for runtime errors + --- ### Sprint 3 (Weeks 5-6): Phase 1 Completion @@ -476,8 +485,8 @@ pyvis>=0.3.0 ### Phase 1 Completion Criteria - [x] All tools go through reranking - [x] System can argumentatively reject a tool -- [ ] ToolReflection cycle works automatically on errors -- [ ] Context Repository provides relevant prompts +- [x] ToolReflection cycle works automatically on errors +- [x] Context Repository provides relevant prompts - [x] Metrics include success_score and trajectory_id ### Phase 2 Completion Criteria @@ -509,5 +518,5 @@ pyvis>=0.3.0 --- *Document created: 2025* -*Version: 1.0* -*Status: Planning* +*Version: 1.1* +*Status: Phase 1 in progress (Sprint 2 completed)* diff --git a/metrics.jsonl b/metrics.jsonl new file mode 100644 index 0000000..206c137 --- /dev/null +++ b/metrics.jsonl @@ -0,0 +1,34 @@ +{"timestamp": "2026-04-27T20:31:30.363107+00:00", "event": "plugin_execution", "plugin_name": "echo", "version": "v0", "execution_time_ms": 82.34510200003342, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:31.370022+00:00", "event": "plugin_execution", "plugin_name": "slow", "version": "v0", "execution_time_ms": 1002.2860459999947, "success": false, "error_type": "TimeoutError", "traceback": null, "import_risk_score": 1} +{"timestamp": "2026-04-27T20:31:31.373427+00:00", "event": "plugin_execution", "plugin_name": "echo", "version": "v0", "execution_time_ms": 1.0, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:31.464824+00:00", "event": "plugin_execution", "plugin_name": "echo", "version": "v0", "execution_time_ms": 87.41144900000108, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:31.487800+00:00", "event": "version_change", "plugin_name": "overwrite", "old_version": "v0", "new_version": "v1_20260427_203131"} +{"timestamp": "2026-04-27T20:31:31.489041+00:00", "event": "plugin_execution", "plugin_name": "overwrite", "version": "v1", "execution_time_ms": 0.18471499998895524, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:31.503893+00:00", "event": "version_change", "plugin_name": "versioned_plugin", "old_version": "v0", "new_version": "v1_20260427_203131"} +{"timestamp": "2026-04-27T20:31:31.603871+00:00", "event": "plugin_execution", "plugin_name": "reloadable", "version": "v0", "execution_time_ms": 96.17241000000831, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:31.720859+00:00", "event": "plugin_execution", "plugin_name": "withinit", "version": "v0", "execution_time_ms": 79.7653500000024, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:31.807285+00:00", "event": "plugin_execution", "plugin_name": "scalar", "version": "v0", "execution_time_ms": 83.35234800000535, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:31.907895+00:00", "event": "plugin_execution", "plugin_name": "faulty", "version": "v0", "execution_time_ms": 97.4959519999743, "success": false, "error_type": "RuntimeError", "traceback": "Traceback (most recent call last):\n File \"/workspace/core/sandbox_wrapper.py\", line 62, in main\n result = run_fn(input_data)\n ^^^^^^^^^^^^^^^^^^\n File \"/tmp/pytest-of-root/pytest-0/test_call_plugin_raises_except0/plugins/faulty.py\", line 2, in run\n raise ValueError(\"oops\")\nValueError: oops\n", "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:32.023598+00:00", "event": "plugin_execution", "plugin_name": "versioned", "version": "v0", "execution_time_ms": 102.04897199997731, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:32.029958+00:00", "event": "version_change", "plugin_name": "rollme", "old_version": "v0", "new_version": "v1_20260427_203132"} +{"timestamp": "2026-04-27T20:31:32.041702+00:00", "event": "plugin_execution", "plugin_name": "rollme", "version": "v1", "execution_time_ms": 0.19913400001314585, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:32.042648+00:00", "event": "rollback", "plugin_name": "rollme", "from_version": "v1", "to_version": "v1_20260427_203132"} +{"timestamp": "2026-04-27T20:31:32.043330+00:00", "event": "plugin_execution", "plugin_name": "rollme", "version": "v1", "execution_time_ms": 0.15033500000072308, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:33.387819+00:00", "event": "dependency_request", "plugin_name": "net_plugin", "requested": ["requests"], "pending": ["requests"]} +{"timestamp": "2026-04-27T20:31:43.688498+00:00", "event": "plugin_execution", "plugin_name": "echo", "version": "v0", "execution_time_ms": 82.14405899997246, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:44.704817+00:00", "event": "plugin_execution", "plugin_name": "slow", "version": "v0", "execution_time_ms": 1002.1589440000298, "success": false, "error_type": "TimeoutError", "traceback": null, "import_risk_score": 1} +{"timestamp": "2026-04-27T20:31:44.708343+00:00", "event": "plugin_execution", "plugin_name": "echo", "version": "v0", "execution_time_ms": 1.0, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:44.786072+00:00", "event": "plugin_execution", "plugin_name": "echo", "version": "v0", "execution_time_ms": 74.32693999999174, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:44.807417+00:00", "event": "version_change", "plugin_name": "overwrite", "old_version": "v0", "new_version": "v1_20260427_203144"} +{"timestamp": "2026-04-27T20:31:44.808677+00:00", "event": "plugin_execution", "plugin_name": "overwrite", "version": "v1", "execution_time_ms": 0.2297489999705249, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:44.820895+00:00", "event": "version_change", "plugin_name": "versioned_plugin", "old_version": "v0", "new_version": "v1_20260427_203144"} +{"timestamp": "2026-04-27T20:31:44.928675+00:00", "event": "plugin_execution", "plugin_name": "reloadable", "version": "v0", "execution_time_ms": 103.82989999999381, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:45.084312+00:00", "event": "plugin_execution", "plugin_name": "withinit", "version": "v0", "execution_time_ms": 99.04423399996176, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:45.186682+00:00", "event": "plugin_execution", "plugin_name": "scalar", "version": "v0", "execution_time_ms": 99.27595499999597, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:45.288261+00:00", "event": "plugin_execution", "plugin_name": "faulty", "version": "v0", "execution_time_ms": 98.46046400002706, "success": false, "error_type": "RuntimeError", "traceback": "Traceback (most recent call last):\n File \"/workspace/core/sandbox_wrapper.py\", line 62, in main\n result = run_fn(input_data)\n ^^^^^^^^^^^^^^^^^^\n File \"/tmp/pytest-of-root/pytest-1/test_call_plugin_raises_except0/plugins/faulty.py\", line 2, in run\n raise ValueError(\"oops\")\nValueError: oops\n", "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:45.382332+00:00", "event": "plugin_execution", "plugin_name": "versioned", "version": "v0", "execution_time_ms": 89.9486850000244, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:45.387722+00:00", "event": "version_change", "plugin_name": "rollme", "old_version": "v0", "new_version": "v1_20260427_203145"} +{"timestamp": "2026-04-27T20:31:45.389921+00:00", "event": "plugin_execution", "plugin_name": "rollme", "version": "v1", "execution_time_ms": 0.45444100004488064, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:45.401148+00:00", "event": "rollback", "plugin_name": "rollme", "from_version": "v1", "to_version": "v1_20260427_203145"} +{"timestamp": "2026-04-27T20:31:45.401953+00:00", "event": "plugin_execution", "plugin_name": "rollme", "version": "v1", "execution_time_ms": 0.15798100002939464, "success": true, "error_type": null, "traceback": null, "import_risk_score": 0} +{"timestamp": "2026-04-27T20:31:46.750735+00:00", "event": "dependency_request", "plugin_name": "net_plugin", "requested": ["requests"], "pending": ["requests"]} diff --git a/tests/test_agent_recommender.py b/tests/test_agent_recommender.py new file mode 100644 index 0000000..03c0524 --- /dev/null +++ b/tests/test_agent_recommender.py @@ -0,0 +1,57 @@ +"""Tests for core.agents.recommender.""" + +from core.agents.recommender import AgentRecommender, CommitteeBuilder, TrajectoryEncoder + + +def test_trajectory_encoder_roundtrip() -> None: + encoder = TrajectoryEncoder() + events = [ + {"success_score": 0.9, "execution_time_ms": 100, "step_number": 1}, + {"success_score": 0.7, "execution_time_ms": 300, "step_number": 2}, + ] + vector = encoder.encode_trajectory(events) + decoded = encoder.decode_vector(vector) + assert len(vector) == 4 + assert decoded["expected_success_score"] > 0.0 + assert decoded["expected_latency_ms"] > 0.0 + + +def test_recommender_prefers_higher_score_agent() -> None: + recommender = AgentRecommender() + history = [ + {"agent_id": "coder_a", "task_type": "python", "success_score": 0.9, "execution_time_ms": 150}, + {"agent_id": "coder_a", "task_type": "python", "success_score": 0.8, "execution_time_ms": 180}, + {"agent_id": "coder_b", "task_type": "python", "success_score": 0.4, "execution_time_ms": 120}, + ] + recommender.train_on_history(history) + recs = recommender.recommend_agents("write parser", {"task_type": "python"}, top_k=2) + assert len(recs) == 2 + assert recs[0]["agent_id"] == "coder_a" + assert recs[0]["score"] >= recs[1]["score"] + + +def test_get_agent_success_rate_with_fallback() -> None: + recommender = AgentRecommender() + history = [ + {"agent_id": "planner", "task_type": "analysis", "success_score": 1.0}, + {"agent_id": "planner", "task_type": "debug", "success_score": 0.0}, + ] + recommender.train_on_history(history) + assert recommender.get_agent_success_rate("planner", "analysis") == 1.0 + assert 0.0 <= recommender.get_agent_success_rate("planner", "unknown-task") <= 1.0 + + +def test_committee_builder_respects_constraints_and_diversity() -> None: + builder = CommitteeBuilder() + recommendations = [ + {"agent_id": "a", "task_type": "python", "score": 0.9}, + {"agent_id": "b", "task_type": "python", "score": 0.8}, + {"agent_id": "c", "task_type": "debug", "score": 0.85}, + ] + committee = builder.build_committee( + recommendations, + constraints={"max_members": 2, "min_score": 0.81}, + ) + assert len(committee) == 2 + optimized = builder.optimize_committee_composition(committee) + assert optimized[0]["task_type"] in {"python", "debug"} diff --git a/tests/test_context_repository.py b/tests/test_context_repository.py new file mode 100644 index 0000000..bb0bb76 --- /dev/null +++ b/tests/test_context_repository.py @@ -0,0 +1,378 @@ +"""Tests for the Context Prompt Repository (Sprint 2 - Task 1.3.1).""" + +import pytest +from core.context_repository import ( + ContextPromptRepository, + PromptTemplate, + SemanticIndex, + get_repository, +) + + +class TestPromptTemplate: + """Tests for PromptTemplate class.""" + + def test_template_creation(self): + """Test creating a prompt template.""" + template = PromptTemplate( + template_id="test_template", + template="Hello {name}, welcome to {place}!", + description="A greeting template", + variables=["name", "place"], + tags=["greeting", "welcome"], + ) + + assert template.template_id == "test_template" + assert template.description == "A greeting template" + assert len(template.variables) == 2 + assert len(template.tags) == 2 + assert template.usage_count == 0 + assert template.success_rate == 0.5 + + def test_template_rendering(self): + """Test rendering a template with variables.""" + template = PromptTemplate( + template_id="test", + template="Calculate {a} + {b} = {result}", + variables=["a", "b", "result"], + ) + + rendered = template.render(a=5, b=3, result=8) + assert rendered == "Calculate 5 + 3 = 8" + assert template.usage_count == 1 + + def test_template_rendering_missing_variable(self): + """Test that missing variables raise ValueError.""" + template = PromptTemplate( + template_id="test", + template="Hello {name}!", + variables=["name"], + ) + + with pytest.raises(ValueError, match="Missing required variable"): + template.render() + + def test_template_to_dict(self): + """Test converting template to dictionary.""" + template = PromptTemplate( + template_id="test", + template="Test {value}", + description="Test template", + variables=["value"], + tags=["test"], + metadata={"key": "value"}, + ) + + data = template.to_dict() + assert data["template_id"] == "test" + assert data["template"] == "Test {value}" + assert data["description"] == "Test template" + assert data["variables"] == ["value"] + assert data["tags"] == ["test"] + assert data["metadata"] == {"key": "value"} + + def test_template_from_dict(self): + """Test creating template from dictionary.""" + data = { + "template_id": "restored", + "template": "Restored {item}", + "description": "Restored template", + "variables": ["item"], + "tags": ["restore"], + "metadata": {}, + "usage_count": 5, + "success_rate": 0.8, + } + + template = PromptTemplate.from_dict(data) + assert template.template_id == "restored" + assert template.usage_count == 5 + assert template.success_rate == 0.8 + + +class TestSemanticIndex: + """Tests for SemanticIndex class.""" + + def test_build_index(self): + """Test building a semantic index.""" + index = SemanticIndex() + prompts = [ + PromptTemplate( + template_id=f"template_{i}", + template=f"Template {i} for testing", + tags=["test", f"tag{i}"], + ) + for i in range(5) + ] + + index.build_index(prompts) + assert index._built is True + assert len(index._index) == 5 + + def test_similarity_search(self): + """Test semantic similarity search.""" + index = SemanticIndex() + prompts = [ + PromptTemplate( + template_id="python_code", + template="Generate Python code for {task}", + description="Python code generation", + tags=["code", "python"], + ), + PromptTemplate( + template_id="javascript_code", + template="Generate JavaScript code for {task}", + description="JavaScript code generation", + tags=["code", "javascript"], + ), + PromptTemplate( + template_id="debug_error", + template="Debug this error: {error}", + description="Error debugging", + tags=["debug", "error"], + ), + ] + + index.build_index(prompts) + + # Search for Python-related templates + results = index.similarity_search("python code generation", top_k=2) + assert len(results) <= 2 + assert results[0][0] == "python_code" + + def test_similarity_search_empty_index(self): + """Test search on unbuilt index returns empty.""" + index = SemanticIndex() + results = index.similarity_search("test query") + assert results == [] + + def test_keyword_extraction(self): + """Test keyword extraction removes stopwords.""" + index = SemanticIndex() + text = "The quick brown fox jumps over the lazy dog" + keywords = index._extract_keywords(text) + + # Should not contain common stopwords + assert "the" not in keywords + + # Should contain meaningful words (at least some of them) + assert len(keywords) > 0 + assert any(word in keywords for word in ["quick", "brown", "fox", "jumps", "lazy", "dog"]) + + +class TestContextPromptRepository: + """Tests for ContextPromptRepository class.""" + + def test_store_prompt(self): + """Test storing a prompt template.""" + repo = ContextPromptRepository() + + template = repo.store_prompt( + template_id="test_store", + prompt_template="Test {value}", + description="Test storage", + tags=["test"], + ) + + assert template.template_id == "test_store" + assert "test_store" in repo.list_templates() + + def test_store_prompt_auto_detect_variables(self): + """Test automatic variable detection when storing.""" + repo = ContextPromptRepository() + + template = repo.store_prompt( + template_id="auto_vars", + prompt_template="Hello {name}, you are {age} years old", + ) + + assert "name" in template.variables + assert "age" in template.variables + + def test_retrieve_prompts(self): + """Test retrieving prompts by query.""" + repo = ContextPromptRepository() + repo.initialize_with_defaults() + + results = repo.retrieve_prompts("python code", top_k=3) + assert len(results) > 0 + assert any("python" in t.tags for t in results) + + def test_get_context_for_task(self): + """Test getting context for a specific task type.""" + repo = ContextPromptRepository() + repo.initialize_with_defaults() + + context = repo.get_context_for_task( + "debugging", + {"language": "python"}, + top_k=2, + ) + + assert len(context) > 0 + assert "template_id" in context[0] + assert "template" in context[0] + + def test_get_template(self): + """Test getting a specific template by ID.""" + repo = ContextPromptRepository() + repo.store_prompt( + template_id="specific", + prompt_template="Specific {test}", + ) + + template = repo.get_template("specific") + assert template is not None + assert template.template_id == "specific" + + # Non-existent template + missing = repo.get_template("nonexistent") + assert missing is None + + def test_update_success_rate(self): + """Test updating template success rate.""" + repo = ContextPromptRepository() + repo.store_prompt( + template_id="rate_test", + prompt_template="Test {x}", + ) + + # Initial rate should be 0.5 + template = repo.get_template("rate_test") + assert template.success_rate == 0.5 + + # Update with success + repo.update_success_rate("rate_test", True) + template = repo.get_template("rate_test") + assert template.success_rate > 0.5 + + # Update with failure + repo.update_success_rate("rate_test", False) + template = repo.get_template("rate_test") + assert template.success_rate < 0.6 # EMA should reduce it + + def test_list_templates(self): + """Test listing templates with optional tag filter.""" + repo = ContextPromptRepository() + repo.store_prompt( + template_id="tagged1", + prompt_template="Test 1", + tags=["alpha", "beta"], + ) + repo.store_prompt( + template_id="tagged2", + prompt_template="Test 2", + tags=["beta", "gamma"], + ) + repo.store_prompt( + template_id="untagged", + prompt_template="Test 3", + ) + + # All templates + all_templates = repo.list_templates() + assert len(all_templates) == 3 + + # Filtered by tag + beta_templates = repo.list_templates(tags_filter=["beta"]) + assert len(beta_templates) == 2 + assert "tagged1" in beta_templates + assert "tagged2" in beta_templates + + def test_export_import_templates(self): + """Test exporting and importing templates.""" + repo1 = ContextPromptRepository() + repo1.store_prompt( + template_id="export_test", + prompt_template="Export {this}", + tags=["export"], + ) + + # Export + exported = repo1.export_templates() + assert "export_test" in exported + + # Import to new repository + repo2 = ContextPromptRepository() + count = repo2.import_templates(exported) + assert count == 1 + + # Verify imported template + template = repo2.get_template("export_test") + assert template is not None + assert template.template == "Export {this}" + + def test_initialize_with_defaults(self): + """Test initialization with default templates.""" + repo = ContextPromptRepository() + repo.initialize_with_defaults() + + templates = repo.list_templates() + assert len(templates) >= 5 # At least 5 default templates + + # Check specific default templates exist + assert "code_generation_python" in templates + assert "debug_error" in templates + assert "code_review" in templates + + def test_singleton_get_repository(self): + """Test singleton pattern for get_repository.""" + repo1 = get_repository() + repo2 = get_repository() + + assert repo1 is repo2 # Same instance + + +class TestIntegration: + """Integration tests for context repository.""" + + def test_full_workflow(self): + """Test complete workflow: store, search, retrieve, render.""" + repo = ContextPromptRepository() + + # Store custom template + repo.store_prompt( + template_id="custom_analysis", + prompt_template="Analyze this {data_type} data: {data}\n\nGoals: {goals}", + description="Custom data analysis template", + tags=["analysis", "custom", "data"], + ) + + # Search for it + results = repo.retrieve_prompts("data analysis custom", top_k=5) + assert any(t.template_id == "custom_analysis" for t in results) + + # Get context for task + context = repo.get_context_for_task("analysis", {"data_type": "numeric"}) + assert len(context) > 0 + + # Render template + template = repo.get_template("custom_analysis") + rendered = template.render( + data_type="numeric", + data="[1, 2, 3, 4, 5]", + goals="Find patterns and outliers", + ) + + assert "numeric" in rendered + assert "[1, 2, 3, 4, 5]" in rendered + assert "Find patterns and outliers" in rendered + + def test_success_rate_tracking(self): + """Test tracking success rates through multiple updates.""" + repo = ContextPromptRepository() + repo.store_prompt( + template_id="tracked", + prompt_template="Track {this}", + ) + + # Simulate multiple uses with varying success + for _ in range(5): + repo.update_success_rate("tracked", True) + for _ in range(2): + repo.update_success_rate("tracked", False) + + template = repo.get_template("tracked") + # Should be above 0.5 due to more successes + assert template.success_rate > 0.5 + assert template.success_rate < 1.0 diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 9944a87..786c648 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -278,3 +278,40 @@ def test_aggregate_mixed_with_and_without_scores(metrics_file: Path) -> None: stats = result["calc"] assert stats["total_executions"] == 3 assert stats["avg_success_score"] == 0.7 # Only 2 events with scores: (0.8 + 0.6) / 2 + + +def test_get_execution_events_filters_by_trajectory(metrics_file: Path) -> None: + metrics.log_execution("a", "v1", 1.0, True, trajectory_id="t1", step_number=1, metrics_file=metrics_file) + metrics.log_execution("a", "v1", 2.0, True, trajectory_id="t2", step_number=1, metrics_file=metrics_file) + events = metrics.get_execution_events(metrics_file=metrics_file, trajectory_id="t1") + assert len(events) == 1 + assert events[0]["trajectory_id"] == "t1" + + +def test_build_agent_history_uses_explicit_and_derived_scores(metrics_file: Path) -> None: + metrics.log_execution( + plugin_name="agent_planner", + version="v1", + execution_time_ms=10.0, + success=True, + task_type="analysis", + success_score=0.9, + metrics_file=metrics_file, + ) + metrics.log_execution( + plugin_name="agent_coder", + version="v1", + execution_time_ms=20.0, + success=False, + task_type="python", + metrics_file=metrics_file, + ) + + history = metrics.build_agent_history(metrics_file=metrics_file) + assert len(history) == 2 + assert history[0]["agent_id"] == "agent_planner" + assert history[0]["task_type"] == "analysis" + assert history[0]["success_score"] == 0.9 + assert history[1]["agent_id"] == "agent_coder" + assert history[1]["task_type"] == "python" + assert history[1]["success_score"] == 0.0 diff --git a/tests/test_reflection.py b/tests/test_reflection.py new file mode 100644 index 0000000..23978ce --- /dev/null +++ b/tests/test_reflection.py @@ -0,0 +1,461 @@ +"""Tests for the Tool Reflection Cycle (Sprint 2 - Task 1.2.1).""" + +import pytest +from core.reflection import ( + CorrectionGenerator, + CorrectionResult, + ErrorAnalyzer, + ErrorCategory, + ErrorAnalysis, + ReflectionLoop, + get_reflection_loop, +) + + +class TestErrorCategory: + """Tests for ErrorCategory enum.""" + + def test_error_categories_exist(self): + """Test that all expected error categories exist.""" + assert ErrorCategory.SYNTAX_ERROR is not None + assert ErrorCategory.RUNTIME_ERROR is not None + assert ErrorCategory.TIMEOUT_ERROR is not None + assert ErrorCategory.RESOURCE_ERROR is not None + assert ErrorCategory.PERMISSION_ERROR is not None + assert ErrorCategory.NOT_FOUND_ERROR is not None + assert ErrorCategory.VALIDATION_ERROR is not None + assert ErrorCategory.NETWORK_ERROR is not None + assert ErrorCategory.UNKNOWN_ERROR is not None + + +class TestErrorAnalysis: + """Tests for ErrorAnalysis dataclass.""" + + def test_error_analysis_creation(self): + """Test creating an ErrorAnalysis instance.""" + analysis = ErrorAnalysis( + error_category=ErrorCategory.RUNTIME_ERROR, + error_message="Test error", + error_type="TestError", + root_cause="Test root cause", + confidence=0.8, + suggestions=["Fix this", "Fix that"], + ) + + assert analysis.error_category == ErrorCategory.RUNTIME_ERROR + assert analysis.error_message == "Test error" + assert len(analysis.suggestions) == 2 + + def test_error_analysis_to_dict(self): + """Test converting ErrorAnalysis to dictionary.""" + analysis = ErrorAnalysis( + error_category=ErrorCategory.SYNTAX_ERROR, + error_message="Syntax issue", + error_type="SyntaxError", + stack_trace="Traceback...", + confidence=0.9, + ) + + data = analysis.to_dict() + assert data["error_category"] == "SYNTAX_ERROR" + assert data["error_message"] == "Syntax issue" + assert data["error_type"] == "SyntaxError" + assert data["stack_trace"] == "Traceback..." + assert data["confidence"] == 0.9 + + +class TestCorrectionResult: + """Tests for CorrectionResult dataclass.""" + + def test_correction_result_success(self): + """Test successful correction result.""" + result = CorrectionResult( + success=True, + corrected_call={"name": "test", "input": {"x": 1}}, + explanation="Fixed the issue", + confidence=0.85, + ) + + assert result.success is True + assert result.corrected_call is not None + assert result.requires_human_review is False + + def test_correction_result_failure(self): + """Test failed correction result.""" + result = CorrectionResult( + success=False, + explanation="Cannot fix automatically", + requires_human_review=True, + alternative_approaches=["Try manual fix", "Ask for help"], + ) + + assert result.success is False + assert result.corrected_call is None + assert result.requires_human_review is True + assert len(result.alternative_approaches) == 2 + + def test_correction_result_to_dict(self): + """Test converting CorrectionResult to dictionary.""" + result = CorrectionResult( + success=True, + corrected_call={"name": "test"}, + explanation="Success", + confidence=0.7, + ) + + data = result.to_dict() + assert data["success"] is True + assert data["corrected_call"] == {"name": "test"} + assert data["explanation"] == "Success" + + +class TestErrorAnalyzer: + """Tests for ErrorAnalyzer class.""" + + def test_analyzer_initialization(self): + """Test ErrorAnalyzer initialization.""" + analyzer = ErrorAnalyzer() + assert analyzer._error_patterns is not None + assert len(analyzer._error_patterns) > 0 + + def test_analyze_runtime_error(self): + """Test analyzing a runtime error (TypeError).""" + analyzer = ErrorAnalyzer() + + tool_call = {"name": "run_code", "input": {"code": "x = None + 1"}} + result = { + "error": "unsupported operand type(s)", + "error_type": "TypeError", + } + + analysis = analyzer.analyze_error(tool_call, result) + + assert analysis.error_category == ErrorCategory.RUNTIME_ERROR + assert analysis.confidence > 0 + assert len(analysis.suggestions) > 0 + + def test_analyze_syntax_error(self): + """Test analyzing a syntax error.""" + analyzer = ErrorAnalyzer() + + tool_call = {"name": "run_code", "input": {"code": "if True print('hi')"}} + result = { + "error": "invalid syntax", + "error_type": "SyntaxError", + } + + analysis = analyzer.analyze_error(tool_call, result) + + assert analysis.error_category == ErrorCategory.SYNTAX_ERROR + assert "syntax" in analysis.error_message.lower() + + def test_analyze_not_found_error(self): + """Test analyzing a file not found error.""" + analyzer = ErrorAnalyzer() + + tool_call = {"name": "read_file", "input": {"path": "/missing.txt"}} + result = { + "error": "[Errno 2] No such file or directory", + "error_type": "FileNotFoundError", + } + + analysis = analyzer.analyze_error(tool_call, result) + + assert analysis.error_category == ErrorCategory.NOT_FOUND_ERROR + + def test_analyze_permission_error(self): + """Test analyzing a permission error.""" + analyzer = ErrorAnalyzer() + + tool_call = {"name": "write_file", "input": {"path": "/root/test.txt"}} + result = { + "error": "[Errno 13] Permission denied", + "error_type": "PermissionError", + } + + analysis = analyzer.analyze_error(tool_call, result) + + assert analysis.error_category == ErrorCategory.PERMISSION_ERROR + + def test_analyze_timeout_error(self): + """Test analyzing a timeout error.""" + analyzer = ErrorAnalyzer() + + tool_call = {"name": "long_running_task", "input": {}} + result = { + "error": "Execution timed out after 30 seconds", + "error_type": "TimeoutError", + } + + analysis = analyzer.analyze_error(tool_call, result) + + assert analysis.error_category == ErrorCategory.TIMEOUT_ERROR + + def test_analyze_with_traceback(self): + """Test analyzing error with full traceback.""" + analyzer = ErrorAnalyzer() + + tool_call = {"name": "test", "input": {}} + result = {"error": "KeyError: 'missing'", "error_type": "KeyError"} + traceback_str = """Traceback (most recent call last): + File "test.py", line 10, in + data['missing'] +KeyError: 'missing'""" + + analysis = analyzer.analyze_error(tool_call, result, traceback_str) + + assert analysis.stack_trace is not None + assert "KeyError" in analysis.stack_trace + + def test_categorize_error_quick(self): + """Test quick error categorization without full analysis.""" + analyzer = ErrorAnalyzer() + + category = analyzer.categorize_error("ValueError", "Invalid value provided") + assert category == ErrorCategory.RUNTIME_ERROR + + category = analyzer.categorize_error("ModuleNotFoundError", "No module named 'xyz'") + assert category == ErrorCategory.NOT_FOUND_ERROR + + def test_analyze_unknown_error(self): + """Test analyzing an unknown error type.""" + analyzer = ErrorAnalyzer() + + tool_call = {"name": "test", "input": {}} + result = { + "error": "Some weird error", + "error_type": "WeirdCustomError", + } + + analysis = analyzer.analyze_error(tool_call, result) + + assert analysis.error_category == ErrorCategory.UNKNOWN_ERROR + assert analysis.confidence == 0.0 + + +class TestCorrectionGenerator: + """Tests for CorrectionGenerator class.""" + + def test_generator_initialization(self): + """Test CorrectionGenerator initialization.""" + generator = CorrectionGenerator() + assert generator._analyzer is not None + + def test_generate_runtime_error_correction(self): + """Test generating correction for runtime error.""" + analyzer = ErrorAnalyzer() + generator = CorrectionGenerator() + + tool_call = {"name": "run_code", "input": {"code": "result = data['key']"}} + result = {"error": "'NoneType' object is not subscriptable", "error_type": "TypeError"} + + analysis = analyzer.analyze_error(tool_call, result) + correction = generator.generate_correction(analysis, tool_call) + + assert correction.success is True + assert correction.corrected_call is not None + assert "try:" in correction.corrected_call["input"]["code"] + assert "except" in correction.corrected_call["input"]["code"] + + def test_generate_syntax_error_correction(self): + """Test generating correction for syntax error.""" + analyzer = ErrorAnalyzer() + generator = CorrectionGenerator() + + tool_call = {"name": "run_code", "input": {"code": "if True x = 1"}} + result = {"error": "invalid syntax", "error_type": "SyntaxError"} + + analysis = analyzer.analyze_error(tool_call, result) + correction = generator.generate_correction(analysis, tool_call) + + # Syntax errors can't be auto-fixed + assert correction.success is False + assert "syntax" in correction.explanation.lower() + assert len(correction.alternative_approaches) > 0 + + def test_generate_not_found_correction(self): + """Test generating correction for not found error.""" + analyzer = ErrorAnalyzer() + generator = CorrectionGenerator() + + tool_call = {"name": "my_plugin", "input": {}} + result = {"error": "Plugin not found", "error_type": "NotFoundError"} + + analysis = analyzer.analyze_error(tool_call, result) + correction = generator.generate_correction(analysis, tool_call) + + assert correction.success is False + assert "not found" in correction.explanation.lower() + + def test_validate_correction(self): + """Test validating a proposed correction.""" + generator = CorrectionGenerator() + + # Valid correction + valid_call = {"name": "test", "input": {"x": 1}} + assert generator.validate_correction(valid_call) is True + + # Invalid corrections + assert generator.validate_correction(None) is False + assert generator.validate_correction({}) is False + assert generator.validate_correction({"name": "test"}) is False + assert generator.validate_correction({"name": "test", "input": "not_a_dict"}) is False + + def test_wrap_with_error_handling(self): + """Test wrapping code with error handling.""" + generator = CorrectionGenerator() + + code = "result = 1 / 0" + wrapped = generator._wrap_with_error_handling(code) + + assert "try:" in wrapped + assert "except Exception" in wrapped + assert "traceback" in wrapped + assert code in wrapped + + +class TestReflectionLoop: + """Tests for ReflectionLoop class.""" + + def test_loop_initialization(self): + """Test ReflectionLoop initialization.""" + loop = ReflectionLoop() + assert loop._analyzer is not None + assert loop._correction_generator is not None + assert loop._max_history == 100 + + def test_run_reflection_cycle(self): + """Test running a complete reflection cycle.""" + loop = ReflectionLoop() + + tool_call = {"name": "test_tool", "input": {"code": "x = None + 1"}} + result = {"error": "unsupported operand type", "error_type": "TypeError"} + + reflection_result = loop.run_reflection_cycle(tool_call, result) + + assert "analysis" in reflection_result + assert "correction" in reflection_result + assert "recommendation" in reflection_result + + assert reflection_result["analysis"]["error_type"] == "TypeError" + + def test_record_reflection_history(self): + """Test that reflection events are recorded in history.""" + loop = ReflectionLoop(max_history=10) + + # Run multiple reflection cycles + for i in range(15): + tool_call = {"name": f"tool_{i}", "input": {}} + result = {"error": f"Error {i}", "error_type": "RuntimeError"} + loop.run_reflection_cycle(tool_call, result) + + # History should be limited to max_history + history = loop.get_reflection_history(limit=100) + assert len(history) == 10 # Limited by max_history + + def test_get_tool_success_rate(self): + """Test tracking tool success rates.""" + loop = ReflectionLoop() + + # Initial rate should be 0.5 (default prior) + assert loop.get_tool_success_rate("new_tool") == 0.5 + + # Simulate successes and failures through reflection cycles + for _ in range(3): + tool_call = {"name": "tested_tool", "input": {"code": "pass"}} + # Force successful correction + result = {"error": "test", "error_type": "TypeError"} + loop.run_reflection_cycle(tool_call, result) + + # Rate should reflect history + rate = loop.get_tool_success_rate("tested_tool") + assert 0.0 <= rate <= 1.0 + + def test_clear_history(self): + """Test clearing reflection history.""" + loop = ReflectionLoop() + + # Add some history + tool_call = {"name": "test", "input": {}} + result = {"error": "test", "error_type": "Error"} + loop.run_reflection_cycle(tool_call, result) + + assert len(loop.get_reflection_history()) > 0 + + # Clear history + loop.clear_history() + + assert len(loop.get_reflection_history()) == 0 + + def test_log_reflection_event(self): + """Test logging reflection events.""" + loop = ReflectionLoop() + + event_data = {"tool_name": "test", "event": "reflection_test"} + loop.log_reflection_event(event_data) + + # Should not raise any exceptions + + +class TestSingleton: + """Tests for singleton pattern.""" + + def test_get_reflection_loop_singleton(self): + """Test that get_reflection_loop returns same instance.""" + loop1 = get_reflection_loop() + loop2 = get_reflection_loop() + + assert loop1 is loop2 + + +class TestIntegration: + """Integration tests for reflection cycle.""" + + def test_full_reflection_workflow(self): + """Test complete workflow: error -> analysis -> correction.""" + loop = get_reflection_loop() + + # Simulate a realistic error scenario + tool_call = { + "name": "run_plugin", + "input": { + "code": "data = None\nresult = data['key']" + }, + } + result = { + "error": "'NoneType' object is not subscriptable", + "error_type": "TypeError", + } + traceback_str = "Traceback...\nTypeError: 'NoneType' object is not subscriptable" + + # Run reflection cycle + reflection = loop.run_reflection_cycle(tool_call, result, traceback_str) + + # Verify all components worked together + assert reflection["analysis"]["error_category"] == "RUNTIME_ERROR" + assert reflection["correction"]["success"] is True + assert "try:" in reflection["correction"]["corrected_call"]["input"]["code"] + + # Check recommendation + assert "correction" in reflection["recommendation"].lower() or \ + "apply" in reflection["recommendation"].lower() + + def test_multiple_error_types(self): + """Test handling various error types in sequence.""" + loop = ReflectionLoop() + + error_scenarios = [ + ({"name": "t1", "input": {}}, {"error": "invalid syntax", "error_type": "SyntaxError"}), + ({"name": "t2", "input": {}}, {"error": "file not found", "error_type": "FileNotFoundError"}), + ({"name": "t3", "input": {}}, {"error": "permission denied", "error_type": "PermissionError"}), + ({"name": "t4", "input": {}}, {"error": "connection refused", "error_type": "ConnectionError"}), + ] + + for tool_call, result in error_scenarios: + reflection = loop.run_reflection_cycle(tool_call, result) + assert "analysis" in reflection + assert "correction" in reflection + + # Check that all were recorded + history = loop.get_reflection_history(limit=10) + assert len(history) == len(error_scenarios)