diff --git a/codeframe/persistence/schema_manager.py b/codeframe/persistence/schema_manager.py index 19e86d5c..500c65ed 100644 --- a/codeframe/persistence/schema_manager.py +++ b/codeframe/persistence/schema_manager.py @@ -28,46 +28,22 @@ def __init__(self, conn: sqlite3.Connection): def create_schema(self) -> None: """Create all database tables and indexes. - Creates the complete v1.0 flattened schema with all migrations applied. + Creates the minimal control-plane schema (auth, api keys, audit log, + interactive sessions). v2 domain data lives in per-workspace DBs. This method is idempotent - safe to call multiple times. """ cursor = self.conn.cursor() - # Authentication tables + # Authentication tables (users, accounts, sessions, verification, api_keys) self._create_auth_tables(cursor) - # Core project tables - self._create_project_tables(cursor) - - # Issue and task tables - self._create_issue_task_tables(cursor) - - # Agent management tables - self._create_agent_tables(cursor) - - # Blocker management tables - self._create_blocker_tables(cursor) - - # Quality and testing tables - self._create_quality_tables(cursor) - - # Memory and context tables - self._create_memory_context_tables(cursor) - - # Checkpoint and git tracking tables - self._create_checkpoint_git_tables(cursor) - - # Metrics and audit tables - self._create_metrics_audit_tables(cursor) + # Audit log table + self._create_audit_log_table(cursor) # Interactive session tables self._create_interactive_session_tables(cursor) - # Apply schema migrations for existing databases BEFORE creating indexes - # (indexes may reference columns added by migrations) - self._apply_migrations(cursor) - - # Create all indexes (after migrations so all columns exist) + # Create indexes self._create_indexes(cursor) self.conn.commit() @@ -75,161 +51,6 @@ def create_schema(self) -> None: # Ensure default admin user exists self._ensure_default_admin_user() - def _apply_migrations(self, cursor: sqlite3.Cursor) -> None: - """Apply schema migrations for existing databases. - - Handles adding new columns to existing tables. - These are idempotent - safe to run multiple times. - """ - # Migration: Add depends_on column to issues table (cf-207) - self._add_column_if_not_exists( - cursor, "issues", "depends_on", "TEXT" - ) - - # Migration: Add missing columns to tasks table for older databases - # Core columns that may be missing - self._add_column_if_not_exists( - cursor, "tasks", "project_id", "INTEGER" - ) - self._add_column_if_not_exists( - cursor, "tasks", "issue_id", "INTEGER" - ) - self._add_column_if_not_exists( - cursor, "tasks", "parent_issue_number", "TEXT" - ) - self._add_column_if_not_exists( - cursor, "tasks", "task_number", "TEXT" - ) - self._add_column_if_not_exists( - cursor, "tasks", "description", "TEXT" - ) - self._add_column_if_not_exists( - cursor, "tasks", "status", "TEXT", "'pending'" - ) - self._add_column_if_not_exists( - cursor, "tasks", "priority", "INTEGER", "0" - ) - self._add_column_if_not_exists( - cursor, "tasks", "workflow_step", "INTEGER" - ) - self._add_column_if_not_exists( - cursor, "tasks", "depends_on", "TEXT" - ) - self._add_column_if_not_exists( - cursor, "tasks", "created_at", "TIMESTAMP", "CURRENT_TIMESTAMP" - ) - self._add_column_if_not_exists( - cursor, "tasks", "completed_at", "TIMESTAMP" - ) - self._add_column_if_not_exists( - cursor, "tasks", "assigned_to", "TEXT" - ) - self._add_column_if_not_exists( - cursor, "tasks", "can_parallelize", "BOOLEAN", "FALSE" - ) - self._add_column_if_not_exists( - cursor, "tasks", "requires_mcp", "BOOLEAN", "FALSE" - ) - self._add_column_if_not_exists( - cursor, "tasks", "estimated_tokens", "INTEGER" - ) - self._add_column_if_not_exists( - cursor, "tasks", "actual_tokens", "INTEGER" - ) - self._add_column_if_not_exists( - cursor, "tasks", "commit_sha", "TEXT" - ) - self._add_column_if_not_exists( - cursor, "tasks", "quality_gate_status", "TEXT", "'pending'" - ) - self._add_column_if_not_exists( - cursor, "tasks", "quality_gate_failures", "JSON" - ) - self._add_column_if_not_exists( - cursor, "tasks", "requires_human_approval", "BOOLEAN", "FALSE" - ) - - # Migration: Add effort estimation columns to tasks table (Phase 1) - self._add_column_if_not_exists( - cursor, "tasks", "estimated_hours", "REAL" - ) - self._add_column_if_not_exists( - cursor, "tasks", "complexity_score", "INTEGER" - ) - self._add_column_if_not_exists( - cursor, "tasks", "uncertainty_level", "TEXT" - ) - self._add_column_if_not_exists( - cursor, "tasks", "resource_requirements", "TEXT" - ) - - # Migration: Add supervisor intervention context to tasks table - self._add_column_if_not_exists( - cursor, "tasks", "intervention_context", "JSON" - ) - - def _add_column_if_not_exists( - self, - cursor: sqlite3.Cursor, - table_name: str, - column_name: str, - column_type: str, - default_value: str = None, - ) -> None: - """Add a column to a table if it doesn't exist. - - Args: - cursor: SQLite cursor - table_name: Table to modify - column_name: Column to add - column_type: SQLite column type - default_value: Optional default value for the column - - Raises: - ValueError: If table_name, column_name, or column_type contain invalid characters - """ - # SECURITY: Validate identifiers to prevent SQL injection. - # Only alphanumeric + underscore allowed (standard SQL identifier rules). - import re - identifier_pattern = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$') - for name, value in [("table_name", table_name), ("column_name", column_name), ("column_type", column_type)]: - if not identifier_pattern.match(value): - raise ValueError(f"Invalid SQL identifier for {name}: {value}") - - # SECURITY: Validate default_value to only allow safe SQL literals. - # Allowed: NULL, TRUE, FALSE, CURRENT_TIMESTAMP, integers, floats, - # or single-quoted strings (with no embedded quotes). - if default_value is not None: - safe_literal_pattern = re.compile( - r"^(NULL|TRUE|FALSE|CURRENT_TIMESTAMP|" # SQL keywords - r"-?\d+|" # Integers (including negative) - r"-?\d+\.\d+|" # Floats - r"'[^']*')$", # Single-quoted strings (no embedded quotes) - re.IGNORECASE - ) - if not safe_literal_pattern.match(default_value): - raise ValueError( - f"Invalid SQL literal for default_value: {default_value}. " - "Only NULL, TRUE, FALSE, CURRENT_TIMESTAMP, numbers, or " - "single-quoted strings (without embedded quotes) are allowed." - ) - - # Check if column exists - cursor.execute(f"PRAGMA table_info({table_name})") - columns = {row[1] for row in cursor.fetchall()} - - if column_name not in columns: - # Add the column - if default_value is not None: - cursor.execute( - f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type} DEFAULT {default_value}" - ) - else: - cursor.execute( - f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_type}" - ) - logger.info(f"Added column {column_name} to {table_name}") - def _create_auth_tables(self, cursor: sqlite3.Cursor) -> None: """Create authentication tables (fastapi-users compatible).""" cursor.execute( @@ -338,503 +159,6 @@ def _create_auth_tables(self, cursor: sqlite3.Cursor) -> None: """ ) - def _create_project_tables(self, cursor: sqlite3.Cursor) -> None: - """Create project and project_users tables.""" - # Projects table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS projects ( - id INTEGER PRIMARY KEY, - name TEXT NOT NULL, - description TEXT NOT NULL, - user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, - source_type TEXT CHECK(source_type IN ('git_remote', 'local_path', 'upload', 'empty')) DEFAULT 'empty', - source_location TEXT, - source_branch TEXT DEFAULT 'main', - workspace_path TEXT NOT NULL, - git_initialized BOOLEAN DEFAULT FALSE, - current_commit TEXT, - status TEXT CHECK(status IN ('init', 'planning', 'running', 'active', 'paused', 'completed')), - phase TEXT CHECK(phase IN ('discovery', 'planning', 'active', 'review', 'complete')) DEFAULT 'discovery', - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - paused_at TIMESTAMP NULL, - config JSON - ) - """ - ) - - # Project users table (authorization) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS project_users ( - project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, - user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, - role TEXT NOT NULL CHECK(role IN ('owner', 'collaborator', 'viewer')), - granted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - PRIMARY KEY (project_id, user_id) - ) - """ - ) - - def _create_issue_task_tables(self, cursor: sqlite3.Cursor) -> None: - """Create issues and tasks tables.""" - # Issues table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS issues ( - id INTEGER PRIMARY KEY, - project_id INTEGER REFERENCES projects(id), - issue_number TEXT NOT NULL, - title TEXT NOT NULL, - description TEXT, - status TEXT CHECK(status IN ('pending', 'in_progress', 'completed', 'failed')), - priority INTEGER CHECK(priority BETWEEN 0 AND 4), - workflow_step INTEGER, - depends_on TEXT, - created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, - completed_at TIMESTAMP, - UNIQUE(project_id, issue_number) - ) - """ - ) - - # Tasks table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS tasks ( - id INTEGER PRIMARY KEY, - project_id INTEGER REFERENCES projects(id), - issue_id INTEGER REFERENCES issues(id), - task_number TEXT, - parent_issue_number TEXT, - title TEXT NOT NULL, - description TEXT, - status TEXT CHECK(status IN ('pending', 'assigned', 'in_progress', 'blocked', 'completed', 'failed')), - assigned_to TEXT, - depends_on TEXT, - can_parallelize BOOLEAN DEFAULT FALSE, - priority INTEGER CHECK(priority BETWEEN 0 AND 4), - workflow_step INTEGER, - requires_mcp BOOLEAN DEFAULT FALSE, - estimated_tokens INTEGER, - actual_tokens INTEGER, - commit_sha TEXT, - created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, - completed_at TIMESTAMP, - quality_gate_status TEXT CHECK(quality_gate_status IN ('pending', 'running', 'passed', 'failed')) DEFAULT 'pending', - quality_gate_failures JSON, - requires_human_approval BOOLEAN DEFAULT FALSE, - -- Effort estimation fields (Phase 1) - estimated_hours REAL, - complexity_score INTEGER CHECK(complexity_score BETWEEN 1 AND 5), - uncertainty_level TEXT CHECK(uncertainty_level IN ('low', 'medium', 'high')), - resource_requirements TEXT, - -- Supervisor intervention context (JSON for flexible structure) - intervention_context JSON - ) - """ - ) - - # Task dependencies junction table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS task_dependencies ( - id INTEGER PRIMARY KEY, - task_id INTEGER NOT NULL, - depends_on_task_id INTEGER NOT NULL, - FOREIGN KEY (task_id) REFERENCES tasks(id), - FOREIGN KEY (depends_on_task_id) REFERENCES tasks(id), - UNIQUE(task_id, depends_on_task_id) - ) - """ - ) - - def _create_agent_tables(self, cursor: sqlite3.Cursor) -> None: - """Create agent and project_agents tables.""" - # Agents table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS agents ( - id TEXT PRIMARY KEY, - type TEXT NOT NULL, - project_id INTEGER REFERENCES projects(id) ON DELETE CASCADE, - provider TEXT, - maturity_level TEXT CHECK(maturity_level IN ('directive', 'coaching', 'supporting', 'delegating')), - status TEXT CHECK(status IN ('idle', 'working', 'blocked', 'offline')), - current_task_id INTEGER REFERENCES tasks(id), - last_heartbeat TIMESTAMP, - metrics JSON, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - # Project-Agent junction table (many-to-many) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS project_agents ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, - agent_id TEXT NOT NULL REFERENCES agents(id) ON DELETE CASCADE, - role TEXT NOT NULL, - assigned_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - unassigned_at TIMESTAMP, - is_active BOOLEAN DEFAULT TRUE, - CHECK(unassigned_at IS NULL OR unassigned_at >= assigned_at) - ) - """ - ) - - def _create_blocker_tables(self, cursor: sqlite3.Cursor) -> None: - """Create blockers table.""" - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS blockers ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - agent_id TEXT NOT NULL, - project_id INTEGER NOT NULL, - task_id INTEGER, - blocker_type TEXT NOT NULL CHECK(blocker_type IN ('SYNC', 'ASYNC')), - question TEXT NOT NULL, - answer TEXT, - status TEXT NOT NULL DEFAULT 'PENDING' CHECK(status IN ('PENDING', 'RESOLVED', 'EXPIRED')), - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - resolved_at TIMESTAMP, - FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE, - FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE - ) - """ - ) - - def _create_quality_tables(self, cursor: sqlite3.Cursor) -> None: - """Create quality, testing, and code review tables.""" - # Lint results table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS lint_results ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - task_id INTEGER NOT NULL, - linter TEXT NOT NULL CHECK(linter IN ('ruff', 'eslint', 'other')), - error_count INTEGER NOT NULL DEFAULT 0, - warning_count INTEGER NOT NULL DEFAULT 0, - files_linted INTEGER NOT NULL DEFAULT 0, - output TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (task_id) REFERENCES tasks(id) ON DELETE CASCADE - ) - """ - ) - - # Test results table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS test_results ( - id INTEGER PRIMARY KEY, - task_id INTEGER NOT NULL REFERENCES tasks(id), - status TEXT NOT NULL CHECK(status IN ('passed', 'failed', 'error', 'timeout', 'no_tests')), - passed INTEGER DEFAULT 0, - failed INTEGER DEFAULT 0, - errors INTEGER DEFAULT 0, - skipped INTEGER DEFAULT 0, - duration REAL DEFAULT 0.0, - output TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - # Correction attempts table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS correction_attempts ( - id INTEGER PRIMARY KEY, - task_id INTEGER NOT NULL REFERENCES tasks(id), - attempt_number INTEGER NOT NULL CHECK(attempt_number BETWEEN 1 AND 3), - error_analysis TEXT NOT NULL, - fix_description TEXT NOT NULL, - code_changes TEXT DEFAULT '', - test_result_id INTEGER REFERENCES test_results(id), - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - # Code reviews table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS code_reviews ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - task_id INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, - agent_id TEXT NOT NULL, - project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, - file_path TEXT NOT NULL, - line_number INTEGER, - severity TEXT NOT NULL CHECK(severity IN ('critical', 'high', 'medium', 'low', 'info')), - category TEXT NOT NULL CHECK(category IN ('security', 'performance', 'quality', 'maintainability', 'style')), - message TEXT NOT NULL, - recommendation TEXT, - code_snippet TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - # Task evidence table (for evidence-based quality enforcement) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS task_evidence ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - task_id INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, - agent_id TEXT NOT NULL, - language TEXT NOT NULL, - framework TEXT, - - -- Test results - total_tests INTEGER NOT NULL, - passed_tests INTEGER NOT NULL, - failed_tests INTEGER NOT NULL, - skipped_tests INTEGER NOT NULL, - pass_rate REAL NOT NULL, - coverage REAL, - test_output TEXT NOT NULL, - - -- Skip violations - skip_violations_count INTEGER NOT NULL DEFAULT 0, - skip_violations_json TEXT, - skip_check_passed BOOLEAN NOT NULL, - - -- Quality metrics - quality_metrics_json TEXT NOT NULL, - - -- Verification status - verified BOOLEAN NOT NULL, - verification_errors TEXT, - - -- Metadata - timestamp TEXT NOT NULL, - task_description TEXT NOT NULL, - - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - def _create_memory_context_tables(self, cursor: sqlite3.Cursor) -> None: - """Create memory and context management tables.""" - # Memory table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS memory ( - id INTEGER PRIMARY KEY, - project_id INTEGER REFERENCES projects(id), - category TEXT CHECK(category IN ('pattern', 'decision', 'gotcha', 'preference', 'conversation', 'discovery_state', 'discovery_answers', 'prd')), - key TEXT, - value TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - # Add unique constraint on memory table to prevent duplicate keys - # First, check if the index already exists - cursor.execute( - "SELECT name FROM sqlite_master WHERE type='index' AND name='idx_memory_unique_key'" - ) - index_exists = cursor.fetchone() is not None - - if not index_exists: - # Migration: Clean up duplicate entries before creating unique index - # Keep the most recent entry (highest id) for each (project_id, category, key) - cursor.execute( - """ - DELETE FROM memory - WHERE id NOT IN ( - SELECT MAX(id) - FROM memory - GROUP BY project_id, category, key - ) - """ - ) - deleted_count = cursor.rowcount - if deleted_count > 0: - logger.info( - f"Migration: Removed {deleted_count} duplicate memory entries" - ) - - cursor.execute( - """ - CREATE UNIQUE INDEX IF NOT EXISTS idx_memory_unique_key - ON memory(project_id, category, key) - """ - ) - - # Context items table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS context_items ( - id TEXT PRIMARY KEY, - project_id INTEGER REFERENCES projects(id), - agent_id TEXT NOT NULL, - item_type TEXT, - content TEXT, - importance_score FLOAT, - importance_reasoning TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - last_accessed TIMESTAMP, - access_count INTEGER DEFAULT 0, - current_tier TEXT CHECK(current_tier IN ('hot', 'warm', 'cold')), - manual_pin BOOLEAN DEFAULT FALSE - ) - """ - ) - - def _create_checkpoint_git_tables(self, cursor: sqlite3.Cursor) -> None: - """Create checkpoint, git tracking, and deployment tables.""" - # Checkpoints table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS checkpoints ( - id INTEGER PRIMARY KEY, - project_id INTEGER REFERENCES projects(id), - trigger TEXT, - state_snapshot JSON, - git_commit TEXT, - db_backup_path TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - name TEXT, - description TEXT, - database_backup_path TEXT, - context_snapshot_path TEXT, - metadata JSON - ) - """ - ) - - # Context checkpoints table (for flash save) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS context_checkpoints ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - agent_id TEXT NOT NULL, - checkpoint_data TEXT NOT NULL, - items_count INTEGER NOT NULL, - items_archived INTEGER NOT NULL, - hot_items_retained INTEGER NOT NULL, - token_count INTEGER NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - # Changelog table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS changelog ( - id INTEGER PRIMARY KEY, - project_id INTEGER REFERENCES projects(id), - agent_id TEXT, - task_id INTEGER, - action TEXT, - details JSON, - timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - # Git branches table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS git_branches ( - id INTEGER PRIMARY KEY, - issue_id INTEGER REFERENCES issues(id), - branch_name TEXT NOT NULL, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - merged_at TIMESTAMP, - merge_commit TEXT, - status TEXT CHECK(status IN ('active', 'merged', 'abandoned')) DEFAULT 'active' - ) - """ - ) - - # Deployments table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS deployments ( - id INTEGER PRIMARY KEY, - commit_hash TEXT NOT NULL, - environment TEXT CHECK(environment IN ('staging', 'production')), - status TEXT CHECK(status IN ('success', 'failed')), - output TEXT, - duration_seconds REAL, - triggered_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - - # Pull requests table (Sprint 11 - GitHub PR integration) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS pull_requests ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, - issue_id INTEGER REFERENCES issues(id) ON DELETE SET NULL, - branch_name TEXT NOT NULL, - pr_number INTEGER, - pr_url TEXT, - title TEXT NOT NULL, - body TEXT, - base_branch TEXT DEFAULT 'main', - head_branch TEXT NOT NULL, - status TEXT CHECK(status IN ('draft', 'open', 'merged', 'closed')) DEFAULT 'open', - merge_commit_sha TEXT, - created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - merged_at TIMESTAMP, - closed_at TIMESTAMP, - github_created_at TIMESTAMP, - github_updated_at TIMESTAMP - ) - """ - ) - - def _create_metrics_audit_tables(self, cursor: sqlite3.Cursor) -> None: - """Create metrics, token usage, and audit log tables.""" - # Token usage table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS token_usage ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - task_id INTEGER REFERENCES tasks(id) ON DELETE SET NULL, - agent_id TEXT NOT NULL, - project_id INTEGER NOT NULL REFERENCES projects(id) ON DELETE CASCADE, - model_name TEXT NOT NULL, - input_tokens INTEGER NOT NULL CHECK(input_tokens >= 0), - output_tokens INTEGER NOT NULL CHECK(output_tokens >= 0), - estimated_cost_usd REAL NOT NULL CHECK(estimated_cost_usd >= 0), - actual_cost_usd REAL CHECK(actual_cost_usd >= 0), - call_type TEXT CHECK(call_type IN ('task_execution', 'code_review', 'coordination', 'other')), - timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - session_id TEXT DEFAULT NULL - ) - """ - ) - - # Audit logs table - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS audit_logs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - event_type TEXT NOT NULL, - user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, - resource_type TEXT NOT NULL, - resource_id INTEGER, - ip_address TEXT, - metadata TEXT, - timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - """ - ) - def _create_interactive_session_tables(self, cursor: sqlite3.Cursor) -> None: """Create interactive_sessions and session_messages tables.""" cursor.execute( @@ -871,158 +195,60 @@ def _create_interactive_session_tables(self, cursor: sqlite3.Cursor) -> None: """ ) - def _create_indexes(self, cursor: sqlite3.Cursor) -> None: - """Create all database indexes for performance.""" - # Issues indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_issues_number ON issues(project_id, issue_number)" - ) - - # Tasks indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_tasks_issue_number ON tasks(parent_issue_number)" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_tasks_pending_priority ON tasks(project_id, status, priority, created_at)" - ) - # Index for agent maturity queries (get_tasks_by_agent) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_tasks_assigned_to ON tasks(assigned_to, project_id, created_at)" - ) - - # Project-Agent indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_project_agents_project_active ON project_agents(project_id, is_active) WHERE is_active = TRUE" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_project_agents_agent_active ON project_agents(agent_id, is_active) WHERE is_active = TRUE" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_project_agents_assigned_at ON project_agents(assigned_at)" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_project_agents_unassigned ON project_agents(unassigned_at) WHERE unassigned_at IS NOT NULL" - ) - cursor.execute( - "CREATE UNIQUE INDEX IF NOT EXISTS idx_project_agents_unique_active ON project_agents(project_id, agent_id, is_active) WHERE is_active = TRUE" - ) - - # Blocker indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_blockers_status_created ON blockers(status, created_at)" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_blockers_agent_status ON blockers(agent_id, status)" - ) - cursor.execute("CREATE INDEX IF NOT EXISTS idx_blockers_task_id ON blockers(task_id)") - - # Lint results indexes - cursor.execute("CREATE INDEX IF NOT EXISTS idx_lint_results_task ON lint_results(task_id)") - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_lint_results_created ON lint_results(created_at DESC)" - ) - - # Context items indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_context_project_agent ON context_items(project_id, agent_id, current_tier)" - ) - - # Context checkpoints indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_checkpoints_agent_created ON context_checkpoints(agent_id, created_at DESC)" - ) - - # Test results indexes - cursor.execute("CREATE INDEX IF NOT EXISTS idx_test_results_task ON test_results(task_id)") - - # Correction attempts indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_correction_attempts_task ON correction_attempts(task_id)" - ) - - # Task dependencies indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_task_dependencies_task ON task_dependencies(task_id)" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_task_dependencies_depends_on ON task_dependencies(depends_on_task_id)" - ) - - # Code reviews indexes - cursor.execute("CREATE INDEX IF NOT EXISTS idx_reviews_task ON code_reviews(task_id)") - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_reviews_severity ON code_reviews(severity, created_at)" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_reviews_project ON code_reviews(project_id, created_at)" - ) - - # Task evidence indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_task_evidence_task ON task_evidence(task_id)" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_task_evidence_verified ON task_evidence(verified, created_at DESC)" - ) - - # Token usage indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_token_usage_agent ON token_usage(agent_id, timestamp)" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_token_usage_project ON token_usage(project_id, timestamp)" - ) - cursor.execute("CREATE INDEX IF NOT EXISTS idx_token_usage_task ON token_usage(task_id)") - - # Checkpoints indexes - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_checkpoints_project ON checkpoints(project_id, created_at DESC)" - ) - - # Pull requests indexes (Sprint 11 - GitHub PR integration) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_pull_requests_project ON pull_requests(project_id, status)" - ) - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_pull_requests_issue ON pull_requests(issue_id)" - ) + def _create_audit_log_table(self, cursor: sqlite3.Cursor) -> None: + """Create the audit log table.""" cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_pull_requests_branch ON pull_requests(project_id, branch_name)" + """ + CREATE TABLE IF NOT EXISTS audit_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event_type TEXT NOT NULL, + user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, + resource_type TEXT NOT NULL, + resource_id INTEGER, + ip_address TEXT, + metadata TEXT, + timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ ) + def _create_indexes(self, cursor: sqlite3.Cursor) -> None: + """Create indexes for the live platform tables.""" # Interactive session indexes cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_interactive_sessions_workspace ON interactive_sessions(workspace_path, state)" + "CREATE INDEX IF NOT EXISTS idx_interactive_sessions_workspace " + "ON interactive_sessions(workspace_path, state)" ) cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_interactive_sessions_state ON interactive_sessions(state, created_at DESC)" + "CREATE INDEX IF NOT EXISTS idx_interactive_sessions_state " + "ON interactive_sessions(state, created_at DESC)" ) cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_session_messages_session ON session_messages(session_id, created_at)" + "CREATE INDEX IF NOT EXISTS idx_session_messages_session " + "ON session_messages(session_id, created_at)" ) - # Audit logs indexes + # Audit log indexes cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_audit_logs_user_id ON audit_logs(user_id, timestamp DESC)" + "CREATE INDEX IF NOT EXISTS idx_audit_logs_user_id " + "ON audit_logs(user_id, timestamp DESC)" ) cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_audit_logs_event_type ON audit_logs(event_type, timestamp DESC)" + "CREATE INDEX IF NOT EXISTS idx_audit_logs_event_type " + "ON audit_logs(event_type, timestamp DESC)" ) cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_audit_logs_resource ON audit_logs(resource_type, resource_id, timestamp DESC)" + "CREATE INDEX IF NOT EXISTS idx_audit_logs_resource " + "ON audit_logs(resource_type, resource_id, timestamp DESC)" ) - # Authentication indexes + # Authentication indexes (api_keys/accounts indexes are created inline + # with their tables in _create_auth_tables) cursor.execute("CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON sessions(user_id)") - cursor.execute("CREATE INDEX IF NOT EXISTS idx_sessions_expires_at ON sessions(expires_at)") - cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_project_users_user_id ON project_users(user_id)" - ) cursor.execute( - "CREATE INDEX IF NOT EXISTS idx_project_users_user_project ON project_users(user_id, project_id)" + "CREATE INDEX IF NOT EXISTS idx_sessions_expires_at ON sessions(expires_at)" ) - cursor.execute("CREATE INDEX IF NOT EXISTS idx_projects_user_id ON projects(user_id)") def _ensure_default_admin_user(self) -> None: """Ensure default admin user exists in database for initial setup. diff --git a/tests/agents/test_lead_agent_intervention.py b/tests/agents/test_lead_agent_intervention.py deleted file mode 100644 index b00fb54b..00000000 --- a/tests/agents/test_lead_agent_intervention.py +++ /dev/null @@ -1,239 +0,0 @@ -""" -Tests for LeadAgent supervisor intervention logic. - -Test coverage for tactical pattern-based intervention: -- Pattern matching on file conflict errors -- Intervention context applied to task -- Intervention instructions generated correctly -- Error handling when patterns don't match - -Following strict TDD methodology. -""" - -import pytest -from codeframe.agents.lead_agent import LeadAgent -from codeframe.agents.tactical_patterns import InterventionStrategy, TacticalPattern -from codeframe.persistence.database import Database -from codeframe.core.models import Task, TaskStatus - -pytestmark = pytest.mark.v2 - - -class TestLeadAgentIntervention: - """Test supervisor intervention logic.""" - - @pytest.fixture - def lead_agent(self, temp_db_path): - """Create a LeadAgent with real database.""" - db = Database(temp_db_path) - db.initialize() - project_id = db.create_project("test-project", "Test project") - - agent = LeadAgent( - project_id=project_id, - db=db, - api_key="sk-ant-test-key", - ) - return agent - - @pytest.fixture - def sample_task(self, lead_agent): - """Create a sample task for testing.""" - db = lead_agent.db - project_id = lead_agent.project_id - - # Create issue first - issue_id = db.create_issue({ - "project_id": project_id, - "issue_number": "1.0", - "title": "Test Issue", - "status": "pending", - "priority": 0, - "workflow_step": 1, - }) - - # Create task - task_id = db.create_task_with_issue( - project_id=project_id, - issue_id=issue_id, - task_number="1.0.1", - parent_issue_number="1.0", - title="Test Task", - description="Test task description", - status=TaskStatus.PENDING, - priority=0, - workflow_step=1, - can_parallelize=False, - ) - - # Return as Task object - return Task( - id=task_id, - project_id=project_id, - issue_id=issue_id, - task_number="1.0.1", - parent_issue_number="1.0", - title="Test Task", - description="Test task description", - status=TaskStatus.PENDING, - priority=0, - workflow_step=1, - ) - - -class TestFileConflictIntervention(TestLeadAgentIntervention): - """Test file conflict intervention handling.""" - - def test_handle_file_conflict_creates_intervention_context( - self, lead_agent, sample_task - ): - """Test that file conflict intervention creates proper context.""" - error = FileExistsError("File already exists: src/component.tsx") - pattern = TacticalPattern( - pattern_id="file_already_exists", - error_pattern=r"file.*exists", - category="file_conflict", - intervention_strategy=InterventionStrategy.CONVERT_CREATE_TO_EDIT, - ) - - lead_agent._handle_file_conflict_intervention(sample_task, error, pattern) - - # Verify context was saved - context = lead_agent.db.get_task_intervention_context(sample_task.id) - - assert context is not None - assert context["intervention_applied"] is True - assert context["pattern_matched"] == "file_already_exists" - assert context["strategy"] == "convert_create_to_edit" - - def test_handle_file_conflict_extracts_file_path( - self, lead_agent, sample_task - ): - """Test that file path is extracted from error message.""" - error = FileExistsError("File already exists: src/Button.tsx") - pattern = TacticalPattern( - pattern_id="file_already_exists", - error_pattern=r"file.*exists", - category="file_conflict", - intervention_strategy=InterventionStrategy.CONVERT_CREATE_TO_EDIT, - ) - - lead_agent._handle_file_conflict_intervention(sample_task, error, pattern) - - context = lead_agent.db.get_task_intervention_context(sample_task.id) - - assert "src/Button.tsx" in context["existing_files"] - - def test_handle_file_conflict_includes_workspace_files( - self, lead_agent, sample_task - ): - """Test that workspace state files are included in context.""" - # Add files to workspace state - lead_agent.update_workspace_state( - task_id=sample_task.id, - files_created=["existing_file.py"], - ) - - error = FileExistsError("File already exists: new_file.py") - pattern = TacticalPattern( - pattern_id="file_already_exists", - error_pattern=r"file.*exists", - category="file_conflict", - intervention_strategy=InterventionStrategy.CONVERT_CREATE_TO_EDIT, - ) - - lead_agent._handle_file_conflict_intervention(sample_task, error, pattern) - - context = lead_agent.db.get_task_intervention_context(sample_task.id) - - # Both files should be in existing_files - assert "existing_file.py" in context["existing_files"] - assert "new_file.py" in context["existing_files"] - - def test_handle_file_conflict_includes_instruction( - self, lead_agent, sample_task - ): - """Test that intervention includes clear instruction.""" - error = FileExistsError("File already exists: src/file.py") - pattern = TacticalPattern( - pattern_id="file_already_exists", - error_pattern=r"file.*exists", - category="file_conflict", - intervention_strategy=InterventionStrategy.CONVERT_CREATE_TO_EDIT, - ) - - lead_agent._handle_file_conflict_intervention(sample_task, error, pattern) - - context = lead_agent.db.get_task_intervention_context(sample_task.id) - - assert "instruction" in context - assert "modify" in context["instruction"].lower() - assert "create" in context["instruction"].lower() - - -class TestInterventionInstruction(TestLeadAgentIntervention): - """Test intervention instruction generation.""" - - def test_convert_create_to_edit_instruction(self, lead_agent): - """Test CONVERT_CREATE_TO_EDIT instruction.""" - instruction = lead_agent._get_intervention_instruction( - InterventionStrategy.CONVERT_CREATE_TO_EDIT - ) - - assert "modify" in instruction.lower() - assert "create" in instruction.lower() - assert "existing_files" in instruction - - def test_skip_file_creation_instruction(self, lead_agent): - """Test SKIP_FILE_CREATION instruction.""" - instruction = lead_agent._get_intervention_instruction( - InterventionStrategy.SKIP_FILE_CREATION - ) - - assert "skip" in instruction.lower() - assert "existing_files" in instruction - - def test_create_backup_instruction(self, lead_agent): - """Test CREATE_BACKUP instruction.""" - instruction = lead_agent._get_intervention_instruction( - InterventionStrategy.CREATE_BACKUP - ) - - assert "backup" in instruction.lower() - - def test_retry_with_context_instruction(self, lead_agent): - """Test RETRY_WITH_CONTEXT instruction.""" - instruction = lead_agent._get_intervention_instruction( - InterventionStrategy.RETRY_WITH_CONTEXT - ) - - assert "existing_files" in instruction - - -class TestPatternMatcherIntegration(TestLeadAgentIntervention): - """Test integration with TacticalPatternMatcher.""" - - def test_pattern_matcher_initialized(self, lead_agent): - """Test that pattern matcher is initialized.""" - assert lead_agent._tactical_pattern_matcher is not None - - def test_pattern_matcher_has_file_exists_pattern(self, lead_agent): - """Test that matcher has file_already_exists pattern.""" - matcher = lead_agent._tactical_pattern_matcher - - # Test that it matches - result = matcher.match_error("FileExistsError: File already exists") - - assert result is not None - assert result.pattern_id == "file_already_exists" - - def test_pattern_matcher_returns_diagnostics(self, lead_agent): - """Test that match_error_with_diagnostics returns diagnostics.""" - matcher = lead_agent._tactical_pattern_matcher - - result, diagnostics = matcher.match_error_with_diagnostics( - "FileExistsError: File exists" - ) - - assert diagnostics["patterns_checked"] > 0 - assert diagnostics["matched_pattern"] == "file_already_exists" diff --git a/tests/agents/test_lead_agent_scheduling.py b/tests/agents/test_lead_agent_scheduling.py deleted file mode 100644 index 967f4cb1..00000000 --- a/tests/agents/test_lead_agent_scheduling.py +++ /dev/null @@ -1,317 +0,0 @@ -"""Tests for LeadAgent task scheduling integration. - -TDD tests for integrating TaskScheduler with LeadAgent: -- schedule_project_tasks() - Create project schedule from tasks -- Integration with effort estimation and dependency resolver -""" - -import pytest - -from codeframe.agents.lead_agent import LeadAgent -from codeframe.persistence.database import Database -from codeframe.core.models import Issue, TaskStatus -from codeframe.planning.task_scheduler import ScheduleResult - -pytestmark = pytest.mark.v2 - - -@pytest.fixture -def temp_db(temp_db_path): - """Create initialized database for testing.""" - db = Database(temp_db_path) - db.initialize() - return db - - -@pytest.fixture -def project_with_tasks(temp_db): - """Create a project with tasks that have dependencies and effort estimates.""" - project_id = temp_db.create_project("test-project", "Test scheduling project") - - # Create an issue for the tasks - issue = Issue( - project_id=project_id, - issue_number="1", - title="Test Issue", - description="Test issue for scheduling", - status=TaskStatus.PENDING, - priority=2, - workflow_step=1, - ) - issue_id = temp_db.create_issue(issue) - - # Create tasks with dependencies and estimated_hours - # Task structure: A -> (B, C) -> D - task_a = temp_db.create_task_with_issue( - project_id=project_id, - issue_id=issue_id, - task_number="1.1", - parent_issue_number="1", - title="Task A - Foundation", - description="Foundation task", - status=TaskStatus.PENDING, - priority=2, - workflow_step=1, - can_parallelize=False, - estimated_hours=2.0, - complexity_score=2, - ) - - task_b = temp_db.create_task_with_issue( - project_id=project_id, - issue_id=issue_id, - task_number="1.2", - parent_issue_number="1", - title="Task B - Feature", - description="Feature task", - status=TaskStatus.PENDING, - priority=2, - workflow_step=1, - can_parallelize=True, - estimated_hours=3.0, - complexity_score=3, - ) - # Add dependency: B depends on A - temp_db.add_task_dependency(task_b, task_a) - - task_c = temp_db.create_task_with_issue( - project_id=project_id, - issue_id=issue_id, - task_number="1.3", - parent_issue_number="1", - title="Task C - Integration", - description="Integration task", - status=TaskStatus.PENDING, - priority=2, - workflow_step=1, - can_parallelize=True, - estimated_hours=1.0, - complexity_score=2, - ) - # Add dependency: C depends on A - temp_db.add_task_dependency(task_c, task_a) - - task_d = temp_db.create_task_with_issue( - project_id=project_id, - issue_id=issue_id, - task_number="1.4", - parent_issue_number="1", - title="Task D - Finalization", - description="Final task", - status=TaskStatus.PENDING, - priority=2, - workflow_step=1, - can_parallelize=False, - estimated_hours=2.0, - complexity_score=2, - ) - # Add dependencies: D depends on B and C - temp_db.add_task_dependency(task_d, task_b) - temp_db.add_task_dependency(task_d, task_c) - - return { - "project_id": project_id, - "issue_id": issue_id, - "task_ids": [task_a, task_b, task_c, task_d], - "db": temp_db, - } - - -@pytest.mark.unit -class TestScheduleProjectTasks: - """Test schedule_project_tasks() method.""" - - def test_schedule_project_tasks_returns_schedule_result(self, project_with_tasks): - """Test that schedule_project_tasks returns a ScheduleResult.""" - db = project_with_tasks["db"] - project_id = project_with_tasks["project_id"] - - agent = LeadAgent(project_id=project_id, db=db, api_key="sk-ant-test-key") - - result = agent.schedule_project_tasks() - - assert isinstance(result, ScheduleResult) - assert hasattr(result, "task_assignments") - assert hasattr(result, "total_duration") - assert hasattr(result, "timeline") - - def test_schedule_project_tasks_includes_all_tasks(self, project_with_tasks): - """Test that schedule includes all project tasks.""" - db = project_with_tasks["db"] - project_id = project_with_tasks["project_id"] - task_ids = project_with_tasks["task_ids"] - - agent = LeadAgent(project_id=project_id, db=db, api_key="sk-ant-test-key") - - result = agent.schedule_project_tasks() - - # All 4 tasks should be scheduled - assert len(result.task_assignments) == 4 - for task_id in task_ids: - assert task_id in result.task_assignments - - def test_schedule_uses_estimated_hours(self, project_with_tasks): - """Test that schedule uses estimated_hours from tasks.""" - db = project_with_tasks["db"] - project_id = project_with_tasks["project_id"] - task_ids = project_with_tasks["task_ids"] - - agent = LeadAgent(project_id=project_id, db=db, api_key="sk-ant-test-key") - - result = agent.schedule_project_tasks() - - # Task A duration should be 2.0 hours (as set in fixture) - task_a_id = task_ids[0] - assignment = result.task_assignments[task_a_id] - duration = assignment.end_time - assignment.start_time - assert duration == 2.0 - - def test_schedule_respects_dependencies(self, project_with_tasks): - """Test that schedule respects task dependencies.""" - db = project_with_tasks["db"] - project_id = project_with_tasks["project_id"] - task_ids = project_with_tasks["task_ids"] - - agent = LeadAgent(project_id=project_id, db=db, api_key="sk-ant-test-key") - - result = agent.schedule_project_tasks() - - task_a_id, task_b_id, task_c_id, task_d_id = task_ids - - # B and C must start after A ends - a_end = result.task_assignments[task_a_id].end_time - b_start = result.task_assignments[task_b_id].start_time - c_start = result.task_assignments[task_c_id].start_time - - assert b_start >= a_end - assert c_start >= a_end - - # D must start after both B and C end - b_end = result.task_assignments[task_b_id].end_time - c_end = result.task_assignments[task_c_id].end_time - d_start = result.task_assignments[task_d_id].start_time - - assert d_start >= max(b_end, c_end) - - def test_schedule_with_multiple_agents(self, project_with_tasks): - """Test scheduling with multiple agents reduces duration.""" - db = project_with_tasks["db"] - project_id = project_with_tasks["project_id"] - - agent = LeadAgent(project_id=project_id, db=db, api_key="sk-ant-test-key") - - # Schedule with 1 agent (serial) - result_1_agent = agent.schedule_project_tasks(agents_available=1) - - # Schedule with 2 agents (parallel B and C) - result_2_agents = agent.schedule_project_tasks(agents_available=2) - - # 2 agents should be faster or equal - assert result_2_agents.total_duration <= result_1_agent.total_duration - - def test_schedule_with_default_duration_for_missing_estimates(self, temp_db): - """Test that tasks without estimated_hours use default duration.""" - project_id = temp_db.create_project("test-project", "Test project") - - # Create an issue - issue = Issue( - project_id=project_id, - issue_number="1", - title="Test Issue", - description="Test", - status=TaskStatus.PENDING, - priority=2, - workflow_step=1, - ) - issue_id = temp_db.create_issue(issue) - - # Create task without estimated_hours - task_id = temp_db.create_task_with_issue( - project_id=project_id, - issue_id=issue_id, - task_number="1.1", - parent_issue_number="1", - title="Task without estimate", - description="Test task", - status=TaskStatus.PENDING, - priority=2, - workflow_step=1, - can_parallelize=False, - # No estimated_hours - ) - - agent = LeadAgent(project_id=project_id, db=temp_db, api_key="sk-ant-test-key") - - result = agent.schedule_project_tasks() - - # Task should be scheduled with default duration (1.0 hour) - assert task_id in result.task_assignments - duration = result.task_assignments[task_id].end_time - result.task_assignments[task_id].start_time - assert duration == 1.0 # Default duration - - def test_schedule_empty_project_returns_empty_schedule(self, temp_db): - """Test scheduling project with no tasks returns empty schedule.""" - project_id = temp_db.create_project("empty-project", "No tasks") - - agent = LeadAgent(project_id=project_id, db=temp_db, api_key="sk-ant-test-key") - - result = agent.schedule_project_tasks() - - assert isinstance(result, ScheduleResult) - assert len(result.task_assignments) == 0 - assert result.total_duration == 0.0 - - -@pytest.mark.unit -class TestGetProjectScheduleInfo: - """Test helper methods for scheduling information.""" - - def test_get_project_duration_hours(self, project_with_tasks): - """Test getting total project duration in hours.""" - db = project_with_tasks["db"] - project_id = project_with_tasks["project_id"] - - agent = LeadAgent(project_id=project_id, db=db, api_key="sk-ant-test-key") - - result = agent.schedule_project_tasks(agents_available=2) - - # With 2 agents: A(2) + max(B(3), C(1)) + D(2) = 7 hours - assert result.total_duration == 7.0 - - def test_schedule_timeline_has_events(self, project_with_tasks): - """Test that schedule timeline contains start/end events.""" - db = project_with_tasks["db"] - project_id = project_with_tasks["project_id"] - - agent = LeadAgent(project_id=project_id, db=db, api_key="sk-ant-test-key") - - result = agent.schedule_project_tasks() - - # Should have start and end events for each task - assert len(result.timeline) == 8 # 4 tasks * 2 events each - - event_types = [e.event_type for e in result.timeline] - assert event_types.count("start") == 4 - assert event_types.count("end") == 4 - - -@pytest.mark.unit -class TestScheduleWithProgress: - """Test scheduling with partially completed tasks.""" - - def test_schedule_excludes_completed_tasks_duration(self, project_with_tasks): - """Test that completed tasks affect remaining schedule correctly.""" - db = project_with_tasks["db"] - project_id = project_with_tasks["project_id"] - task_ids = project_with_tasks["task_ids"] - - # Mark task A as completed - db.update_task(task_ids[0], {"status": TaskStatus.COMPLETED.value}) - - agent = LeadAgent(project_id=project_id, db=db, api_key="sk-ant-test-key") - - # Schedule should still include all tasks for planning purposes - # but completed tasks can be handled by predict_completion_date - result = agent.schedule_project_tasks() - - assert len(result.task_assignments) == 4 diff --git a/tests/agents/test_lead_agent_workspace_state.py b/tests/agents/test_lead_agent_workspace_state.py deleted file mode 100644 index 74aa78a9..00000000 --- a/tests/agents/test_lead_agent_workspace_state.py +++ /dev/null @@ -1,166 +0,0 @@ -""" -Tests for LeadAgent workspace state tracking. - -Test coverage for supervisor intervention support: -- Workspace state initialization -- Updating workspace state after task execution -- Getting workspace context for intervention -- File deduplication in workspace context - -Following strict TDD methodology. -""" - -import pytest -from codeframe.agents.lead_agent import LeadAgent -from codeframe.persistence.database import Database - -pytestmark = pytest.mark.v2 - - -class TestLeadAgentWorkspaceState: - """Test workspace state tracking for supervisor intervention.""" - - @pytest.fixture - def lead_agent(self, temp_db_path): - """Create a LeadAgent with real database.""" - db = Database(temp_db_path) - db.initialize() - project_id = db.create_project("test-project", "Test project") - - agent = LeadAgent( - project_id=project_id, - db=db, - api_key="sk-ant-test-key", - ) - return agent - - def test_workspace_state_initialized_empty(self, lead_agent): - """Test that workspace state is initialized as empty dict.""" - assert lead_agent._workspace_state == {} - - def test_update_workspace_state_creates_entry(self, lead_agent): - """Test that update creates new entry for task.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=["src/new_file.py"], - files_modified=["src/existing.py"], - ) - - assert 1 in lead_agent._workspace_state - assert "src/new_file.py" in lead_agent._workspace_state[1]["files_created"] - assert "src/existing.py" in lead_agent._workspace_state[1]["files_modified"] - - def test_update_workspace_state_appends_to_existing(self, lead_agent): - """Test that update appends to existing task entry.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=["file1.py"], - ) - lead_agent.update_workspace_state( - task_id=1, - files_created=["file2.py"], - ) - - assert "file1.py" in lead_agent._workspace_state[1]["files_created"] - assert "file2.py" in lead_agent._workspace_state[1]["files_created"] - - def test_update_workspace_state_handles_none(self, lead_agent): - """Test that update handles None values gracefully.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=None, - files_modified=None, - ) - - assert 1 in lead_agent._workspace_state - assert lead_agent._workspace_state[1]["files_created"] == [] - assert lead_agent._workspace_state[1]["files_modified"] == [] - - def test_get_workspace_context_returns_all_files(self, lead_agent): - """Test that get_workspace_context returns all tracked files.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=["file1.py"], - ) - lead_agent.update_workspace_state( - task_id=2, - files_created=["file2.py"], - ) - - context = lead_agent.get_workspace_context(task_id=3) - - assert "file1.py" in context["existing_files"] - assert "file2.py" in context["existing_files"] - - def test_get_workspace_context_deduplicates_files(self, lead_agent): - """Test that duplicate files are removed from context.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=["shared.py"], - ) - lead_agent.update_workspace_state( - task_id=2, - files_modified=["shared.py"], # Same file modified by task 2 - ) - - context = lead_agent.get_workspace_context(task_id=3) - - # Should only appear once - assert context["existing_files"].count("shared.py") == 1 - - def test_get_workspace_context_includes_files_by_task(self, lead_agent): - """Test that context includes files grouped by task.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=["task1_file.py"], - ) - lead_agent.update_workspace_state( - task_id=2, - files_created=["task2_file.py"], - ) - - context = lead_agent.get_workspace_context(task_id=1) - - assert 1 in context["files_by_task"] - assert 2 in context["files_by_task"] - assert "task1_file.py" in context["files_by_task"][1]["files_created"] - - def test_get_workspace_context_includes_task_specific_files(self, lead_agent): - """Test that context includes files for specific task.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=["my_file.py"], - files_modified=["other.py"], - ) - - context = lead_agent.get_workspace_context(task_id=1) - - assert "my_file.py" in context["task_specific_files"]["files_created"] - assert "other.py" in context["task_specific_files"]["files_modified"] - - def test_get_workspace_context_empty_for_unknown_task(self, lead_agent): - """Test that task_specific_files is empty for unknown task.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=["file.py"], - ) - - context = lead_agent.get_workspace_context(task_id=999) - - assert context["task_specific_files"] == {} - # But existing_files should still include all files - assert "file.py" in context["existing_files"] - - def test_workspace_state_preserves_order(self, lead_agent): - """Test that file order is preserved in existing_files.""" - lead_agent.update_workspace_state( - task_id=1, - files_created=["first.py", "second.py", "third.py"], - ) - - context = lead_agent.get_workspace_context(task_id=1) - - # Order should be preserved - files = context["existing_files"] - assert files.index("first.py") < files.index("second.py") - assert files.index("second.py") < files.index("third.py") diff --git a/tests/api/conftest.py b/tests/api/conftest.py index fa389833..7f5152a4 100644 --- a/tests/api/conftest.py +++ b/tests/api/conftest.py @@ -205,23 +205,22 @@ def clean_database_between_tests(api_client: TestClient) -> Generator[None, None db = server.app.state.db cursor = db.conn.cursor() - # Delete all rows from tables (in reverse dependency order) - cursor.execute("DELETE FROM code_reviews") - cursor.execute("DELETE FROM token_usage") - cursor.execute("DELETE FROM context_items") - cursor.execute("DELETE FROM checkpoints") - cursor.execute("DELETE FROM memory") - cursor.execute("DELETE FROM blockers") - cursor.execute("DELETE FROM changelog") # References projects, tasks - cursor.execute("DELETE FROM tasks") - cursor.execute("DELETE FROM git_branches") # Must be before issues (FK constraint) - cursor.execute("DELETE FROM issues") - cursor.execute("DELETE FROM project_agents") # Multi-agent junction table - cursor.execute("DELETE FROM agents") - cursor.execute("DELETE FROM sessions") # Auth sessions - cursor.execute("DELETE FROM project_users") # Project-user relationships - cursor.execute("DELETE FROM projects") - # Note: Keep users table (especially default admin user with id=1) + # Clear every table except `users` (keeps the default admin id=1). + # Only touch tables that actually exist — the control-plane schema is + # intentionally minimal, so legacy tables may not be present. + existing = { + row[0] + for row in cursor.execute( + "SELECT name FROM sqlite_master WHERE type='table' " + "AND name NOT LIKE 'sqlite_%'" + ).fetchall() + } + cursor.execute("PRAGMA foreign_keys = OFF") + for table in existing - {"users"}: + # Identifier comes from sqlite_master (not user input); quote it + # anyway so reserved words / unusual names are handled safely. + cursor.execute(f'DELETE FROM "{table}"') + cursor.execute("PRAGMA foreign_keys = ON") db.conn.commit() diff --git a/tests/cli/test_stats_commands.py b/tests/cli/test_stats_commands.py deleted file mode 100644 index 2da220e9..00000000 --- a/tests/cli/test_stats_commands.py +++ /dev/null @@ -1,251 +0,0 @@ -"""Tests for CLI stats commands (headless token/cost tracking). - -TDD approach: Write tests first, then implement. -Tests the `cf stats tokens`, `cf stats costs`, and `cf stats export` commands. -""" - -import csv -import json -import os -from datetime import datetime, timezone - -import pytest -from typer.testing import CliRunner - -from codeframe.cli.stats_commands import stats_app -from codeframe.core.models import CallType, TokenUsage -from codeframe.persistence.database import Database - -pytestmark = pytest.mark.v2 - -runner = CliRunner() - - -def _seed_project_and_tasks(db): - """Create a project and tasks to satisfy FK constraints.""" - cursor = db.conn.cursor() - cursor.execute( - "INSERT INTO projects (name, description, source_type, source_branch, workspace_path) " - "VALUES (?, ?, ?, ?, ?)", - ("test-project", "Test project", "empty", "main", "/tmp/test"), - ) - cursor.execute( - "INSERT INTO tasks (project_id, title, description, status, priority) " - "VALUES (?, ?, ?, ?, ?)", - (1, "Task 1", "First task", "in_progress", 0), - ) - cursor.execute( - "INSERT INTO tasks (project_id, title, description, status, priority) " - "VALUES (?, ?, ?, ?, ?)", - (1, "Task 2", "Second task", "in_progress", 0), - ) - db.conn.commit() - - -@pytest.fixture -def workspace_with_tokens(tmp_path): - """Create a workspace with seeded token usage data.""" - codeframe_dir = tmp_path / ".codeframe" - codeframe_dir.mkdir() - db = Database(codeframe_dir / "state.db") - db.initialize() - - _seed_project_and_tasks(db) - - records = [ - TokenUsage( - task_id=1, - agent_id="react-agent", - project_id=1, - model_name="claude-sonnet-4-5", - input_tokens=1000, - output_tokens=500, - estimated_cost_usd=0.0105, - call_type=CallType.TASK_EXECUTION, - timestamp=datetime(2026, 3, 10, 10, 0, 0, tzinfo=timezone.utc), - ), - TokenUsage( - task_id=1, - agent_id="react-agent", - project_id=1, - model_name="claude-sonnet-4-5", - input_tokens=2000, - output_tokens=800, - estimated_cost_usd=0.018, - call_type=CallType.TASK_EXECUTION, - timestamp=datetime(2026, 3, 10, 11, 0, 0, tzinfo=timezone.utc), - ), - TokenUsage( - task_id=2, - agent_id="react-agent", - project_id=1, - model_name="claude-haiku-4", - input_tokens=500, - output_tokens=200, - estimated_cost_usd=0.0012, - call_type=CallType.CODE_REVIEW, - timestamp=datetime(2026, 3, 12, 10, 0, 0, tzinfo=timezone.utc), - ), - ] - - for record in records: - db.save_token_usage(record) - - db.close() - return tmp_path - - -@pytest.fixture -def empty_workspace(tmp_path): - """Create a workspace with initialized DB but no token data.""" - codeframe_dir = tmp_path / ".codeframe" - codeframe_dir.mkdir() - db = Database(codeframe_dir / "state.db") - db.initialize() - db.close() - return tmp_path - - -# ============================================================================= -# cf stats tokens -# ============================================================================= - - -class TestStatsTokens: - """Tests for 'cf stats tokens' command.""" - - def test_stats_tokens_no_workspace(self, tmp_path, monkeypatch): - """Should show error when no workspace exists.""" - monkeypatch.chdir(tmp_path) - result = runner.invoke(stats_app, ["tokens"]) - assert result.exit_code == 1 - assert "No workspace found" in result.output - - def test_stats_tokens_empty(self, empty_workspace, monkeypatch): - """Should show zeros when no token data exists.""" - monkeypatch.chdir(empty_workspace) - result = runner.invoke(stats_app, ["tokens"]) - assert result.exit_code == 0 - assert "0" in result.output - - def test_stats_tokens_with_data(self, workspace_with_tokens, monkeypatch): - """Should show correct summary with seeded data.""" - monkeypatch.chdir(workspace_with_tokens) - result = runner.invoke(stats_app, ["tokens"]) - assert result.exit_code == 0 - # Total tokens: 1000+500 + 2000+800 + 500+200 = 5000 - assert "5,000" in result.output or "5000" in result.output - # Should show input/output breakdown - assert "Input" in result.output - assert "Output" in result.output - - def test_stats_tokens_task_filter(self, workspace_with_tokens, monkeypatch): - """Should show per-task breakdown when --task is provided.""" - monkeypatch.chdir(workspace_with_tokens) - result = runner.invoke(stats_app, ["tokens", "--task", "1"]) - assert result.exit_code == 0 - # Task 1 tokens: 1000+500 + 2000+800 = 4300 - assert "4,300" in result.output or "4300" in result.output - - -# ============================================================================= -# cf stats costs -# ============================================================================= - - -class TestStatsCosts: - """Tests for 'cf stats costs' command.""" - - def test_stats_costs_no_workspace(self, tmp_path, monkeypatch): - """Should show error when no workspace exists.""" - monkeypatch.chdir(tmp_path) - result = runner.invoke(stats_app, ["costs"]) - assert result.exit_code == 1 - assert "No workspace found" in result.output - - def test_stats_costs_default(self, workspace_with_tokens, monkeypatch): - """Should show all-time costs.""" - monkeypatch.chdir(workspace_with_tokens) - result = runner.invoke(stats_app, ["costs"]) - assert result.exit_code == 0 - assert "$" in result.output - # Total cost: 0.0105 + 0.018 + 0.0012 = 0.0297 - assert "0.0297" in result.output - - def test_stats_costs_period_month(self, workspace_with_tokens, monkeypatch): - """Should respect period filter for 'month'.""" - monkeypatch.chdir(workspace_with_tokens) - result = runner.invoke(stats_app, ["costs", "--period", "month"]) - assert result.exit_code == 0 - assert "$" in result.output - - def test_stats_costs_period_week(self, workspace_with_tokens, monkeypatch): - """Should respect period filter for 'week'.""" - monkeypatch.chdir(workspace_with_tokens) - result = runner.invoke(stats_app, ["costs", "--period", "week"]) - assert result.exit_code == 0 - - def test_stats_costs_period_day(self, workspace_with_tokens, monkeypatch): - """Should respect period filter for 'day'.""" - monkeypatch.chdir(workspace_with_tokens) - result = runner.invoke(stats_app, ["costs", "--period", "day"]) - assert result.exit_code == 0 - - -# ============================================================================= -# cf stats export -# ============================================================================= - - -class TestStatsExport: - """Tests for 'cf stats export' command.""" - - def test_stats_export_no_workspace(self, tmp_path, monkeypatch): - """Should show error when no workspace exists.""" - monkeypatch.chdir(tmp_path) - output_file = str(tmp_path / "out.csv") - result = runner.invoke(stats_app, ["export", "--format", "csv", "--output", output_file]) - assert result.exit_code == 1 - - def test_stats_export_csv(self, workspace_with_tokens, monkeypatch): - """Should create a valid CSV file.""" - monkeypatch.chdir(workspace_with_tokens) - output_file = str(workspace_with_tokens / "tokens.csv") - result = runner.invoke(stats_app, ["export", "--format", "csv", "--output", output_file]) - assert result.exit_code == 0 - assert os.path.exists(output_file) - - with open(output_file) as f: - reader = csv.DictReader(f) - rows = list(reader) - assert len(rows) == 3 - assert "input_tokens" in rows[0] - - def test_stats_export_json(self, workspace_with_tokens, monkeypatch): - """Should create a valid JSON file.""" - monkeypatch.chdir(workspace_with_tokens) - output_file = str(workspace_with_tokens / "tokens.json") - result = runner.invoke(stats_app, ["export", "--format", "json", "--output", output_file]) - assert result.exit_code == 0 - assert os.path.exists(output_file) - - with open(output_file) as f: - data = json.load(f) - assert "records" in data - assert len(data["records"]) == 3 - - def test_stats_export_csv_task_filter(self, workspace_with_tokens, monkeypatch): - """Should export only records for a specific task.""" - monkeypatch.chdir(workspace_with_tokens) - output_file = str(workspace_with_tokens / "task1.csv") - result = runner.invoke( - stats_app, ["export", "--format", "csv", "--output", output_file, "--task", "1"] - ) - assert result.exit_code == 0 - assert os.path.exists(output_file) - - with open(output_file) as f: - reader = csv.DictReader(f) - rows = list(reader) - # Task 1 has 2 records - assert len(rows) == 2 diff --git a/tests/core/test_phase_manager.py b/tests/core/test_phase_manager.py deleted file mode 100644 index c0333438..00000000 --- a/tests/core/test_phase_manager.py +++ /dev/null @@ -1,241 +0,0 @@ -"""Unit tests for PhaseManager class. - -Tests phase transition validation, execution, and phase requirements. -Following TDD principles - these tests are written before the implementation. -""" - -import pytest -from pathlib import Path - -from codeframe.persistence.database import Database -from codeframe.core.phase_manager import ( - PhaseManager, - VALID_TRANSITIONS, - PHASE_STEPS, - ProjectNotFoundError, - InvalidPhaseTransitionError, -) -from codeframe.core.models import ProjectPhase -from tests.conftest import setup_test_user - - -class TestPhaseManagerCanTransition: - """Tests for can_transition() validation method.""" - - def test_discovery_to_planning_valid(self): - """Discovery can transition to planning.""" - assert PhaseManager.can_transition("discovery", "planning") is True - - def test_planning_to_active_valid(self): - """Planning can transition to active.""" - assert PhaseManager.can_transition("planning", "active") is True - - def test_planning_to_discovery_valid(self): - """Planning can go back to discovery.""" - assert PhaseManager.can_transition("planning", "discovery") is True - - def test_active_to_review_valid(self): - """Active can transition to review.""" - assert PhaseManager.can_transition("active", "review") is True - - def test_active_to_planning_valid(self): - """Active can go back to planning.""" - assert PhaseManager.can_transition("active", "planning") is True - - def test_review_to_complete_valid(self): - """Review can transition to complete.""" - assert PhaseManager.can_transition("review", "complete") is True - - def test_review_to_active_valid(self): - """Review can go back to active.""" - assert PhaseManager.can_transition("review", "active") is True - - def test_discovery_to_active_invalid(self): - """Discovery cannot skip directly to active.""" - assert PhaseManager.can_transition("discovery", "active") is False - - def test_discovery_to_review_invalid(self): - """Discovery cannot skip to review.""" - assert PhaseManager.can_transition("discovery", "review") is False - - def test_complete_to_discovery_invalid(self): - """Complete cannot go back to discovery.""" - assert PhaseManager.can_transition("complete", "discovery") is False - - def test_unknown_from_phase(self): - """Unknown from_phase returns False.""" - assert PhaseManager.can_transition("unknown", "planning") is False - - def test_unknown_to_phase(self): - """Unknown to_phase returns False.""" - assert PhaseManager.can_transition("discovery", "unknown") is False - - def test_same_phase_invalid(self): - """Cannot transition to same phase.""" - assert PhaseManager.can_transition("discovery", "discovery") is False - - -class TestPhaseManagerTransition: - """Tests for transition() execution method.""" - - def test_transition_success(self, temp_db_path: Path): - """Successful phase transition updates database.""" - db = Database(temp_db_path) - db.initialize() - setup_test_user(db, user_id=1) - - # Create project in discovery phase - project_id = db.create_project( - user_id=1, - name="Test Project", - description="Test", - phase="discovery", - ) - - # Transition to planning - PhaseManager.transition(project_id, "planning", db) - - # Verify phase updated - project = db.get_project(project_id) - assert project["phase"] == "planning" - - def test_transition_invalid_raises_400(self, temp_db_path: Path): - """Invalid transition raises InvalidPhaseTransitionError.""" - db = Database(temp_db_path) - db.initialize() - setup_test_user(db, user_id=1) - - project_id = db.create_project( - user_id=1, - name="Test Project", - description="Test", - phase="discovery", - ) - - # Try invalid transition (discovery -> review) - with pytest.raises(InvalidPhaseTransitionError) as exc_info: - PhaseManager.transition(project_id, "review", db) - - assert exc_info.value.from_phase == "discovery" - assert exc_info.value.to_phase == "review" - assert "Invalid phase transition" in str(exc_info.value) - - def test_transition_nonexistent_project_raises_404(self, temp_db_path: Path): - """Transition on non-existent project raises ProjectNotFoundError.""" - db = Database(temp_db_path) - db.initialize() - - with pytest.raises(ProjectNotFoundError) as exc_info: - PhaseManager.transition(99999, "planning", db) - - assert exc_info.value.project_id == 99999 - - def test_transition_chain(self, temp_db_path: Path): - """Test full phase transition chain from discovery to complete.""" - db = Database(temp_db_path) - db.initialize() - setup_test_user(db, user_id=1) - - project_id = db.create_project( - user_id=1, - name="Test Project", - description="Test", - phase="discovery", - ) - - # Walk through the full lifecycle - transitions = ["planning", "active", "review", "complete"] - - for target_phase in transitions: - PhaseManager.transition(project_id, target_phase, db) - project = db.get_project(project_id) - assert project["phase"] == target_phase - - -class TestPhaseManagerGetPhaseRequirements: - """Tests for get_phase_requirements() method.""" - - def test_discovery_requirements(self): - """Get discovery phase requirements.""" - reqs = PhaseManager.get_phase_requirements("discovery") - - assert reqs["phase"] == "discovery" - assert reqs["steps"]["total"] == 4 - assert reqs["steps"]["description"] == "Discovery Phase" - assert "planning" in reqs["valid_next_phases"] - - def test_planning_requirements(self): - """Get planning phase requirements.""" - reqs = PhaseManager.get_phase_requirements("planning") - - assert reqs["phase"] == "planning" - assert reqs["steps"]["total"] == 4 - assert "active" in reqs["valid_next_phases"] - assert "discovery" in reqs["valid_next_phases"] - - def test_active_requirements(self): - """Get active phase requirements.""" - reqs = PhaseManager.get_phase_requirements("active") - - assert reqs["phase"] == "active" - assert reqs["steps"]["total"] == 5 - assert reqs["steps"]["description"] == "Development Phase" - - def test_review_requirements(self): - """Get review phase requirements.""" - reqs = PhaseManager.get_phase_requirements("review") - - assert reqs["phase"] == "review" - assert reqs["steps"]["total"] == 3 - assert "complete" in reqs["valid_next_phases"] - assert "active" in reqs["valid_next_phases"] - - def test_complete_requirements(self): - """Get complete phase requirements.""" - reqs = PhaseManager.get_phase_requirements("complete") - - assert reqs["phase"] == "complete" - assert reqs["steps"]["total"] == 1 - assert reqs["steps"]["description"] == "Complete" - - def test_invalid_phase_raises_error(self): - """Invalid phase raises ValueError.""" - with pytest.raises(ValueError) as exc_info: - PhaseManager.get_phase_requirements("invalid_phase") - - assert "Unknown phase" in str(exc_info.value) - - -class TestPhaseConfigurationCompleteness: - """Tests to ensure phase configuration is complete.""" - - def test_all_project_phases_have_transitions(self): - """All ProjectPhase enum values should be in VALID_TRANSITIONS.""" - for phase in ProjectPhase: - assert phase.value in VALID_TRANSITIONS, f"Missing transitions for {phase.value}" - - def test_all_project_phases_have_steps(self): - """All ProjectPhase enum values should be in PHASE_STEPS.""" - for phase in ProjectPhase: - assert phase.value in PHASE_STEPS, f"Missing steps for {phase.value}" - - def test_phase_steps_have_required_keys(self): - """Each phase step config has total and description.""" - for phase, steps in PHASE_STEPS.items(): - assert "total" in steps, f"Missing 'total' for {phase}" - assert "description" in steps, f"Missing 'description' for {phase}" - assert isinstance(steps["total"], int), f"'total' should be int for {phase}" - assert isinstance(steps["description"], str), f"'description' should be str for {phase}" - - def test_valid_transitions_reference_valid_phases(self): - """All phases in VALID_TRANSITIONS reference valid phases.""" - all_phases = set(VALID_TRANSITIONS.keys()) - - for from_phase, to_phases in VALID_TRANSITIONS.items(): - for to_phase in to_phases: - # All target phases should also be defined (except potentially 'shipped') - if to_phase not in all_phases: - # Allow 'shipped' as a terminal state not yet in enum - assert to_phase in ["shipped"], ( - f"Unknown target phase '{to_phase}' from '{from_phase}'" - ) diff --git a/tests/core/test_project_get_status.py b/tests/core/test_project_get_status.py deleted file mode 100644 index 74d353a8..00000000 --- a/tests/core/test_project_get_status.py +++ /dev/null @@ -1,623 +0,0 @@ -""" -Comprehensive tests for Project.get_status() method. - -Tests cover: -- Task statistics aggregation -- Agent counting (active/idle) -- Progress percentage calculation -- Blocker counting -- Quality metrics integration -- Last activity timestamp formatting -- Error handling (missing project, no database, empty data) -""" - -import pytest -from datetime import datetime, timezone, timedelta -from pathlib import Path -import tempfile - -from codeframe.core.project import Project -from codeframe.core.models import TaskStatus, Task, ProjectStatus -from codeframe.persistence.database import Database - - -@pytest.fixture -def test_db(): - """Create a temporary test database.""" - with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: - db_path = Path(f.name) - - db = Database(db_path) - db.initialize() - - yield db - - # Cleanup - db.conn.close() - if db_path.exists(): - db_path.unlink() - - -@pytest.fixture -def test_project_dir(): - """Create a temporary project directory.""" - with tempfile.TemporaryDirectory() as tmpdir: - project_dir = Path(tmpdir) - codeframe_dir = project_dir / ".codeframe" - codeframe_dir.mkdir(parents=True, exist_ok=True) - - yield project_dir - - -@pytest.fixture -def project_with_db(test_db, test_project_dir): - """Create a Project instance with a test database.""" - # Create project in database first - project_id = test_db.create_project( - name="test_project", status="active", description="Test project" - ) - - # Create project instance - project = Project(project_dir=test_project_dir) - project.db = test_db - project._status = ProjectStatus.ACTIVE - - # Save config - from codeframe.core.config import ProjectConfig - - config = ProjectConfig(project_name="test_project", project_type="python") - project.config.save(config) - - yield project, project_id, test_db - - -class TestGetStatusBasic: - """Test basic get_status() functionality.""" - - def test_no_database_returns_minimal_status(self, test_project_dir): - """Test that get_status() returns minimal status when database is not initialized.""" - project = Project(project_dir=test_project_dir) - from codeframe.core.config import ProjectConfig - - config = ProjectConfig(project_name="test_project", project_type="python") - project.config.save(config) - - status = project.get_status() - - assert status["id"] is None - assert status["name"] == "test_project" - assert status["status"] == "init" - assert status["tasks"]["total"] == 0 - assert status["agents"]["total"] == 0 - assert status["progress_pct"] == 0.0 - assert status["blockers"] == 0 - assert status["quality"] is None - assert status["last_activity"] == "No activity yet" - - def test_project_not_in_database_returns_minimal_status(self, project_with_db): - """Test that get_status() returns minimal status when project not found in database.""" - project, project_id, test_db = project_with_db - - # Change config to non-existent project - from codeframe.core.config import ProjectConfig - - config = ProjectConfig(project_name="nonexistent_project", project_type="python") - project.config.save(config) - - status = project.get_status() - - assert status["id"] is None - assert status["name"] == "nonexistent_project" - assert status["tasks"]["total"] == 0 - assert status["last_activity"] == "No activity yet" - - def test_empty_project_returns_zero_counts(self, project_with_db): - """Test that an empty project (no tasks, no agents) returns zero counts.""" - project, project_id, test_db = project_with_db - - status = project.get_status() - - assert status["id"] == project_id - assert status["name"] == "test_project" - assert status["tasks"] == { - "total": 0, - "completed": 0, - "in_progress": 0, - "blocked": 0, - "pending": 0, - } - assert status["agents"] == {"active": 0, "idle": 0, "total": 0} - assert status["progress_pct"] == 0.0 - assert status["blockers"] == 0 - assert status["last_activity"] == "No activity yet" - - -class TestTaskAggregation: - """Test task statistics aggregation.""" - - def test_mixed_task_statuses_aggregate_correctly(self, project_with_db): - """Test that tasks with mixed statuses are aggregated correctly.""" - project, project_id, test_db = project_with_db - - # Create tasks with different statuses - tasks_data = [ - {"status": TaskStatus.COMPLETED, "count": 5}, - {"status": TaskStatus.IN_PROGRESS, "count": 3}, - {"status": TaskStatus.BLOCKED, "count": 2}, - {"status": TaskStatus.PENDING, "count": 4}, - {"status": TaskStatus.ASSIGNED, "count": 1}, # Should count as pending - ] - - task_num = 1 - for task_data in tasks_data: - for _ in range(task_data["count"]): - test_db.create_task( - Task( - project_id=project_id, - task_number=f"1.{task_num}", - title=f"Task {task_num}", - description="Test task", - status=task_data["status"], - ) - ) - task_num += 1 - - status = project.get_status() - - assert status["tasks"]["total"] == 15 - assert status["tasks"]["completed"] == 5 - assert status["tasks"]["in_progress"] == 3 - assert status["tasks"]["blocked"] == 2 - assert status["tasks"]["pending"] == 5 # 4 PENDING + 1 ASSIGNED - - def test_failed_tasks_included_in_total(self, project_with_db): - """Test that FAILED tasks are included in total count.""" - project, project_id, test_db = project_with_db - - # Create a mix of tasks including FAILED - for i, status in enumerate([TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.PENDING], 1): - test_db.create_task( - Task( - project_id=project_id, - task_number=f"1.{i}", - title=f"Task {i}", - status=status, - ) - ) - - status = project.get_status() - - assert status["tasks"]["total"] == 3 - assert status["tasks"]["completed"] == 1 - assert status["tasks"]["pending"] == 1 - # FAILED tasks don't have a specific counter but are in total - - -class TestProgressCalculation: - """Test progress percentage calculation.""" - - def test_progress_0_percent_no_tasks(self, project_with_db): - """Test that progress is 0% when there are no tasks.""" - project, project_id, test_db = project_with_db - - status = project.get_status() - - assert status["progress_pct"] == 0.0 - - def test_progress_0_percent_no_completed_tasks(self, project_with_db): - """Test that progress is 0% when there are tasks but none completed.""" - project, project_id, test_db = project_with_db - - for i in range(1, 6): - test_db.create_task( - Task( - project_id=project_id, - task_number=f"1.{i}", - title=f"Task {i}", - status=TaskStatus.PENDING, - ) - ) - - status = project.get_status() - - assert status["progress_pct"] == 0.0 - - def test_progress_50_percent(self, project_with_db): - """Test that progress is 50% when half the tasks are completed.""" - project, project_id, test_db = project_with_db - - for i in range(1, 11): - status_val = TaskStatus.COMPLETED if i <= 5 else TaskStatus.PENDING - test_db.create_task( - Task( - project_id=project_id, - task_number=f"1.{i}", - title=f"Task {i}", - status=status_val, - ) - ) - - status = project.get_status() - - assert status["progress_pct"] == 50.0 - - def test_progress_100_percent(self, project_with_db): - """Test that progress is 100% when all tasks are completed.""" - project, project_id, test_db = project_with_db - - for i in range(1, 6): - test_db.create_task( - Task( - project_id=project_id, - task_number=f"1.{i}", - title=f"Task {i}", - status=TaskStatus.COMPLETED, - ) - ) - - status = project.get_status() - - assert status["progress_pct"] == 100.0 - - def test_progress_rounded_to_one_decimal(self, project_with_db): - """Test that progress percentage is rounded to 1 decimal place.""" - project, project_id, test_db = project_with_db - - # Create 7 tasks, 2 completed -> 28.571...% -> 28.6% - for i in range(1, 8): - status_val = TaskStatus.COMPLETED if i <= 2 else TaskStatus.PENDING - test_db.create_task( - Task( - project_id=project_id, - task_number=f"1.{i}", - title=f"Task {i}", - status=status_val, - ) - ) - - status = project.get_status() - - assert status["progress_pct"] == 28.6 - - -class TestAgentCounting: - """Test agent counting (active/idle).""" - - def test_no_agents_returns_zero_counts(self, project_with_db): - """Test that projects with no agents return zero counts.""" - project, project_id, test_db = project_with_db - - status = project.get_status() - - assert status["agents"] == {"active": 0, "idle": 0, "total": 0} - - def test_agents_with_status_working_counted_as_active(self, project_with_db): - """Test that agents with status='working' are counted as active.""" - project, project_id, test_db = project_with_db - - # Create agent and assign to project - agent_id = "agent-001" - from codeframe.core.models import AgentMaturity - - test_db.create_agent( - agent_id=agent_id, - agent_type="backend", - provider="anthropic", - maturity_level=AgentMaturity.D4, - ) - test_db.update_agent(agent_id, {"status": "working"}) - test_db.assign_agent_to_project(project_id, agent_id, role="backend") - - status = project.get_status() - - assert status["agents"]["total"] == 1 - assert status["agents"]["active"] == 1 - assert status["agents"]["idle"] == 0 - - def test_agents_with_current_task_counted_as_active(self, project_with_db): - """Test that agents with current_task_id are counted as active.""" - project, project_id, test_db = project_with_db - from codeframe.core.models import AgentMaturity - - # Create task - task = Task( - project_id=project_id, task_number="1.1", title="Task 1", status=TaskStatus.IN_PROGRESS - ) - task_id = test_db.create_task(task) - - # Create agent with current_task_id - agent_id = "agent-002" - test_db.create_agent( - agent_id=agent_id, - agent_type="backend", - provider="anthropic", - maturity_level=AgentMaturity.D4, - ) - test_db.update_agent(agent_id, {"status": "idle", "current_task_id": task_id}) - test_db.assign_agent_to_project(project_id, agent_id, role="backend") - - status = project.get_status() - - assert status["agents"]["total"] == 1 - assert status["agents"]["active"] == 1 - assert status["agents"]["idle"] == 0 - - def test_idle_agents_counted_correctly(self, project_with_db): - """Test that idle agents (no task, status=idle) are counted correctly.""" - project, project_id, test_db = project_with_db - from codeframe.core.models import AgentMaturity - - # Create idle agent - agent_id = "agent-003" - test_db.create_agent( - agent_id=agent_id, - agent_type="backend", - provider="anthropic", - maturity_level=AgentMaturity.D4, - ) - test_db.update_agent(agent_id, {"status": "idle", "current_task_id": None}) - test_db.assign_agent_to_project(project_id, agent_id, role="backend") - - status = project.get_status() - - assert status["agents"]["total"] == 1 - assert status["agents"]["active"] == 0 - assert status["agents"]["idle"] == 1 - - def test_mixed_active_idle_agents(self, project_with_db): - """Test counting with a mix of active and idle agents.""" - project, project_id, test_db = project_with_db - from codeframe.core.models import AgentMaturity - - # Create a task for agent-002 - task = Task( - project_id=project_id, task_number="1.1", title="Task 1", status=TaskStatus.IN_PROGRESS - ) - task_id = test_db.create_task(task) - - # Create 2 active agents and 1 idle agent - agents_data = [ - {"id": "agent-001", "status": "working", "task_id": None}, # Active (status=working) - {"id": "agent-002", "status": "idle", "task_id": task_id}, # Active (has task) - {"id": "agent-003", "status": "idle", "task_id": None}, # Idle - ] - - for agent_data in agents_data: - test_db.create_agent( - agent_id=agent_data["id"], - agent_type="backend", - provider="anthropic", - maturity_level=AgentMaturity.D4, - ) - test_db.update_agent( - agent_data["id"], - {"status": agent_data["status"], "current_task_id": agent_data["task_id"]}, - ) - test_db.assign_agent_to_project(project_id, agent_data["id"], role="backend") - - status = project.get_status() - - assert status["agents"]["total"] == 3 - assert status["agents"]["active"] == 2 - assert status["agents"]["idle"] == 1 - - -class TestBlockerCounting: - """Test blocker counting.""" - - def test_no_blockers_returns_zero(self, project_with_db): - """Test that projects with no blockers return zero.""" - project, project_id, test_db = project_with_db - - status = project.get_status() - - assert status["blockers"] == 0 - - def test_pending_blockers_counted(self, project_with_db): - """Test that pending blockers are counted correctly.""" - project, project_id, test_db = project_with_db - - # Create pending blockers - for i in range(1, 4): - test_db.create_blocker( - agent_id=f"agent-{i}", - project_id=project_id, - task_id=None, # No associated task - question=f"Question {i}?", - blocker_type="SYNC", - ) - - status = project.get_status() - - assert status["blockers"] == 3 - - def test_resolved_blockers_not_counted(self, project_with_db): - """Test that RESOLVED blockers are not counted.""" - project, project_id, test_db = project_with_db - - # Create pending and resolved blockers - blocker_id = test_db.create_blocker( - agent_id="agent-001", - project_id=project_id, - task_id=None, # No associated task - question="Question?", - blocker_type="SYNC", - ) - - # Resolve the blocker - test_db.resolve_blocker(blocker_id, answer="Answer") - - status = project.get_status() - - assert status["blockers"] == 0 - - -class TestQualityMetrics: - """Test quality metrics integration.""" - - def test_no_quality_data_returns_none(self, project_with_db): - """Test that projects with no quality data return None for quality metrics.""" - project, project_id, test_db = project_with_db - - status = project.get_status() - - assert status["quality"] is None - - def test_quality_metrics_retrieved_from_tracker(self, project_with_db): - """Test that quality metrics are retrieved from QualityTracker when available.""" - project, project_id, test_db = project_with_db - - # Create quality history file - from codeframe.enforcement.quality_tracker import QualityTracker, QualityMetrics - - tracker = QualityTracker(project_path=project.project_dir) - metrics = QualityMetrics( - timestamp=datetime.now(timezone.utc).isoformat(), - response_count=5, - test_pass_rate=95.5, - coverage_percentage=87.5, - total_tests=100, - passed_tests=95, - failed_tests=5, - ) - tracker.record(metrics) - - status = project.get_status() - - assert status["quality"] is not None - assert status["quality"]["test_pass_rate"] == 95.5 - assert status["quality"]["coverage_pct"] == 87.5 - - -class TestLastActivityFormatting: - """Test last activity timestamp formatting.""" - - def test_no_activity_returns_no_activity_yet(self, project_with_db): - """Test that projects with no activity return 'No activity yet'.""" - project, project_id, test_db = project_with_db - - status = project.get_status() - - assert status["last_activity"] == "No activity yet" - - def test_recent_activity_formatted_as_just_now(self, project_with_db): - """Test that activity within the last minute is formatted as 'just now'.""" - project, project_id, test_db = project_with_db - - # Create recent activity (within last 30 seconds) - cursor = test_db.conn.cursor() - timestamp = (datetime.now(timezone.utc) - timedelta(seconds=30)).isoformat() - cursor.execute( - """ - INSERT INTO changelog (project_id, agent_id, action, timestamp) - VALUES (?, ?, ?, ?) - """, - (project_id, "agent-001", "test_action", timestamp), - ) - test_db.conn.commit() - - status = project.get_status() - - assert status["last_activity"] == "just now" - - def test_activity_formatted_as_minutes_ago(self, project_with_db): - """Test that activity within the last hour is formatted as 'X minutes ago'.""" - project, project_id, test_db = project_with_db - - # Manually insert activity with timestamp 5 minutes ago - cursor = test_db.conn.cursor() - timestamp = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat() - cursor.execute( - """ - INSERT INTO changelog (project_id, agent_id, action, timestamp) - VALUES (?, ?, ?, ?) - """, - (project_id, "agent-001", "test_action", timestamp), - ) - test_db.conn.commit() - - status = project.get_status() - - assert "minute" in status["last_activity"] - assert "ago" in status["last_activity"] - - def test_activity_formatted_as_hours_ago(self, project_with_db): - """Test that activity within the last day is formatted as 'X hours ago'.""" - project, project_id, test_db = project_with_db - - # Manually insert activity with timestamp 3 hours ago - cursor = test_db.conn.cursor() - timestamp = (datetime.now(timezone.utc) - timedelta(hours=3)).isoformat() - cursor.execute( - """ - INSERT INTO changelog (project_id, agent_id, action, timestamp) - VALUES (?, ?, ?, ?) - """, - (project_id, "agent-001", "test_action", timestamp), - ) - test_db.conn.commit() - - status = project.get_status() - - assert "hour" in status["last_activity"] - assert "ago" in status["last_activity"] - - def test_activity_formatted_as_days_ago(self, project_with_db): - """Test that activity older than a day is formatted as 'X days ago'.""" - project, project_id, test_db = project_with_db - - # Manually insert activity with timestamp 2 days ago - cursor = test_db.conn.cursor() - timestamp = (datetime.now(timezone.utc) - timedelta(days=2)).isoformat() - cursor.execute( - """ - INSERT INTO changelog (project_id, agent_id, action, timestamp) - VALUES (?, ?, ?, ?) - """, - (project_id, "agent-001", "test_action", timestamp), - ) - test_db.conn.commit() - - status = project.get_status() - - assert "day" in status["last_activity"] - assert "ago" in status["last_activity"] - - def test_singular_plural_formatting(self, project_with_db): - """Test that singular/plural formatting is correct (1 minute vs 2 minutes).""" - project, project_id, test_db = project_with_db - - # Manually insert activity with timestamp 1 minute ago - cursor = test_db.conn.cursor() - timestamp = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat() - cursor.execute( - """ - INSERT INTO changelog (project_id, agent_id, action, timestamp) - VALUES (?, ?, ?, ?) - """, - (project_id, "agent-001", "test_action", timestamp), - ) - test_db.conn.commit() - - status = project.get_status() - - assert "1 minute ago" in status["last_activity"] - - -class TestErrorHandling: - """Test error handling.""" - - def test_database_error_returns_error_status(self, project_with_db): - """Test that database errors return a valid error status.""" - project, project_id, test_db = project_with_db - - # Close the database connection to simulate error - test_db.conn.close() - - status = project.get_status() - - # Should still return a valid dictionary with error field - assert isinstance(status, dict) - assert "error" in status - assert status["tasks"]["total"] == 0 - assert status["last_activity"] == "Error retrieving activity" diff --git a/tests/persistence/test_intervention_context.py b/tests/persistence/test_intervention_context.py deleted file mode 100644 index c7dd663d..00000000 --- a/tests/persistence/test_intervention_context.py +++ /dev/null @@ -1,225 +0,0 @@ -""" -Tests for intervention context database methods. - -Test coverage for supervisor intervention persistence: -- Setting intervention context on tasks -- Retrieving intervention context -- Clearing intervention context -- JSON serialization/deserialization - -Following strict TDD methodology (RED-GREEN-REFACTOR). -""" - -import pytest -from codeframe.persistence.database import Database -from codeframe.core.models import TaskStatus - -pytestmark = pytest.mark.v2 - - -class TestInterventionContextMethods: - """Test intervention context database operations.""" - - @pytest.fixture - def db_with_task(self, tmp_path): - """Create database with a test task.""" - db = Database(":memory:") - db.initialize() - - # Create project - project_id = db.create_project("test", "Test project") - - # Create issue - issue_id = db.create_issue({ - "project_id": project_id, - "issue_number": "1.0", - "title": "Test Issue", - "status": "pending", - "priority": 0, - "workflow_step": 1, - }) - - # Create task - task_id = db.create_task_with_issue( - project_id=project_id, - issue_id=issue_id, - task_number="1.0.1", - parent_issue_number="1.0", - title="Test Task", - description="Test task description", - status=TaskStatus.PENDING, - priority=0, - workflow_step=1, - can_parallelize=False, - ) - - return db, task_id - - def test_update_intervention_context_sets_value(self, db_with_task): - """Test that intervention context can be set on a task.""" - db, task_id = db_with_task - - context = { - "intervention_applied": True, - "pattern_matched": "file_already_exists", - "existing_files": ["src/components/Button.tsx"], - "instruction": "Use edit operations for existing files", - "strategy": "convert_create_to_edit", - } - - db.update_task_intervention_context(task_id, context) - - # Verify it was set - result = db.get_task_intervention_context(task_id) - assert result is not None - assert result["intervention_applied"] is True - assert result["pattern_matched"] == "file_already_exists" - assert result["existing_files"] == ["src/components/Button.tsx"] - - def test_get_intervention_context_returns_none_when_not_set(self, db_with_task): - """Test that get returns None when no context is set.""" - db, task_id = db_with_task - - result = db.get_task_intervention_context(task_id) - - assert result is None - - def test_clear_intervention_context_removes_value(self, db_with_task): - """Test that intervention context can be cleared.""" - db, task_id = db_with_task - - # Set context first - context = { - "intervention_applied": True, - "pattern_matched": "file_already_exists", - } - db.update_task_intervention_context(task_id, context) - - # Verify it was set - assert db.get_task_intervention_context(task_id) is not None - - # Clear it - db.clear_task_intervention_context(task_id) - - # Verify it was cleared - assert db.get_task_intervention_context(task_id) is None - - def test_intervention_context_handles_complex_structure(self, db_with_task): - """Test that complex nested structures are preserved.""" - db, task_id = db_with_task - - context = { - "intervention_applied": True, - "pattern_matched": "file_already_exists", - "existing_files": [ - "src/components/Button.tsx", - "src/components/Header.tsx", - "src/utils/helpers.py", - ], - "instruction": "Use edit operations for existing files", - "strategy": "convert_create_to_edit", - "metadata": { - "attempt_count": 2, - "previous_errors": ["FileExistsError: Button.tsx"], - }, - } - - db.update_task_intervention_context(task_id, context) - - result = db.get_task_intervention_context(task_id) - - assert result["existing_files"] == [ - "src/components/Button.tsx", - "src/components/Header.tsx", - "src/utils/helpers.py", - ] - assert result["metadata"]["attempt_count"] == 2 - assert "FileExistsError" in result["metadata"]["previous_errors"][0] - - def test_intervention_context_can_be_updated(self, db_with_task): - """Test that intervention context can be overwritten.""" - db, task_id = db_with_task - - # Set initial context - initial_context = { - "intervention_applied": True, - "existing_files": ["file1.py"], - } - db.update_task_intervention_context(task_id, initial_context) - - # Update with new context - updated_context = { - "intervention_applied": True, - "existing_files": ["file1.py", "file2.py"], - "retry_count": 2, - } - db.update_task_intervention_context(task_id, updated_context) - - result = db.get_task_intervention_context(task_id) - - assert result["existing_files"] == ["file1.py", "file2.py"] - assert result["retry_count"] == 2 - - def test_intervention_context_update_via_update_task(self, db_with_task): - """Test that intervention_context can be set via generic update_task.""" - db, task_id = db_with_task - import json - - context = { - "intervention_applied": True, - "pattern_matched": "file_already_exists", - } - - # Update using the generic update_task method - db.update_task(task_id, {"intervention_context": json.dumps(context)}) - - # Verify it was set (need to parse JSON manually here since update_task - # doesn't parse it) - result = db.get_task_intervention_context(task_id) - - assert result is not None - assert result["intervention_applied"] is True - - def test_get_intervention_context_nonexistent_task(self, db_with_task): - """Test that get returns None for nonexistent task.""" - db, _ = db_with_task - - result = db.get_task_intervention_context(99999) - - assert result is None - - def test_get_task_includes_intervention_context(self, db_with_task): - """Test that db.get_task() returns Task with intervention_context populated. - - This verifies the _row_to_task() deserialization round-trip, which is - critical for LeadAgent re-fetching tasks after intervention is applied. - """ - db, task_id = db_with_task - - context = { - "intervention_applied": True, - "pattern_matched": "file_already_exists", - "existing_files": ["src/app.py"], - "strategy": "convert_create_to_edit", - "intervention_retry_count": 1, - } - db.update_task_intervention_context(task_id, context) - - # Re-fetch via get_task (uses _row_to_task internally) - task = db.get_task(task_id) - - assert task is not None - assert task.intervention_context is not None - assert task.intervention_context["intervention_applied"] is True - assert task.intervention_context["strategy"] == "convert_create_to_edit" - assert task.intervention_context["existing_files"] == ["src/app.py"] - assert task.intervention_context["intervention_retry_count"] == 1 - - def test_get_task_returns_none_intervention_context_when_unset(self, db_with_task): - """Task.intervention_context is None when no context has been set.""" - db, task_id = db_with_task - - task = db.get_task(task_id) - - assert task is not None - assert task.intervention_context is None diff --git a/tests/persistence/test_token_repository.py b/tests/persistence/test_token_repository.py deleted file mode 100644 index 65b50f7a..00000000 --- a/tests/persistence/test_token_repository.py +++ /dev/null @@ -1,214 +0,0 @@ -"""Tests for TokenRepository query methods (Issue #314 Step 3). - -Tests for: -- get_task_token_summary: SQL aggregate for a single task -- get_batch_token_usage: Filter by list of task_ids -- get_workspace_token_usage: All records, no project filter -""" - -import pytest -from datetime import datetime, timedelta, timezone - -from codeframe.core.models import CallType, TokenUsage -from codeframe.persistence.database import Database - -pytestmark = pytest.mark.v2 - - -@pytest.fixture -def db(): - """Create in-memory database for testing.""" - database = Database(":memory:") - database.initialize() - - # Create test project - cursor = database.conn.cursor() - cursor.execute( - "INSERT INTO projects (name, description, workspace_path, status) VALUES (?, ?, ?, ?)", - ("test-project", "Test project", "/tmp/test", "active"), - ) - database.conn.commit() - - return database - - -def _create_task(db, project_id=1, task_id_hint=None): - """Helper to create a task and return its ID.""" - cursor = db.conn.cursor() - cursor.execute( - "INSERT INTO tasks (project_id, title, description, status) VALUES (?, ?, ?, ?)", - (project_id, f"Task {task_id_hint or 'x'}", "Test task", "in_progress"), - ) - db.conn.commit() - return cursor.lastrowid - - -def _save_usage(db, task_id=None, agent_id="agent-001", project_id=1, - model_name="claude-sonnet-4-5", input_tokens=1000, - output_tokens=500, cost=0.0105, call_type=CallType.TASK_EXECUTION, - timestamp=None): - """Helper to save a token usage record.""" - if timestamp is None: - timestamp = datetime.now(timezone.utc) - usage = TokenUsage( - task_id=task_id, - agent_id=agent_id, - project_id=project_id, - model_name=model_name, - input_tokens=input_tokens, - output_tokens=output_tokens, - estimated_cost_usd=cost, - actual_cost_usd=None, - call_type=call_type, - timestamp=timestamp, - ) - return db.save_token_usage(usage) - - -# ============================================================================ -# get_task_token_summary -# ============================================================================ - - -def test_get_task_token_summary_single_call(db): - """Test task summary with a single LLM call.""" - tid = _create_task(db) - _save_usage(db, task_id=tid, input_tokens=1000, output_tokens=500, cost=0.0105) - - summary = db.get_task_token_summary(task_id=tid) - - assert summary["task_id"] == tid - assert summary["total_input_tokens"] == 1000 - assert summary["total_output_tokens"] == 500 - assert summary["total_tokens"] == 1500 - assert summary["total_cost_usd"] == pytest.approx(0.0105, abs=1e-6) - assert summary["call_count"] == 1 - - -def test_get_task_token_summary_multiple_calls(db): - """Test task summary aggregates multiple LLM calls.""" - tid = _create_task(db) - _save_usage(db, task_id=tid, input_tokens=1000, output_tokens=500, cost=0.01) - _save_usage(db, task_id=tid, input_tokens=2000, output_tokens=1000, cost=0.02) - - summary = db.get_task_token_summary(task_id=tid) - - assert summary["total_input_tokens"] == 3000 - assert summary["total_output_tokens"] == 1500 - assert summary["total_tokens"] == 4500 - assert summary["total_cost_usd"] == pytest.approx(0.03, abs=1e-6) - assert summary["call_count"] == 2 - - -def test_get_task_token_summary_no_records(db): - """Test task summary returns zeros when no records exist.""" - summary = db.get_task_token_summary(task_id=999) - - assert summary["task_id"] == 999 - assert summary["total_input_tokens"] == 0 - assert summary["total_output_tokens"] == 0 - assert summary["total_tokens"] == 0 - assert summary["total_cost_usd"] == 0.0 - assert summary["call_count"] == 0 - - -def test_get_task_token_summary_excludes_other_tasks(db): - """Test task summary only includes records for the specified task.""" - tid1 = _create_task(db) - tid2 = _create_task(db) - _save_usage(db, task_id=tid1, input_tokens=1000, output_tokens=500, cost=0.01) - _save_usage(db, task_id=tid2, input_tokens=2000, output_tokens=1000, cost=0.02) - - summary = db.get_task_token_summary(task_id=tid1) - - assert summary["total_input_tokens"] == 1000 - assert summary["call_count"] == 1 - - -# ============================================================================ -# get_batch_token_usage -# ============================================================================ - - -def test_get_batch_token_usage(db): - """Test getting token usage for a batch of task IDs.""" - tid1 = _create_task(db) - tid2 = _create_task(db) - tid3 = _create_task(db) - _save_usage(db, task_id=tid1, input_tokens=100, output_tokens=50, cost=0.001) - _save_usage(db, task_id=tid2, input_tokens=200, output_tokens=100, cost=0.002) - _save_usage(db, task_id=tid3, input_tokens=300, output_tokens=150, cost=0.003) - - records = db.get_batch_token_usage(task_ids=[tid1, tid2]) - - assert len(records) == 2 - task_ids = {r["task_id"] for r in records} - assert task_ids == {tid1, tid2} - - -def test_get_batch_token_usage_with_date_filter(db): - """Test batch token usage with date filtering.""" - now = datetime.now(timezone.utc) - old = now - timedelta(days=10) - - tid1 = _create_task(db) - tid2 = _create_task(db) - _save_usage(db, task_id=tid1, input_tokens=100, output_tokens=50, cost=0.001, timestamp=now) - _save_usage(db, task_id=tid2, input_tokens=200, output_tokens=100, cost=0.002, timestamp=old) - - start = now - timedelta(days=1) - records = db.get_batch_token_usage(task_ids=[tid1, tid2], start_date=start) - - assert len(records) == 1 - assert records[0]["task_id"] == tid1 - - -def test_get_batch_token_usage_empty_list(db): - """Test batch token usage with empty task ID list.""" - tid = _create_task(db) - _save_usage(db, task_id=tid, input_tokens=100, output_tokens=50, cost=0.001) - - records = db.get_batch_token_usage(task_ids=[]) - - assert len(records) == 0 - - -# ============================================================================ -# get_workspace_token_usage -# ============================================================================ - - -def test_get_workspace_token_usage(db): - """Test getting all token usage across the workspace.""" - tid = _create_task(db) - _save_usage(db, task_id=tid, project_id=1, input_tokens=100, output_tokens=50, cost=0.001) - _save_usage(db, task_id=None, project_id=1, input_tokens=200, output_tokens=100, cost=0.002) - - records = db.get_workspace_token_usage() - - assert len(records) == 2 - - -def test_get_workspace_token_usage_with_date_filter(db): - """Test workspace token usage with date filtering.""" - now = datetime.now(timezone.utc) - old = now - timedelta(days=10) - - tid1 = _create_task(db) - tid2 = _create_task(db) - _save_usage(db, task_id=tid1, input_tokens=100, output_tokens=50, cost=0.001, timestamp=now) - _save_usage(db, task_id=tid2, input_tokens=200, output_tokens=100, cost=0.002, timestamp=old) - - start = now - timedelta(days=1) - end = now + timedelta(days=1) - records = db.get_workspace_token_usage(start_date=start, end_date=end) - - assert len(records) == 1 - assert records[0]["task_id"] == tid1 - - -def test_get_workspace_token_usage_empty(db): - """Test workspace token usage when no records exist.""" - records = db.get_workspace_token_usage() - - assert len(records) == 0 diff --git a/tests/persistence/test_token_repository_costs.py b/tests/persistence/test_token_repository_costs.py deleted file mode 100644 index 2cff7a3f..00000000 --- a/tests/persistence/test_token_repository_costs.py +++ /dev/null @@ -1,388 +0,0 @@ -"""Tests for TokenRepository.get_costs_summary (Issue #557). - -The method aggregates token_usage rows into daily buckets for the -cost analytics page. Returns total spend, total tasks, average cost -per task, and a daily series filled with zeros where no data exists. -""" - -import pytest -from datetime import datetime, timedelta, timezone - -from codeframe.core.models import CallType, TokenUsage -from codeframe.persistence.database import Database - -pytestmark = pytest.mark.v2 - - -@pytest.fixture -def db(): - database = Database(":memory:") - database.initialize() - cursor = database.conn.cursor() - cursor.execute( - "INSERT INTO projects (name, description, workspace_path, status) VALUES (?, ?, ?, ?)", - ("test-project", "Test project", "/tmp/test", "active"), - ) - database.conn.commit() - return database - - -def _create_task(db, project_id=1, title="Task"): - cursor = db.conn.cursor() - cursor.execute( - "INSERT INTO tasks (project_id, title, description, status) VALUES (?, ?, ?, ?)", - (project_id, title, "Test", "in_progress"), - ) - db.conn.commit() - return cursor.lastrowid - - -def _save(db, task_id=None, cost=0.01, timestamp=None, project_id=1): - if timestamp is None: - timestamp = datetime.now(timezone.utc) - usage = TokenUsage( - task_id=task_id, - agent_id="agent-001", - project_id=project_id, - model_name="claude-sonnet-4-5", - input_tokens=100, - output_tokens=50, - estimated_cost_usd=cost, - actual_cost_usd=None, - call_type=CallType.TASK_EXECUTION, - timestamp=timestamp, - ) - return db.save_token_usage(usage) - - -class TestGetCostsSummaryEmpty: - def test_empty_table_returns_zeros(self, db): - summary = db.token_usage.get_costs_summary(days=30) - - assert summary["total_spend_usd"] == 0.0 - assert summary["total_tasks"] == 0 - assert summary["avg_cost_per_task"] == 0.0 - # daily should have one entry per day in the range - assert len(summary["daily"]) == 30 - assert all(d["cost_usd"] == 0.0 for d in summary["daily"]) - - def test_default_days_is_30(self, db): - summary = db.token_usage.get_costs_summary(days=30) - assert len(summary["daily"]) == 30 - - -class TestGetCostsSummaryWithData: - def test_aggregates_total_spend(self, db): - t1 = _create_task(db) - t2 = _create_task(db) - now = datetime.now(timezone.utc) - _save(db, task_id=t1, cost=0.50, timestamp=now) - _save(db, task_id=t1, cost=0.25, timestamp=now) - _save(db, task_id=t2, cost=0.30, timestamp=now) - - summary = db.token_usage.get_costs_summary(days=30) - - assert summary["total_spend_usd"] == pytest.approx(1.05) - assert summary["total_tasks"] == 2 # distinct task_ids - assert summary["avg_cost_per_task"] == pytest.approx(1.05 / 2) - - def test_excludes_null_task_ids_from_count(self, db): - t1 = _create_task(db) - now = datetime.now(timezone.utc) - _save(db, task_id=t1, cost=0.10, timestamp=now) - _save(db, task_id=None, cost=0.10, timestamp=now) # standalone call - - summary = db.token_usage.get_costs_summary(days=30) - - assert summary["total_spend_usd"] == pytest.approx(0.20) - assert summary["total_tasks"] == 1 - - def test_daily_buckets_filled_with_zeros(self, db): - t1 = _create_task(db) - # Two records on different days within the range - now = datetime.now(timezone.utc) - _save(db, task_id=t1, cost=0.10, timestamp=now) - _save(db, task_id=t1, cost=0.20, timestamp=now - timedelta(days=3)) - - summary = db.token_usage.get_costs_summary(days=7) - - assert len(summary["daily"]) == 7 - # All entries have date keys and cost_usd keys - for entry in summary["daily"]: - assert "date" in entry - assert "cost_usd" in entry - # Sum of daily matches total - assert sum(d["cost_usd"] for d in summary["daily"]) == pytest.approx(0.30) - - def test_excludes_data_outside_window(self, db): - t1 = _create_task(db) - now = datetime.now(timezone.utc) - _save(db, task_id=t1, cost=0.10, timestamp=now) - # 100 days ago — outside 30-day window - _save(db, task_id=t1, cost=99.0, timestamp=now - timedelta(days=100)) - - summary = db.token_usage.get_costs_summary(days=30) - - assert summary["total_spend_usd"] == pytest.approx(0.10) - - def test_excludes_future_dated_rows(self, db): - """A row with a timestamp past today must not inflate the KPI cards. - - Without an upper bound the daily chart (which is built from a fixed - list of dates within the window) would exclude future rows while the - SUM() KPIs would include them, making the two views disagree. - """ - t1 = _create_task(db) - now = datetime.now(timezone.utc) - _save(db, task_id=t1, cost=0.10, timestamp=now) - _save(db, task_id=t1, cost=42.0, timestamp=now + timedelta(days=2)) - - summary = db.token_usage.get_costs_summary(days=7) - - assert summary["total_spend_usd"] == pytest.approx(0.10) - # And the daily series sum agrees with the KPI total - assert sum(d["cost_usd"] for d in summary["daily"]) == pytest.approx(0.10) - - def test_daily_dates_are_ordered_oldest_to_newest(self, db): - summary = db.token_usage.get_costs_summary(days=7) - dates = [d["date"] for d in summary["daily"]] - assert dates == sorted(dates) - - def test_avg_cost_per_task_zero_when_no_tasks(self, db): - # Record exists but has NULL task_id - now = datetime.now(timezone.utc) - _save(db, task_id=None, cost=0.50, timestamp=now) - - summary = db.token_usage.get_costs_summary(days=30) - - assert summary["total_spend_usd"] == pytest.approx(0.50) - assert summary["total_tasks"] == 0 - assert summary["avg_cost_per_task"] == 0.0 - - -class TestGetCostsSummaryTimestampFormats: - """Records inserted via different timestamp formats must all be picked up. - - SQLite's `CURRENT_TIMESTAMP` produces space-separated values - ("YYYY-MM-DD HH:MM:SS"), Python `.isoformat()` produces T-separated - values with an offset suffix ("YYYY-MM-DDTHH:MM:SS+00:00"). The query - must include both. - """ - - def test_includes_records_with_space_separated_timestamps(self, db): - """A record inserted with SQLite's default timestamp format must be counted.""" - tid = _create_task(db) - # Insert raw with a space-separated timestamp (the schema default format). - # This simulates DEFAULT CURRENT_TIMESTAMP behavior. - now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") - cursor = db.conn.cursor() - cursor.execute( - """ - INSERT INTO token_usage (task_id, agent_id, project_id, model_name, - input_tokens, output_tokens, estimated_cost_usd, call_type, timestamp) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - (tid, "agent-001", 1, "claude-sonnet-4-5", - 100, 50, 0.42, "task_execution", now_str), - ) - db.conn.commit() - - summary = db.token_usage.get_costs_summary(days=7) - - assert summary["total_spend_usd"] == pytest.approx(0.42) - assert summary["total_tasks"] == 1 - - -class TestGetCostsSummaryRangeValidation: - @pytest.mark.parametrize("days", [7, 30, 90]) - def test_valid_ranges(self, db, days): - summary = db.token_usage.get_costs_summary(days=days) - assert len(summary["daily"]) == days - - -# --------------------------------------------------------------------------- -# get_top_tasks_by_cost (Issue #558) — per-task cost breakdown -# --------------------------------------------------------------------------- - - -def _save_with_agent( - db, task_id, cost, agent_id="agent-001", project_id=1, timestamp=None, - input_tokens=100, output_tokens=50, -): - if timestamp is None: - timestamp = datetime.now(timezone.utc) - usage = TokenUsage( - task_id=task_id, - agent_id=agent_id, - project_id=project_id, - model_name="claude-sonnet-4-5", - input_tokens=input_tokens, - output_tokens=output_tokens, - estimated_cost_usd=cost, - actual_cost_usd=None, - call_type=CallType.TASK_EXECUTION, - timestamp=timestamp, - ) - return db.save_token_usage(usage) - - -class TestGetTopTasksByCost: - def test_empty_returns_empty_list(self, db): - result = db.token_usage.get_top_tasks_by_cost(days=30) - assert result == [] - - def test_aggregates_cost_per_task(self, db): - t1 = _create_task(db) - t2 = _create_task(db) - _save_with_agent(db, task_id=t1, cost=0.25) - _save_with_agent(db, task_id=t1, cost=0.50) - _save_with_agent(db, task_id=t2, cost=0.10) - - result = db.token_usage.get_top_tasks_by_cost(days=30) - - # Sorted by cost desc - assert len(result) == 2 - assert result[0]["task_id"] == t1 - assert result[0]["total_cost_usd"] == pytest.approx(0.75) - assert result[0]["input_tokens"] == 200 - assert result[0]["output_tokens"] == 100 - assert result[1]["task_id"] == t2 - assert result[1]["total_cost_usd"] == pytest.approx(0.10) - - def test_includes_most_used_agent_per_task(self, db): - """Task aggregates should report the agent with the most calls.""" - t1 = _create_task(db) - _save_with_agent(db, task_id=t1, cost=0.10, agent_id="react-agent") - _save_with_agent(db, task_id=t1, cost=0.10, agent_id="react-agent") - _save_with_agent(db, task_id=t1, cost=0.10, agent_id="other-agent") - - result = db.token_usage.get_top_tasks_by_cost(days=30) - - assert result[0]["agent_id"] == "react-agent" - - def test_excludes_null_task_ids(self, db): - t1 = _create_task(db) - _save_with_agent(db, task_id=t1, cost=0.10) - _save_with_agent(db, task_id=None, cost=99.0) - - result = db.token_usage.get_top_tasks_by_cost(days=30) - - assert len(result) == 1 - assert result[0]["task_id"] == t1 - - def test_respects_limit(self, db): - for _ in range(15): - tid = _create_task(db) - _save_with_agent(db, task_id=tid, cost=0.01) - - result = db.token_usage.get_top_tasks_by_cost(days=30, limit=10) - assert len(result) == 10 - - def test_excludes_data_outside_window(self, db): - t1 = _create_task(db) - now = datetime.now(timezone.utc) - _save_with_agent(db, task_id=t1, cost=0.10, timestamp=now) - _save_with_agent(db, task_id=t1, cost=99.0, timestamp=now - timedelta(days=100)) - - result = db.token_usage.get_top_tasks_by_cost(days=30) - - assert result[0]["total_cost_usd"] == pytest.approx(0.10) - - def test_supports_text_task_ids(self, db): - """SQLite is type-flexible: TEXT (UUID) task_ids must be aggregated correctly. - - v2 workspaces store task UUIDs in the same INTEGER-declared column; the - aggregation has to group by the raw value without type coercion. The v1 - Database fixture enforces FK(token_usage.task_id → tasks.id), so we - relax it for this test to model the v2 schema where token_usage has no - such constraint. - """ - uuid_a = "task-uuid-aaaa" - uuid_b = "task-uuid-bbbb" - cursor = db.conn.cursor() - cursor.execute("PRAGMA foreign_keys = OFF") - try: - for tid, cost in [(uuid_a, 0.50), (uuid_a, 0.25), (uuid_b, 0.10)]: - cursor.execute( - """ - INSERT INTO token_usage (task_id, agent_id, project_id, model_name, - input_tokens, output_tokens, estimated_cost_usd, call_type, timestamp) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - (tid, "react-agent", 1, "claude-sonnet-4-5", - 100, 50, cost, "task_execution", - datetime.now(timezone.utc).isoformat()), - ) - db.conn.commit() - finally: - cursor.execute("PRAGMA foreign_keys = ON") - - result = db.token_usage.get_top_tasks_by_cost(days=30) - - assert len(result) == 2 - assert result[0]["task_id"] == uuid_a - assert result[0]["total_cost_usd"] == pytest.approx(0.75) - - -# --------------------------------------------------------------------------- -# get_costs_by_agent (Issue #558) — per-agent cost breakdown -# --------------------------------------------------------------------------- - - -class TestGetCostsByAgent: - def test_empty_returns_zero_state(self, db): - result = db.token_usage.get_costs_by_agent(days=30) - assert result["by_agent"] == [] - assert result["total_input_tokens"] == 0 - assert result["total_output_tokens"] == 0 - - def test_aggregates_cost_per_agent(self, db): - t1 = _create_task(db) - _save_with_agent(db, task_id=t1, cost=0.30, agent_id="claude-code") - _save_with_agent(db, task_id=t1, cost=0.20, agent_id="claude-code") - _save_with_agent(db, task_id=t1, cost=0.40, agent_id="codex") - - result = db.token_usage.get_costs_by_agent(days=30) - - # Sorted by cost desc - agents = result["by_agent"] - assert len(agents) == 2 - assert agents[0]["agent_id"] == "claude-code" - assert agents[0]["total_cost_usd"] == pytest.approx(0.50) - assert agents[0]["call_count"] == 2 - assert agents[0]["input_tokens"] == 200 - assert agents[0]["output_tokens"] == 100 - assert agents[1]["agent_id"] == "codex" - assert agents[1]["total_cost_usd"] == pytest.approx(0.40) - - def test_includes_null_task_records(self, db): - """Per-agent totals should include calls not linked to a task.""" - _save_with_agent(db, task_id=None, cost=0.10, agent_id="solo-agent") - - result = db.token_usage.get_costs_by_agent(days=30) - - assert len(result["by_agent"]) == 1 - assert result["by_agent"][0]["agent_id"] == "solo-agent" - - def test_totals_match_sum_of_agents(self, db): - t1 = _create_task(db) - _save_with_agent(db, task_id=t1, cost=0.10, - agent_id="a", input_tokens=100, output_tokens=50) - _save_with_agent(db, task_id=t1, cost=0.10, - agent_id="b", input_tokens=200, output_tokens=75) - - result = db.token_usage.get_costs_by_agent(days=30) - - assert result["total_input_tokens"] == 300 - assert result["total_output_tokens"] == 125 - - def test_excludes_data_outside_window(self, db): - t1 = _create_task(db) - now = datetime.now(timezone.utc) - _save_with_agent(db, task_id=t1, cost=0.10, agent_id="a", timestamp=now) - _save_with_agent(db, task_id=t1, cost=99.0, agent_id="a", - timestamp=now - timedelta(days=100)) - - result = db.token_usage.get_costs_by_agent(days=30) - - assert result["by_agent"][0]["total_cost_usd"] == pytest.approx(0.10)