diff --git a/backend/__init__.py b/backend/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/backend/agents/__init__.py b/backend/agents/__init__.py deleted file mode 100644 index cb8524d81ab4c099ac171dda8164ef4e34a0a531..0000000000000000000000000000000000000000 --- a/backend/agents/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Agent modules — planner and task handlers.""" diff --git a/backend/agents/code_agent.py b/backend/agents/code_agent.py deleted file mode 100644 index feecfe4475f5ce96e3840988149441c455ebe6b4..0000000000000000000000000000000000000000 --- a/backend/agents/code_agent.py +++ /dev/null @@ -1,95 +0,0 @@ -from typing import Any - -from pydantic import BaseModel, Field - - -class GeneratedFile(BaseModel): - path: str - content: str - language: str = "python" - - -class CodeGenerationResult(BaseModel): - files: dict[str, dict[str, str]] - status: str = "generated" - - @classmethod - def from_llm_payload(cls, payload: dict[str, Any]) -> "CodeGenerationResult": - files: dict[str, dict[str, str]] = {} - for item in payload.get("files", []): - path = item["path"].lstrip("/").replace("\\", "/") - files[path] = { - "content": item["content"], - "language": item.get("language", "python"), - } - if not files: - raise ValueError("Code generation produced no files") - return cls(files=files) - - -CODE_GENERATION_SCHEMA: dict[str, Any] = { - "type": "object", - "properties": { - "files": { - "type": "array", - "items": { - "type": "object", - "properties": { - "path": {"type": "string"}, - "content": {"type": "string"}, - "language": {"type": "string"}, - }, - "required": ["path", "content"], - }, - } - }, - "required": ["files"], -} - -CODE_GENERATION_SYSTEM_PROMPT = """You are a code generation agent. -Generate a complete, runnable Python project for the user request. -Return JSON with a files array. Each file must include path, content, and language. - -Requirements: -- Use a flat or package layout that pytest can import (e.g. calculator.py or calculator/__init__.py). -- Include requirements.txt with only third-party packages (pytest is pre-installed in the sandbox). -- For stdlib-only projects, requirements.txt may be empty or contain only comments. -- Do not include tests — a separate agent will generate them. -Return only valid JSON.""" - - -from backend.llm.base import LLMProvider -from backend.orchestration.errors import TaskExecutionError -from backend.orchestration.task_context import TaskContext - - -class CodeGenerationAgent: - def __init__(self, llm: LLMProvider) -> None: - self.llm = llm - - async def generate(self, context: TaskContext) -> dict[str, Any]: - prompt = ( - f"User request:\n{context.user_prompt}\n\n" - "Generate the implementation files for this project." - ) - try: - structured = await self.llm.generate_structured( - prompt, - schema=CODE_GENERATION_SCHEMA, - system_prompt=CODE_GENERATION_SYSTEM_PROMPT, - temperature=0.2, - ) - result = CodeGenerationResult.from_llm_payload(structured) - return result.model_dump() - except ValueError as exc: - raise TaskExecutionError(str(exc), error_type="CODE_GENERATION_ERROR") from exc - except Exception as exc: - raise TaskExecutionError( - f"Code generation failed: {exc}", - error_type="CODE_GENERATION_ERROR", - ) from exc - - -async def run_code_generation(context: TaskContext) -> dict[str, Any]: - agent = CodeGenerationAgent(context.llm) - return await agent.generate(context) diff --git a/backend/agents/diagnosis_agent.py b/backend/agents/diagnosis_agent.py deleted file mode 100644 index 09b2a6375d0c4b00bd46a9fb36b1103d635f4326..0000000000000000000000000000000000000000 --- a/backend/agents/diagnosis_agent.py +++ /dev/null @@ -1,81 +0,0 @@ -from typing import Any - -from pydantic import BaseModel - -from backend.llm.base import LLMProvider -from backend.orchestration.failure_classifier import ClassifiedFailure -from backend.orchestration.task_context import TaskContext - - -class DiagnosisResult(BaseModel): - error_type: str - root_cause: str - recommended_fix: str - affected_files: list[str] = [] - - def to_dict(self) -> dict[str, Any]: - return self.model_dump() - - -DIAGNOSIS_SCHEMA: dict[str, Any] = { - "type": "object", - "properties": { - "error_type": {"type": "string"}, - "root_cause": {"type": "string"}, - "recommended_fix": {"type": "string"}, - "affected_files": {"type": "array", "items": {"type": "string"}}, - }, - "required": ["error_type", "root_cause", "recommended_fix", "affected_files"], -} - -DIAGNOSIS_SYSTEM_PROMPT = """You are a failure diagnosis agent. -Analyze the workflow failure context and return structured JSON with root cause and fix recommendation. -Return only valid JSON.""" - - -class DiagnosisAgent: - def __init__(self, llm: LLMProvider) -> None: - self.llm = llm - - async def diagnose( - self, - *, - user_prompt: str, - failure: ClassifiedFailure, - code_files: dict[str, dict[str, str]], - test_files: dict[str, dict[str, str]], - stdout: str = "", - stderr: str = "", - previous_attempts: list[dict[str, Any]] | None = None, - ) -> DiagnosisResult: - code_listing = "\n".join( - f"--- {path} ---\n{meta['content']}" for path, meta in code_files.items() - ) - test_listing = "\n".join( - f"--- {path} ---\n{meta['content']}" for path, meta in test_files.items() - ) - prompt = ( - f"User request:\n{user_prompt}\n\n" - f"Failure category: {failure.category.value}\n" - f"Failure message: {failure.message}\n" - f"stdout:\n{stdout}\n\nstderr:\n{stderr}\n\n" - f"Code files:\n{code_listing}\n\nTest files:\n{test_listing}\n\n" - f"Previous attempts: {len(previous_attempts or [])}\n" - "Diagnose the failure." - ) - structured = await self.llm.generate_structured( - prompt, - schema=DIAGNOSIS_SCHEMA, - system_prompt=DIAGNOSIS_SYSTEM_PROMPT, - temperature=0.0, - ) - return DiagnosisResult.model_validate(structured) - - -async def stub_diagnose(failure: ClassifiedFailure) -> DiagnosisResult: - return DiagnosisResult( - error_type=failure.category.value, - root_cause=f"Detected {failure.category.value}: {failure.message}", - recommended_fix="Apply automated repair to resolve the reported failure.", - affected_files=[], - ) diff --git a/backend/agents/planner.py b/backend/agents/planner.py deleted file mode 100644 index e79cad52fd7dd6a313033b488b2208b71f94752b..0000000000000000000000000000000000000000 --- a/backend/agents/planner.py +++ /dev/null @@ -1,91 +0,0 @@ -import json -from typing import Any - -from backend.llm.base import LLMProvider -from backend.orchestration.dag_schema import DAGPlan, DAGTaskDefinition, TaskType -from backend.orchestration.errors import PlannerError - -PLANNER_SYSTEM_PROMPT = """You are a software workflow planner. -Given a user coding request, produce a dependency-aware execution DAG as JSON. - -Rules: -- Return ONLY valid JSON matching the schema. -- Use unique task ids (short labels like A, B, C or descriptive ids). -- Task types must be one of: code_generation, test_generation, static_analysis, execute. -- Typical flow: code_generation first, then test_generation and static_analysis in parallel, then execute depending on both. -- depends_on must list task ids that must complete before this task runs. -- Do not create cycles. -""" - -PLANNER_OUTPUT_SCHEMA: dict[str, Any] = { - "type": "object", - "properties": { - "tasks": { - "type": "array", - "items": { - "type": "object", - "properties": { - "id": {"type": "string"}, - "type": { - "type": "string", - "enum": [t.value for t in TaskType], - }, - "depends_on": { - "type": "array", - "items": {"type": "string"}, - }, - }, - "required": ["id", "type", "depends_on"], - }, - } - }, - "required": ["tasks"], -} - - -class PlannerAgent: - """Generates a structured DAG plan from a user prompt.""" - - def __init__(self, llm: LLMProvider) -> None: - self.llm = llm - - async def create_plan(self, user_prompt: str, *, context: str | None = None) -> DAGPlan: - prompt = user_prompt - if context: - prompt = f"Relevant context:\n{context}\n\nUser request:\n{user_prompt}" - - try: - structured = await self.llm.generate_structured( - prompt, - schema=PLANNER_OUTPUT_SCHEMA, - system_prompt=PLANNER_SYSTEM_PROMPT, - temperature=0.0, - ) - return self.parse_plan(structured) - except PlannerError: - raise - except Exception as exc: - raise PlannerError(f"Planner LLM call failed: {exc}") from exc - - @staticmethod - def parse_plan(data: dict[str, Any]) -> DAGPlan: - try: - tasks = [ - DAGTaskDefinition( - id=task["id"], - type=TaskType(task["type"]), - depends_on=task.get("depends_on", []), - ) - for task in data["tasks"] - ] - return DAGPlan(tasks=tasks) - except (KeyError, TypeError, ValueError) as exc: - raise PlannerError(f"Invalid planner output structure: {exc}") from exc - - @staticmethod - def parse_plan_json(raw_json: str) -> DAGPlan: - try: - data = json.loads(raw_json) - except json.JSONDecodeError as exc: - raise PlannerError(f"Planner output is not valid JSON: {exc}") from exc - return PlannerAgent.parse_plan(data) diff --git a/backend/agents/repair_agent.py b/backend/agents/repair_agent.py deleted file mode 100644 index a9c1ee7eb5940e671e97951c38e20077eb4389fa..0000000000000000000000000000000000000000 --- a/backend/agents/repair_agent.py +++ /dev/null @@ -1,101 +0,0 @@ -from typing import Any - -from pydantic import BaseModel - -from backend.agents.code_agent import CodeGenerationResult -from backend.llm.base import LLMProvider -from backend.agents.diagnosis_agent import DiagnosisResult - - -class RepairResult(BaseModel): - files: dict[str, dict[str, str]] - repair_summary: str - status: str = "repaired" - - def to_dict(self) -> dict[str, Any]: - return { - "files": self.files, - "repair_summary": self.repair_summary, - "status": self.status, - } - - -REPAIR_SCHEMA: dict[str, Any] = { - "type": "object", - "properties": { - "files": { - "type": "array", - "items": { - "type": "object", - "properties": { - "path": {"type": "string"}, - "content": {"type": "string"}, - "language": {"type": "string"}, - }, - "required": ["path", "content"], - }, - }, - "repair_summary": {"type": "string"}, - }, - "required": ["files", "repair_summary"], -} - -REPAIR_SYSTEM_PROMPT = """You are a code repair agent. -Given the diagnosis and current project files, produce repaired files as JSON. -Return only valid JSON with files array and repair_summary.""" - - -class RepairAgent: - def __init__(self, llm: LLMProvider) -> None: - self.llm = llm - - async def repair( - self, - *, - user_prompt: str, - diagnosis: DiagnosisResult, - code_files: dict[str, dict[str, str]], - test_files: dict[str, dict[str, str]], - ) -> RepairResult: - all_files = {**code_files, **test_files} - listing = "\n".join( - f"--- {path} ---\n{meta['content']}" for path, meta in all_files.items() - ) - prompt = ( - f"User request:\n{user_prompt}\n\n" - f"Diagnosis:\n{diagnosis.model_dump_json()}\n\n" - f"Current files:\n{listing}\n\n" - "Produce repaired project files." - ) - structured = await self.llm.generate_structured( - prompt, - schema=REPAIR_SCHEMA, - system_prompt=REPAIR_SYSTEM_PROMPT, - temperature=0.1, - ) - generated = CodeGenerationResult.from_llm_payload(structured) - return RepairResult( - files=generated.files, - repair_summary=structured.get("repair_summary", "Repaired project files"), - ) - - -def stub_repair( - code_files: dict[str, dict[str, str]], - test_files: dict[str, dict[str, str]], - *, - fix_code: bool = True, -) -> RepairResult: - """Stub repair that fixes a common broken add() implementation for tests.""" - files = {**code_files, **test_files} - if fix_code and "main.py" in files: - content = files["main.py"]["content"] - if "return a - b" in content or "return a+b-1" in content: - files["main.py"] = { - **files["main.py"], - "content": "def add(a, b):\n return a + b\n", - } - return RepairResult( - files=files, - repair_summary="Stub repair applied", - ) diff --git a/backend/agents/task_handlers.py b/backend/agents/task_handlers.py deleted file mode 100644 index a83f2021abafea40ce325f4d38e138af5a6b1cd3..0000000000000000000000000000000000000000 --- a/backend/agents/task_handlers.py +++ /dev/null @@ -1,101 +0,0 @@ -import asyncio -from collections.abc import Awaitable, Callable -from typing import Any - -from backend.agents.code_agent import run_code_generation -from backend.agents.test_agent import run_test_generation -from backend.analysis.static_analyzer import StaticAnalyzer -from backend.execution.verification import run_verification -from backend.orchestration.dag_schema import TaskType -from backend.orchestration.errors import TaskExecutionError -from backend.orchestration.task_context import TaskContext - -TaskHandler = Callable[[TaskContext], Awaitable[dict[str, Any]]] - -STUB_CODE_FILES = { - "main.py": { - "content": "def add(a, b):\n return a - b\n", - "language": "python", - }, - "requirements.txt": {"content": "pytest\n", "language": "text"}, -} - -STUB_TEST_FILES = { - "tests/test_main.py": { - "content": "from main import add\n\ndef test_add():\n assert add(1, 2) == 3\n", - "language": "python", - } -} - - -async def _handle_code_generation(context: TaskContext) -> dict[str, Any]: - if context.use_stubs: - return await _stub_handler(context, TaskType.CODE_GENERATION) - return await run_code_generation(context) - - -async def _handle_test_generation(context: TaskContext) -> dict[str, Any]: - if context.use_stubs: - return await _stub_handler(context, TaskType.TEST_GENERATION) - return await run_test_generation(context) - - -async def _handle_static_analysis(context: TaskContext) -> dict[str, Any]: - if context.use_stubs: - return await _stub_handler(context, TaskType.STATIC_ANALYSIS) - files = context.generated_files - if not files: - raise TaskExecutionError( - "No generated code available for static analysis", - error_type="STATIC_ANALYSIS_ERROR", - ) - analyzer = StaticAnalyzer() - result = analyzer.analyze_files(files) - if result.status == "FAIL": - raise TaskExecutionError( - "Static analysis failed", - error_type="STATIC_ANALYSIS_ERROR", - ) - return result.to_dict() - - -async def _handle_execute(context: TaskContext) -> dict[str, Any]: - if context.use_stubs and not context.stub_execute_calls_docker: - return await _stub_handler(context, TaskType.EXECUTE) - return await run_verification(context) - - -async def _stub_handler(context: TaskContext, task_type: TaskType) -> dict[str, Any]: - await asyncio.sleep(context.stub_delay) - if context.fail_task == task_type.value: - raise TaskExecutionError( - f"{task_type.value} failed", - error_type=f"{task_type.value.upper()}_ERROR", - ) - if task_type == TaskType.CODE_GENERATION: - return {"status": "generated", "files": dict(STUB_CODE_FILES)} - if task_type == TaskType.TEST_GENERATION: - return {"status": "generated", "files": dict(STUB_TEST_FILES)} - if task_type == TaskType.STATIC_ANALYSIS: - return {"status": "PASS", "errors": [], "warnings": []} - if task_type == TaskType.EXECUTE: - return {"status": "PASS", "stdout": "", "stderr": "", "exit_code": 0} - return {"status": "planned", "files": {}, "task_type": task_type.value} - - -TASK_HANDLERS: dict[TaskType, TaskHandler] = { - TaskType.CODE_GENERATION: _handle_code_generation, - TaskType.TEST_GENERATION: _handle_test_generation, - TaskType.STATIC_ANALYSIS: _handle_static_analysis, - TaskType.EXECUTE: _handle_execute, -} - - -def get_task_handler(task_type: TaskType) -> TaskHandler: - handler = TASK_HANDLERS.get(task_type) - if handler is None: - raise TaskExecutionError( - f"No handler registered for task type '{task_type}'", - error_type="TASK_EXECUTION_ERROR", - ) - return handler diff --git a/backend/agents/test_agent.py b/backend/agents/test_agent.py deleted file mode 100644 index 836dec54b00e4b4a0f8cdc8081ba6c31189c53cf..0000000000000000000000000000000000000000 --- a/backend/agents/test_agent.py +++ /dev/null @@ -1,94 +0,0 @@ -from typing import Any - -from pydantic import BaseModel - -from backend.llm.base import LLMProvider -from backend.orchestration.errors import TaskExecutionError -from backend.orchestration.task_context import TaskContext - - -class TestGenerationResult(BaseModel): - files: dict[str, dict[str, str]] - status: str = "generated" - - @classmethod - def from_llm_payload(cls, payload: dict[str, Any]) -> "TestGenerationResult": - files: dict[str, dict[str, str]] = {} - for item in payload.get("files", []): - path = item["path"].lstrip("/").replace("\\", "/") - files[path] = { - "content": item["content"], - "language": item.get("language", "python"), - } - if not files: - raise ValueError("Test generation produced no files") - return cls(files=files) - - -TEST_GENERATION_SCHEMA: dict[str, Any] = { - "type": "object", - "properties": { - "files": { - "type": "array", - "items": { - "type": "object", - "properties": { - "path": {"type": "string"}, - "content": {"type": "string"}, - "language": {"type": "string"}, - }, - "required": ["path", "content"], - }, - } - }, - "required": ["files"], -} - -TEST_GENERATION_SYSTEM_PROMPT = """You are a test generation agent. -Given generated source code, produce pytest test files. -Return JSON with a files array (paths under tests/). -Do not modify implementation files. -Return only valid JSON.""" - - -class TestGenerationAgent: - def __init__(self, llm: LLMProvider) -> None: - self.llm = llm - - async def generate(self, context: TaskContext) -> dict[str, Any]: - code_files = context.generated_files - if not code_files: - raise TaskExecutionError( - "No generated code available for test generation", - error_type="TEST_GENERATION_ERROR", - ) - - file_listing = "\n\n".join( - f"--- {path} ---\n{meta['content']}" for path, meta in code_files.items() - ) - prompt = ( - f"User request:\n{context.user_prompt}\n\n" - f"Generated source files:\n{file_listing}\n\n" - "Generate pytest tests for the implementation." - ) - try: - structured = await self.llm.generate_structured( - prompt, - schema=TEST_GENERATION_SCHEMA, - system_prompt=TEST_GENERATION_SYSTEM_PROMPT, - temperature=0.2, - ) - result = TestGenerationResult.from_llm_payload(structured) - return result.model_dump() - except ValueError as exc: - raise TaskExecutionError(str(exc), error_type="TEST_GENERATION_ERROR") from exc - except Exception as exc: - raise TaskExecutionError( - f"Test generation failed: {exc}", - error_type="TEST_GENERATION_ERROR", - ) from exc - - -async def run_test_generation(context: TaskContext) -> dict[str, Any]: - agent = TestGenerationAgent(context.llm) - return await agent.generate(context) diff --git a/backend/analysis/__init__.py b/backend/analysis/__init__.py deleted file mode 100644 index 9feb32f0e09b4d13789e4065becc44bc46ffc87c..0000000000000000000000000000000000000000 --- a/backend/analysis/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Static analysis — implemented in Phase 2+.""" diff --git a/backend/analysis/static_analyzer.py b/backend/analysis/static_analyzer.py deleted file mode 100644 index 373c5f555f4e04e86d05c045c861e4874a2f6a95..0000000000000000000000000000000000000000 --- a/backend/analysis/static_analyzer.py +++ /dev/null @@ -1,110 +0,0 @@ -import ast -import re -from dataclasses import dataclass, field -from pathlib import Path -from typing import Any - -from backend.orchestration.errors import TaskExecutionError -from backend.orchestration.task_context import TaskContext - - -@dataclass -class AnalysisIssue: - file: str - line: int - message: str - - -@dataclass -class StaticAnalysisResult: - status: str - errors: list[dict[str, Any]] = field(default_factory=list) - warnings: list[dict[str, Any]] = field(default_factory=list) - - def to_dict(self) -> dict[str, Any]: - return { - "status": self.status, - "errors": self.errors, - "warnings": self.warnings, - } - - -class StaticAnalyzer: - """Analyze generated Python files without executing them.""" - - def analyze_files(self, files: dict[str, dict[str, str]]) -> StaticAnalysisResult: - errors: list[dict[str, Any]] = [] - warnings: list[dict[str, Any]] = [] - - python_files = { - path: meta["content"] - for path, meta in files.items() - if path.endswith(".py") - } - - if not python_files: - errors.append( - {"file": "", "line": 0, "message": "No Python source files found to analyze"} - ) - return StaticAnalysisResult(status="FAIL", errors=errors, warnings=warnings) - - for path, content in python_files.items(): - syntax_issues = self._check_syntax(path, content) - errors.extend(syntax_issues) - warnings.extend(self._check_imports(path, content)) - - status = "FAIL" if errors else "PASS" - return StaticAnalysisResult(status=status, errors=errors, warnings=warnings) - - def _check_syntax(self, path: str, content: str) -> list[dict[str, Any]]: - try: - ast.parse(content, filename=path) - return [] - except SyntaxError as exc: - return [ - { - "file": path, - "line": exc.lineno or 0, - "message": f"Syntax error: {exc.msg}", - } - ] - - def _check_imports(self, path: str, content: str) -> list[dict[str, Any]]: - warnings: list[dict[str, Any]] = [] - for line_no, line in enumerate(content.splitlines(), start=1): - stripped = line.strip() - if stripped.startswith("import ") or stripped.startswith("from "): - if re.search(r"\b(os\.system|subprocess\.|eval\(|exec\()", stripped): - warnings.append( - { - "file": path, - "line": line_no, - "message": "Suspicious import or call pattern detected", - } - ) - return warnings - - -async def run_static_analysis(context: TaskContext) -> dict[str, Any]: - files = context.all_project_files - if not files: - raise TaskExecutionError( - "No generated files available for static analysis", - error_type="STATIC_ANALYSIS_ERROR", - ) - analyzer = StaticAnalyzer() - result = analyzer.analyze_files(files) - if result.status == "FAIL": - raise TaskExecutionError( - "Static analysis failed", - error_type="STATIC_ANALYSIS_ERROR", - ) - return result.to_dict() - - -def analyze_workspace(workspace: Path) -> StaticAnalysisResult: - files: dict[str, dict[str, str]] = {} - for path in workspace.rglob("*.py"): - relative = path.relative_to(workspace).as_posix() - files[relative] = {"content": path.read_text(encoding="utf-8"), "language": "python"} - return StaticAnalyzer().analyze_files(files) diff --git a/backend/api/__init__.py b/backend/api/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/backend/api/auth.py b/backend/api/auth.py deleted file mode 100644 index 6641b5cbcbffacf586df06f1c5eb3bac3b841729..0000000000000000000000000000000000000000 --- a/backend/api/auth.py +++ /dev/null @@ -1,70 +0,0 @@ -from fastapi import APIRouter, Depends, HTTPException, status - -from backend.api.schemas import LoginRequest, SignupRequest, TokenResponse, UserResponse -from backend.core.dependencies import get_current_user, get_repository -from backend.core.security import create_access_token, hash_password, verify_password -from backend.db.models import User -from backend.memory.workflow_repository import WorkflowRepository - -router = APIRouter(prefix="/auth", tags=["auth"]) - - -@router.post("/signup", response_model=UserResponse, status_code=status.HTTP_201_CREATED) -async def signup( - payload: SignupRequest, - repo: WorkflowRepository = Depends(get_repository), -) -> User: - if await repo.get_user_by_email(payload.email): - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered") - if await repo.get_user_by_username(payload.username): - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, detail="Username already taken" - ) - - user = await repo.create_user( - email=payload.email, - username=payload.username, - password_hash=hash_password(payload.password), - ) - return user - - -@router.post("/login", response_model=TokenResponse) -async def login( - payload: LoginRequest, - repo: WorkflowRepository = Depends(get_repository), -) -> TokenResponse: - user = await repo.get_user_by_email(payload.email) - if user is None or not verify_password(payload.password, user.password_hash): - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid email or password", - ) - - token = create_access_token(str(user.id)) - return TokenResponse(access_token=token) - - -@router.post("/guest", response_model=TokenResponse) -async def guest_login( - repo: WorkflowRepository = Depends(get_repository), -) -> TokenResponse: - import uuid - guest_id = str(uuid.uuid4()) - guest_email = f"guest-{guest_id}@autoforge.local" - guest_username = f"Guest_{guest_id[:8]}" - - # Auto-create the guest user with a random password - user = await repo.create_user( - email=guest_email, - username=guest_username, - password_hash=hash_password(guest_id), - ) - - token = create_access_token(str(user.id)) - return TokenResponse(access_token=token) - - -@router.get("/me", response_model=UserResponse) -async def me(current_user: User = Depends(get_current_user)) -> User: - return current_user diff --git a/backend/api/providers.py b/backend/api/providers.py deleted file mode 100644 index 870fbe43b3ca1756aa46e11935db9770bdfbbeae..0000000000000000000000000000000000000000 --- a/backend/api/providers.py +++ /dev/null @@ -1,36 +0,0 @@ -from fastapi import APIRouter - -from backend.api.schemas import ProviderInfo, ProviderListResponse -from backend.llm.providers import list_supported_providers - -router = APIRouter(prefix="/settings", tags=["settings"]) - -OPENAI_MODELS = [ - "gpt-4o", - "gpt-4o-mini", - "gpt-4.1", - "gpt-4.1-mini", -] - -GEMINI_MODELS = [ - "gemini-3.1-pro", - "gemini-3.1-flash", -] - -PROVIDER_MODELS: dict[str, list[str]] = { - "openai": OPENAI_MODELS, - "gemini": GEMINI_MODELS, -} - - -@router.get("/providers", response_model=ProviderListResponse) -async def list_providers() -> ProviderListResponse: - providers = [ - ProviderInfo( - id=provider_id, - name=provider_id.replace("_", " ").title(), - models=PROVIDER_MODELS.get(provider_id, []), - ) - for provider_id in list_supported_providers() - ] - return ProviderListResponse(providers=providers) diff --git a/backend/api/schemas.py b/backend/api/schemas.py deleted file mode 100644 index 6e2e3e1154ee0ab5dc12353ac65615987e96fd03..0000000000000000000000000000000000000000 --- a/backend/api/schemas.py +++ /dev/null @@ -1,146 +0,0 @@ -from datetime import datetime - -from pydantic import BaseModel, ConfigDict, EmailStr, Field - - -class SignupRequest(BaseModel): - email: EmailStr - username: str = Field(min_length=3, max_length=100) - password: str = Field(min_length=8, max_length=128) - - -class LoginRequest(BaseModel): - email: EmailStr - password: str - - -class TokenResponse(BaseModel): - access_token: str - token_type: str = "bearer" - - -class UserResponse(BaseModel): - model_config = ConfigDict(from_attributes=True) - - id: int - email: str - username: str - created_at: datetime - - -class WorkflowCreateRequest(BaseModel): - user_prompt: str = Field(min_length=1) - llm_provider: str = Field(min_length=1, max_length=64) - llm_model: str = Field(min_length=1, max_length=128) - api_key: str | None = Field(default=None, min_length=1) - - -class WorkflowResponse(BaseModel): - model_config = ConfigDict(from_attributes=True) - - workflow_id: str - user_prompt: str - status: str - llm_provider: str | None - llm_model: str | None - planner_output: dict | None - created_at: datetime - completed_at: datetime | None - - -class WorkflowListResponse(BaseModel): - workflows: list[WorkflowResponse] - total: int - - -class ApiCredentialCreateRequest(BaseModel): - provider: str = Field(min_length=1, max_length=64) - api_key: str = Field(min_length=1) - - -class ApiCredentialResponse(BaseModel): - model_config = ConfigDict(from_attributes=True) - - id: int - provider: str - api_key_hint: str - created_at: datetime - updated_at: datetime - - -class HealthResponse(BaseModel): - status: str - database: str - - -class TaskResponse(BaseModel): - model_config = ConfigDict(from_attributes=True) - - task_key: str - task_type: str - status: str - depends_on: list[str] | None - output: dict | None - started_at: datetime | None - completed_at: datetime | None - - -class TaskListResponse(BaseModel): - tasks: list[TaskResponse] - total: int - - -class EventResponse(BaseModel): - model_config = ConfigDict(from_attributes=True, populate_by_name=True) - - id: int - event_type: str - message: str - task_id: int | None - metadata: dict | None = Field(default=None, validation_alias="event_metadata") - timestamp: datetime - - -class AttemptResponse(BaseModel): - model_config = ConfigDict(from_attributes=True) - - id: int - task_id: int | None - attempt_number: int - generated_code: dict | None - generated_tests: dict | None - static_analysis_result: dict | None - stdout: str | None - stderr: str | None - exit_code: int | None - result: str | None - error: dict | None - created_at: datetime - - -class AttemptListResponse(BaseModel): - attempts: list[AttemptResponse] - total: int - - -class WorkflowStartRequest(BaseModel): - api_key: str | None = Field(default=None, min_length=1) - - -class ProviderInfo(BaseModel): - id: str - name: str - models: list[str] - - -class ProviderListResponse(BaseModel): - providers: list[ProviderInfo] - - -class EventListResponse(BaseModel): - events: list[EventResponse] - total: int - - -class WorkflowStartResponse(WorkflowResponse): - pass diff --git a/backend/api/settings.py b/backend/api/settings.py deleted file mode 100644 index 350587198c939c8c8fa2ee4ac6c56814a41236e7..0000000000000000000000000000000000000000 --- a/backend/api/settings.py +++ /dev/null @@ -1,73 +0,0 @@ -from fastapi import APIRouter, Depends, HTTPException, status -from sqlalchemy import select - -from backend.api.schemas import ApiCredentialCreateRequest, ApiCredentialResponse -from backend.core.dependencies import get_current_user, get_repository -from backend.core.encryption import encrypt_value, redact_secret -from backend.db.models import ApiCredential, User -from backend.db.session import get_db -from sqlalchemy.ext.asyncio import AsyncSession - -router = APIRouter(prefix="/settings/api-credentials", tags=["settings"]) - - -@router.post("", response_model=ApiCredentialResponse, status_code=status.HTTP_201_CREATED) -async def create_api_credential( - payload: ApiCredentialCreateRequest, - current_user: User = Depends(get_current_user), - db: AsyncSession = Depends(get_db), -) -> ApiCredentialResponse: - credential = ApiCredential( - user_id=current_user.id, - provider=payload.provider.lower().strip(), - encrypted_api_key=encrypt_value(payload.api_key), - ) - db.add(credential) - await db.flush() - await db.refresh(credential) - return ApiCredentialResponse( - id=credential.id, - provider=credential.provider, - api_key_hint=redact_secret(payload.api_key), - created_at=credential.created_at, - updated_at=credential.updated_at, - ) - - -@router.get("", response_model=list[ApiCredentialResponse]) -async def list_api_credentials( - current_user: User = Depends(get_current_user), - db: AsyncSession = Depends(get_db), -) -> list[ApiCredentialResponse]: - result = await db.execute( - select(ApiCredential).where(ApiCredential.user_id == current_user.id) - ) - credentials = result.scalars().all() - return [ - ApiCredentialResponse( - id=cred.id, - provider=cred.provider, - api_key_hint="********", - created_at=cred.created_at, - updated_at=cred.updated_at, - ) - for cred in credentials - ] - - -@router.delete("/{credential_id}", status_code=status.HTTP_204_NO_CONTENT) -async def delete_api_credential( - credential_id: int, - current_user: User = Depends(get_current_user), - db: AsyncSession = Depends(get_db), -) -> None: - result = await db.execute( - select(ApiCredential).where( - ApiCredential.id == credential_id, - ApiCredential.user_id == current_user.id, - ) - ) - credential = result.scalar_one_or_none() - if credential is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Credential not found") - await db.delete(credential) diff --git a/backend/api/workflows.py b/backend/api/workflows.py deleted file mode 100644 index 8d02e4c4ec7433879ae6a116f9c4838fdc659bf8..0000000000000000000000000000000000000000 --- a/backend/api/workflows.py +++ /dev/null @@ -1,370 +0,0 @@ -import asyncio -import json -import logging - -from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status -from fastapi.responses import StreamingResponse -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from backend.api.schemas import ( - AttemptListResponse, - AttemptResponse, - EventListResponse, - EventResponse, - TaskListResponse, - TaskResponse, - WorkflowCreateRequest, - WorkflowListResponse, - WorkflowResponse, - WorkflowStartRequest, - WorkflowStartResponse, -) -from backend.core.dependencies import get_current_user, get_repository -from backend.core.encryption import decrypt_value -from backend.core.security import decode_access_token -from backend.db.models import ApiCredential, Attempt, Event, User, Workflow, WorkflowStatus -from backend.db.session import AsyncSessionLocal, get_db -from backend.events.publisher import EventPublisher -from backend.llm.providers import get_llm_provider -from backend.memory.workflow_repository import WorkflowRepository -from backend.orchestration.workflow_runner import WorkflowRunner - -logger = logging.getLogger(__name__) -router = APIRouter(prefix="/workflows", tags=["workflows"]) - -TERMINAL_WORKFLOW_STATUSES = { - WorkflowStatus.SUCCESS, - WorkflowStatus.FAILED, - WorkflowStatus.CANCELLED, -} - - -def _to_workflow_response(workflow) -> WorkflowResponse: - return WorkflowResponse( - workflow_id=workflow.workflow_id, - user_prompt=workflow.user_prompt, - status=workflow.status, - llm_provider=workflow.llm_provider, - llm_model=workflow.llm_model, - planner_output=workflow.planner_output, - created_at=workflow.created_at, - completed_at=workflow.completed_at, - ) - - -async def _resolve_api_key( - db: AsyncSession, - user_id: int, - provider: str, - request_api_key: str | None, -) -> str: - if request_api_key: - return request_api_key - - result = await db.execute( - select(ApiCredential).where( - ApiCredential.user_id == user_id, - ApiCredential.provider == provider.lower().strip(), - ) - ) - credential = result.scalar_one_or_none() - if credential is None: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="API key required. Provide api_key or store credentials in settings.", - ) - return decrypt_value(credential.encrypted_api_key) - - -async def _execute_workflow_background( - workflow_id: str, - user_id: int, - api_key: str | None, -) -> None: - async with AsyncSessionLocal() as db: - try: - repo = WorkflowRepository(db) - workflow = await repo.get_workflow_by_public_id(workflow_id, user_id) - if workflow is None: - return - llm = get_llm_provider( - workflow.llm_provider, - api_key=api_key, - model=workflow.llm_model, - ) - import shutil - - has_docker = shutil.which("docker") is not None - if not has_docker: - raise RuntimeError("DOCKER_UNAVAILABLE: Sandbox unavailable — Docker is required to execute generated code.") - - runner = WorkflowRunner(db) - await runner.run(workflow, llm) - await db.commit() - except Exception as exc: - logger.exception("Background workflow execution failed for %s", workflow_id) - await db.rollback() - try: - repo = WorkflowRepository(db) - workflow = await repo.get_workflow_by_public_id(workflow_id, user_id) - if workflow and workflow.status == WorkflowStatus.RUNNING: - runner = WorkflowRunner(db) - await runner._fail_workflow( - workflow, - message=str(exc), - error_type="WORKFLOW_EXECUTION_ERROR", - ) - await db.commit() - except Exception: - logger.exception("Failed to mark workflow %s as failed", workflow_id) - await db.rollback() - - -@router.post("", response_model=WorkflowResponse, status_code=status.HTTP_201_CREATED) -async def create_workflow( - payload: WorkflowCreateRequest, - current_user: User = Depends(get_current_user), - repo: WorkflowRepository = Depends(get_repository), - db: AsyncSession = Depends(get_db), -) -> WorkflowResponse: - workflow = await repo.create_workflow( - user_id=current_user.id, - user_prompt=payload.user_prompt, - llm_provider=payload.llm_provider, - llm_model=payload.llm_model, - ) - if payload.api_key: - existing = await db.execute( - select(ApiCredential).where( - ApiCredential.user_id == current_user.id, - ApiCredential.provider == payload.llm_provider.lower().strip(), - ) - ) - if existing.scalar_one_or_none() is None: - from backend.core.encryption import encrypt_value - - db.add( - ApiCredential( - user_id=current_user.id, - provider=payload.llm_provider.lower().strip(), - encrypted_api_key=encrypt_value(payload.api_key), - ) - ) - return _to_workflow_response(workflow) - - -@router.post("/{workflow_id}/start", response_model=WorkflowStartResponse, status_code=status.HTTP_202_ACCEPTED) -async def start_workflow( - workflow_id: str, - payload: WorkflowStartRequest, - background_tasks: BackgroundTasks, - current_user: User = Depends(get_current_user), - repo: WorkflowRepository = Depends(get_repository), - db: AsyncSession = Depends(get_db), -) -> WorkflowStartResponse: - workflow = await repo.get_workflow_by_public_id(workflow_id, current_user.id) - if workflow is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Workflow not found") - if workflow.status not in {WorkflowStatus.CREATED, WorkflowStatus.FAILED}: - raise HTTPException( - status_code=status.HTTP_409_CONFLICT, - detail=f"Workflow cannot be started from status {workflow.status}", - ) - if not workflow.llm_provider or not workflow.llm_model: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="Workflow is missing LLM provider configuration", - ) - - resolved_key = await _resolve_api_key( - db, - current_user.id, - workflow.llm_provider, - payload.api_key, - ) - - workflow.status = WorkflowStatus.RUNNING - await db.flush() - - background_tasks.add_task( - _execute_workflow_background, - workflow_id, - current_user.id, - resolved_key, - ) - return _to_workflow_response(workflow) - - -@router.get("", response_model=WorkflowListResponse) -async def list_workflows( - current_user: User = Depends(get_current_user), - repo: WorkflowRepository = Depends(get_repository), -) -> WorkflowListResponse: - workflows = await repo.list_workflows_for_user(current_user.id) - items = [_to_workflow_response(wf) for wf in workflows] - return WorkflowListResponse(workflows=items, total=len(items)) - - -@router.get("/{workflow_id}", response_model=WorkflowResponse) -async def get_workflow( - workflow_id: str, - current_user: User = Depends(get_current_user), - repo: WorkflowRepository = Depends(get_repository), -) -> WorkflowResponse: - workflow = await repo.get_workflow_by_public_id(workflow_id, current_user.id) - if workflow is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Workflow not found") - return _to_workflow_response(workflow) - - -@router.get("/{workflow_id}/tasks", response_model=TaskListResponse) -async def list_workflow_tasks( - workflow_id: str, - current_user: User = Depends(get_current_user), - repo: WorkflowRepository = Depends(get_repository), - db: AsyncSession = Depends(get_db), -) -> TaskListResponse: - from backend.memory.task_repository import TaskRepository - - workflow = await repo.get_workflow_by_public_id(workflow_id, current_user.id) - if workflow is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Workflow not found") - - tasks = await TaskRepository(db).get_tasks_for_workflow(workflow.id) - items = [ - TaskResponse( - task_key=task.task_key, - task_type=task.task_type, - status=task.status, - depends_on=task.depends_on, - output=task.output, - started_at=task.started_at, - completed_at=task.completed_at, - ) - for task in tasks - ] - return TaskListResponse(tasks=items, total=len(items)) - - -@router.get("/{workflow_id}/events", response_model=EventListResponse) -async def list_workflow_events( - workflow_id: str, - after_id: int = 0, - current_user: User = Depends(get_current_user), - repo: WorkflowRepository = Depends(get_repository), - db: AsyncSession = Depends(get_db), -) -> EventListResponse: - workflow = await repo.get_workflow_by_public_id(workflow_id, current_user.id) - if workflow is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Workflow not found") - - query = ( - select(Event) - .where(Event.workflow_id == workflow.id, Event.id > after_id) - .order_by(Event.timestamp.asc()) - ) - events = (await db.execute(query)).scalars().all() - items = [ - EventResponse( - id=event.id, - event_type=event.event_type, - message=event.message, - task_id=event.task_id, - event_metadata=event.event_metadata, - timestamp=event.timestamp, - ) - for event in events - ] - return EventListResponse(events=items, total=len(items)) - - -@router.get("/{workflow_id}/events/stream") -async def stream_workflow_events( - workflow_id: str, - token: str = Query(...), - after_id: int = Query(0), - db: AsyncSession = Depends(get_db), -): - user_id_str = decode_access_token(token) - if user_id_str is None: - raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token") - user_id = int(user_id_str) - repo = WorkflowRepository(db) - workflow = await repo.get_workflow_by_public_id(workflow_id, user_id) - if workflow is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Workflow not found") - - async def event_generator(): - cursor = after_id - idle_ticks = 0 - while True: - result = await db.execute( - select(Event) - .where(Event.workflow_id == workflow.id, Event.id > cursor) - .order_by(Event.id.asc()) - ) - new_events = result.scalars().all() - if new_events: - idle_ticks = 0 - for event in new_events: - cursor = event.id - payload = { - "id": event.id, - "event_type": event.event_type, - "message": event.message, - "task_id": event.task_id, - "metadata": event.event_metadata, - "timestamp": event.timestamp.isoformat(), - } - yield f"data: {json.dumps(payload)}\n\n" - else: - idle_ticks += 1 - - wf_result = await db.execute( - select(Workflow.status).where(Workflow.id == workflow.id) - ) - current_status = wf_result.scalar_one_or_none() - if current_status in TERMINAL_WORKFLOW_STATUSES and idle_ticks >= 3: - yield f"data: {json.dumps({'event_type': 'STREAM_END', 'message': 'Workflow finished'})}\n\n" - break - await asyncio.sleep(1) - - return StreamingResponse(event_generator(), media_type="text/event-stream") - - -@router.get("/{workflow_id}/attempts", response_model=AttemptListResponse) -async def list_workflow_attempts( - workflow_id: str, - current_user: User = Depends(get_current_user), - repo: WorkflowRepository = Depends(get_repository), - db: AsyncSession = Depends(get_db), -) -> AttemptListResponse: - workflow = await repo.get_workflow_by_public_id(workflow_id, current_user.id) - if workflow is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Workflow not found") - - result = await db.execute( - select(Attempt) - .where(Attempt.workflow_id == workflow.id) - .order_by(Attempt.attempt_number.asc(), Attempt.id.asc()) - ) - attempts = result.scalars().all() - items = [ - AttemptResponse( - id=attempt.id, - task_id=attempt.task_id, - attempt_number=attempt.attempt_number, - generated_code=attempt.generated_code, - generated_tests=attempt.generated_tests, - static_analysis_result=attempt.static_analysis_result, - stdout=attempt.stdout, - stderr=attempt.stderr, - exit_code=attempt.exit_code, - result=attempt.result, - error=attempt.error, - created_at=attempt.created_at, - ) - for attempt in attempts - ] - return AttemptListResponse(attempts=items, total=len(items)) diff --git a/backend/core/__init__.py b/backend/core/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/backend/core/config.py b/backend/core/config.py deleted file mode 100644 index 947f056ef754dc77d555145ea992a5ad12d5f90e..0000000000000000000000000000000000000000 --- a/backend/core/config.py +++ /dev/null @@ -1,59 +0,0 @@ -import os -import tempfile -from functools import lru_cache - -from pydantic import field_validator -from pydantic_settings import BaseSettings, SettingsConfigDict - - -def _default_workspace_root() -> str: - base = os.path.join(tempfile.gettempdir(), "autoforge", "workspaces") - os.makedirs(base, exist_ok=True) - return base - - -class Settings(BaseSettings): - model_config = SettingsConfigDict( - env_file=".env", - env_file_encoding="utf-8", - case_sensitive=False, - extra="ignore", - ) - - database_url: str = "sqlite+aiosqlite:////tmp/autoforge.db" - - @field_validator("database_url", mode="before") - @classmethod - def normalize_database_url(cls, v: str | None) -> str: - if not v or not isinstance(v, str) or not v.strip(): - return "sqlite+aiosqlite:////tmp/autoforge.db" - v = v.strip() - # If someone accidentally pasted an HTTP/HTTPS URL (e.g. dashboard URL) - if v.startswith("http://") or v.startswith("https://"): - return "sqlite+aiosqlite:////tmp/autoforge.db" - if v.startswith("postgres://"): - return v.replace("postgres://", "postgresql+asyncpg://", 1) - if v.startswith("postgresql://") and not v.startswith("postgresql+asyncpg://"): - return v.replace("postgresql://", "postgresql+asyncpg://", 1) - if not (v.startswith("sqlite") or v.startswith("postgresql")): - return "sqlite+aiosqlite:////tmp/autoforge.db" - return v - secret_key: str = "change-me-in-production-use-a-long-random-string" - encryption_key: str = "change-me-32-byte-base64-fernet-key-here=" - algorithm: str = "HS256" - access_token_expire_minutes: int = 60 * 24 - log_level: str = "INFO" - docker_host: str = "unix:///var/run/docker.sock" - max_retries: int = 4 - workspace_root: str = _default_workspace_root() - default_timeout: int = 300 - cors_origins: str = "http://localhost:3000,http://localhost:5173" - docker_image: str = "autoforge-sandbox:3.12" - docker_memory_limit: str = "512m" - docker_cpu_limit: float = 1.0 - docker_network_disabled: bool = True - - -@lru_cache -def get_settings() -> Settings: - return Settings() diff --git a/backend/core/dependencies.py b/backend/core/dependencies.py deleted file mode 100644 index 0859e036b4ec78ac5b1447deff9e3aaa81140245..0000000000000000000000000000000000000000 --- a/backend/core/dependencies.py +++ /dev/null @@ -1,50 +0,0 @@ -from fastapi import Depends, HTTPException, status -from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer -from sqlalchemy.ext.asyncio import AsyncSession - -from backend.core.security import decode_access_token -from backend.db.models import User -from backend.db.session import get_db -from backend.memory.workflow_repository import WorkflowRepository - -security = HTTPBearer(auto_error=False) - - -async def get_repository(db: AsyncSession = Depends(get_db)) -> WorkflowRepository: - return WorkflowRepository(db) - - -async def get_current_user( - credentials: HTTPAuthorizationCredentials | None = Depends(security), - repo: WorkflowRepository = Depends(get_repository), -) -> User: - if credentials is None or not credentials.credentials: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Not authenticated", - headers={"WWW-Authenticate": "Bearer"}, - ) - - user_id_str = decode_access_token(credentials.credentials) - if user_id_str is None: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid or expired token", - headers={"WWW-Authenticate": "Bearer"}, - ) - - try: - user_id = int(user_id_str) - except ValueError as exc: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Invalid token subject", - ) from exc - - user = await repo.get_user_by_id(user_id) - if user is None: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="User not found", - ) - return user diff --git a/backend/core/encryption.py b/backend/core/encryption.py deleted file mode 100644 index dac73ccdbcbc5774c1c2b73d86d0703fd469b0f6..0000000000000000000000000000000000000000 --- a/backend/core/encryption.py +++ /dev/null @@ -1,30 +0,0 @@ -import base64 -import hashlib - -from cryptography.fernet import Fernet, InvalidToken - -from backend.core.config import get_settings - - -def _fernet() -> Fernet: - settings = get_settings() - key_material = hashlib.sha256(settings.encryption_key.encode()).digest() - fernet_key = base64.urlsafe_b64encode(key_material) - return Fernet(fernet_key) - - -def encrypt_value(value: str) -> str: - return _fernet().encrypt(value.encode()).decode() - - -def decrypt_value(encrypted_value: str) -> str: - try: - return _fernet().decrypt(encrypted_value.encode()).decode() - except InvalidToken as exc: - raise ValueError("Failed to decrypt credential") from exc - - -def redact_secret(value: str, visible_chars: int = 4) -> str: - if len(value) <= visible_chars: - return "*" * len(value) - return f"{'*' * (len(value) - visible_chars)}{value[-visible_chars:]}" diff --git a/backend/core/security.py b/backend/core/security.py deleted file mode 100644 index e7fa0f2f20a7c64979d8e231b9363dfeca2035a8..0000000000000000000000000000000000000000 --- a/backend/core/security.py +++ /dev/null @@ -1,36 +0,0 @@ -from datetime import UTC, datetime, timedelta -from typing import Any - -from jose import JWTError, jwt -from passlib.context import CryptContext - -from backend.core.config import get_settings - -pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") - - -def hash_password(password: str) -> str: - return pwd_context.hash(password) - - -def verify_password(plain_password: str, hashed_password: str) -> bool: - return pwd_context.verify(plain_password, hashed_password) - - -def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str: - settings = get_settings() - expire = datetime.now(UTC) + ( - expires_delta or timedelta(minutes=settings.access_token_expire_minutes) - ) - payload: dict[str, Any] = {"sub": subject, "exp": expire} - return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm) - - -def decode_access_token(token: str) -> str | None: - settings = get_settings() - try: - payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) - subject = payload.get("sub") - return str(subject) if subject else None - except JWTError: - return None diff --git a/backend/db/__init__.py b/backend/db/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/backend/db/migrations/env.py b/backend/db/migrations/env.py deleted file mode 100644 index a2bf488770a7d4c0dbd92a8e48a60f61eafa52b1..0000000000000000000000000000000000000000 --- a/backend/db/migrations/env.py +++ /dev/null @@ -1,52 +0,0 @@ -from logging.config import fileConfig - -from alembic import context -from sqlalchemy import engine_from_config, pool - -from backend.core.config import get_settings -from backend.db.models import Base - -config = context.config -if config.config_file_name is not None: - fileConfig(config.config_file_name) - -target_metadata = Base.metadata - - -def get_url() -> str: - return get_settings().database_url.replace("+asyncpg", "") - - -def run_migrations_offline() -> None: - url = get_url() - context.configure( - url=url, - target_metadata=target_metadata, - literal_binds=True, - dialect_opts={"paramstyle": "named"}, - ) - - with context.begin_transaction(): - context.run_migrations() - - -def run_migrations_online() -> None: - configuration = config.get_section(config.config_ini_section) or {} - configuration["sqlalchemy.url"] = get_url() - connectable = engine_from_config( - configuration, - prefix="sqlalchemy.", - poolclass=pool.NullPool, - ) - - with connectable.connect() as connection: - context.configure(connection=connection, target_metadata=target_metadata) - - with context.begin_transaction(): - context.run_migrations() - - -if context.is_offline_mode(): - run_migrations_offline() -else: - run_migrations_online() diff --git a/backend/db/migrations/versions/001_initial_schema.py b/backend/db/migrations/versions/001_initial_schema.py deleted file mode 100644 index 4284e2f897f227d4290ac372e0c49fad02cc3623..0000000000000000000000000000000000000000 --- a/backend/db/migrations/versions/001_initial_schema.py +++ /dev/null @@ -1,138 +0,0 @@ -"""Initial schema - -Revision ID: 001_initial -Revises: -Create Date: 2026-08-29 -""" - -from typing import Sequence, Union - -import sqlalchemy as sa -from alembic import op -from sqlalchemy.dialects import postgresql - -revision: str = "001_initial" -down_revision: Union[str, None] = None -branch_labels: Union[str, Sequence[str], None] = None -depends_on: Union[str, Sequence[str], None] = None - - -def upgrade() -> None: - op.create_table( - "users", - sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), - sa.Column("email", sa.String(length=255), nullable=False), - sa.Column("username", sa.String(length=100), nullable=False), - sa.Column("password_hash", sa.String(length=255), nullable=False), - sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), - sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), - sa.PrimaryKeyConstraint("id"), - ) - op.create_index(op.f("ix_users_email"), "users", ["email"], unique=True) - op.create_index(op.f("ix_users_username"), "users", ["username"], unique=True) - - op.create_table( - "workflows", - sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), - sa.Column("workflow_id", sa.String(length=64), nullable=False), - sa.Column("user_id", sa.Integer(), nullable=False), - sa.Column("user_prompt", sa.Text(), nullable=False), - sa.Column("status", sa.String(length=32), nullable=False), - sa.Column("planner_output", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - sa.Column("llm_provider", sa.String(length=64), nullable=True), - sa.Column("llm_model", sa.String(length=128), nullable=True), - sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), - sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), - sa.ForeignKeyConstraint(["user_id"], ["users.id"]), - sa.PrimaryKeyConstraint("id"), - ) - op.create_index(op.f("ix_workflows_user_id"), "workflows", ["user_id"], unique=False) - op.create_index(op.f("ix_workflows_workflow_id"), "workflows", ["workflow_id"], unique=True) - - op.create_table( - "api_credentials", - sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), - sa.Column("user_id", sa.Integer(), nullable=False), - sa.Column("provider", sa.String(length=64), nullable=False), - sa.Column("encrypted_api_key", sa.Text(), nullable=False), - sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), - sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), - sa.ForeignKeyConstraint(["user_id"], ["users.id"]), - sa.PrimaryKeyConstraint("id"), - ) - op.create_index(op.f("ix_api_credentials_user_id"), "api_credentials", ["user_id"], unique=False) - - op.create_table( - "tasks", - sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), - sa.Column("workflow_id", sa.Integer(), nullable=False), - sa.Column("task_key", sa.String(length=64), nullable=False), - sa.Column("task_type", sa.String(length=64), nullable=False), - sa.Column("status", sa.String(length=32), nullable=False), - sa.Column("depends_on", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - sa.Column("output", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - sa.Column("started_at", sa.DateTime(timezone=True), nullable=True), - sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), - sa.ForeignKeyConstraint(["workflow_id"], ["workflows.id"]), - sa.PrimaryKeyConstraint("id"), - ) - op.create_index(op.f("ix_tasks_workflow_id"), "tasks", ["workflow_id"], unique=False) - - op.create_table( - "attempts", - sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), - sa.Column("workflow_id", sa.Integer(), nullable=False), - sa.Column("task_id", sa.Integer(), nullable=True), - sa.Column("attempt_number", sa.Integer(), nullable=False), - sa.Column("generated_code", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - sa.Column("generated_tests", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - sa.Column("static_analysis_result", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - sa.Column("stdout", sa.Text(), nullable=True), - sa.Column("stderr", sa.Text(), nullable=True), - sa.Column("exit_code", sa.Integer(), nullable=True), - sa.Column("result", sa.String(length=32), nullable=True), - sa.Column("error", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), - sa.ForeignKeyConstraint(["task_id"], ["tasks.id"]), - sa.ForeignKeyConstraint(["workflow_id"], ["workflows.id"]), - sa.PrimaryKeyConstraint("id"), - ) - op.create_index(op.f("ix_attempts_task_id"), "attempts", ["task_id"], unique=False) - op.create_index(op.f("ix_attempts_workflow_id"), "attempts", ["workflow_id"], unique=False) - - op.create_table( - "events", - sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), - sa.Column("workflow_id", sa.Integer(), nullable=False), - sa.Column("task_id", sa.Integer(), nullable=True), - sa.Column("event_type", sa.String(length=64), nullable=False), - sa.Column("message", sa.Text(), nullable=False), - sa.Column("metadata", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - sa.Column("timestamp", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), - sa.ForeignKeyConstraint(["task_id"], ["tasks.id"]), - sa.ForeignKeyConstraint(["workflow_id"], ["workflows.id"]), - sa.PrimaryKeyConstraint("id"), - ) - op.create_index(op.f("ix_events_event_type"), "events", ["event_type"], unique=False) - op.create_index(op.f("ix_events_timestamp"), "events", ["timestamp"], unique=False) - op.create_index(op.f("ix_events_workflow_id"), "events", ["workflow_id"], unique=False) - - -def downgrade() -> None: - op.drop_index(op.f("ix_events_workflow_id"), table_name="events") - op.drop_index(op.f("ix_events_timestamp"), table_name="events") - op.drop_index(op.f("ix_events_event_type"), table_name="events") - op.drop_table("events") - op.drop_index(op.f("ix_attempts_workflow_id"), table_name="attempts") - op.drop_index(op.f("ix_attempts_task_id"), table_name="attempts") - op.drop_table("attempts") - op.drop_index(op.f("ix_tasks_workflow_id"), table_name="tasks") - op.drop_table("tasks") - op.drop_index(op.f("ix_api_credentials_user_id"), table_name="api_credentials") - op.drop_table("api_credentials") - op.drop_index(op.f("ix_workflows_workflow_id"), table_name="workflows") - op.drop_index(op.f("ix_workflows_user_id"), table_name="workflows") - op.drop_table("workflows") - op.drop_index(op.f("ix_users_username"), table_name="users") - op.drop_index(op.f("ix_users_email"), table_name="users") - op.drop_table("users") diff --git a/backend/db/migrations/versions/__init__.py b/backend/db/migrations/versions/__init__.py deleted file mode 100644 index 3fff3f9bac4362f2485a47d60ac8de2f1ce6afaa..0000000000000000000000000000000000000000 --- a/backend/db/migrations/versions/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Alembic migrations for AutoForge.""" diff --git a/backend/db/models/__init__.py b/backend/db/models/__init__.py deleted file mode 100644 index cc467e5aa6f576f07139cad8dd3e6a8e4c7280d9..0000000000000000000000000000000000000000 --- a/backend/db/models/__init__.py +++ /dev/null @@ -1,150 +0,0 @@ -from datetime import datetime -from enum import StrEnum - -from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, func, JSON -from sqlalchemy.dialects.postgresql import JSONB -from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship - -JSON_TYPE = JSON().with_variant(JSONB, "postgresql") - - -class Base(DeclarativeBase): - pass - - -class WorkflowStatus(StrEnum): - CREATED = "CREATED" - RUNNING = "RUNNING" - SUCCESS = "SUCCESS" - FAILED = "FAILED" - CANCELLED = "CANCELLED" - CANCELLING = "CANCELLING" - - -class TaskStatus(StrEnum): - CREATED = "CREATED" - READY = "READY" - RUNNING = "RUNNING" - SUCCESS = "SUCCESS" - FAILED = "FAILED" - RETRYING = "RETRYING" - CANCELLED = "CANCELLED" - - -class User(Base): - __tablename__ = "users" - - id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True) - username: Mapped[str] = mapped_column(String(100), unique=True, nullable=False, index=True) - password_hash: Mapped[str] = mapped_column(String(255), nullable=False) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), server_default=func.now(), nullable=False - ) - updated_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False - ) - - workflows: Mapped[list["Workflow"]] = relationship(back_populates="user") - api_credentials: Mapped[list["ApiCredential"]] = relationship(back_populates="user") - - -class Workflow(Base): - __tablename__ = "workflows" - - id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - workflow_id: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True) - user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, index=True) - user_prompt: Mapped[str] = mapped_column(Text, nullable=False) - status: Mapped[str] = mapped_column(String(32), nullable=False, default=WorkflowStatus.CREATED) - planner_output: Mapped[dict | None] = mapped_column(JSON_TYPE, nullable=True) - llm_provider: Mapped[str | None] = mapped_column(String(64), nullable=True) - llm_model: Mapped[str | None] = mapped_column(String(128), nullable=True) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), server_default=func.now(), nullable=False - ) - completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) - - user: Mapped["User"] = relationship(back_populates="workflows") - tasks: Mapped[list["Task"]] = relationship(back_populates="workflow", cascade="all, delete-orphan") - attempts: Mapped[list["Attempt"]] = relationship( - back_populates="workflow", cascade="all, delete-orphan" - ) - events: Mapped[list["Event"]] = relationship( - back_populates="workflow", cascade="all, delete-orphan" - ) - - -class Task(Base): - __tablename__ = "tasks" - - id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - workflow_id: Mapped[int] = mapped_column(ForeignKey("workflows.id"), nullable=False, index=True) - task_key: Mapped[str] = mapped_column(String(64), nullable=False) - task_type: Mapped[str] = mapped_column(String(64), nullable=False) - status: Mapped[str] = mapped_column(String(32), nullable=False, default=TaskStatus.CREATED) - depends_on: Mapped[list | None] = mapped_column(JSON_TYPE, nullable=True) - output: Mapped[dict | None] = mapped_column(JSON_TYPE, nullable=True) - started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) - completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) - - workflow: Mapped["Workflow"] = relationship(back_populates="tasks") - attempts: Mapped[list["Attempt"]] = relationship(back_populates="task") - events: Mapped[list["Event"]] = relationship(back_populates="task") - - -class Attempt(Base): - __tablename__ = "attempts" - - id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - workflow_id: Mapped[int] = mapped_column(ForeignKey("workflows.id"), nullable=False, index=True) - task_id: Mapped[int | None] = mapped_column(ForeignKey("tasks.id"), nullable=True, index=True) - attempt_number: Mapped[int] = mapped_column(Integer, nullable=False, default=1) - generated_code: Mapped[dict | None] = mapped_column(JSON_TYPE, nullable=True) - generated_tests: Mapped[dict | None] = mapped_column(JSON_TYPE, nullable=True) - static_analysis_result: Mapped[dict | None] = mapped_column(JSON_TYPE, nullable=True) - stdout: Mapped[str | None] = mapped_column(Text, nullable=True) - stderr: Mapped[str | None] = mapped_column(Text, nullable=True) - exit_code: Mapped[int | None] = mapped_column(Integer, nullable=True) - result: Mapped[str | None] = mapped_column(String(32), nullable=True) - error: Mapped[dict | None] = mapped_column(JSON_TYPE, nullable=True) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), server_default=func.now(), nullable=False - ) - - workflow: Mapped["Workflow"] = relationship(back_populates="attempts") - task: Mapped["Task | None"] = relationship(back_populates="attempts") - - -class Event(Base): - __tablename__ = "events" - - id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - workflow_id: Mapped[int] = mapped_column(ForeignKey("workflows.id"), nullable=False, index=True) - task_id: Mapped[int | None] = mapped_column(ForeignKey("tasks.id"), nullable=True, index=True) - event_type: Mapped[str] = mapped_column(String(64), nullable=False, index=True) - message: Mapped[str] = mapped_column(Text, nullable=False) - event_metadata: Mapped[dict | None] = mapped_column("metadata", JSON_TYPE, nullable=True) - timestamp: Mapped[datetime] = mapped_column( - DateTime(timezone=True), server_default=func.now(), nullable=False, index=True - ) - - workflow: Mapped["Workflow"] = relationship(back_populates="events") - task: Mapped["Task | None"] = relationship(back_populates="events") - - -class ApiCredential(Base): - __tablename__ = "api_credentials" - - id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False, index=True) - provider: Mapped[str] = mapped_column(String(64), nullable=False) - encrypted_api_key: Mapped[str] = mapped_column(Text, nullable=False) - created_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), server_default=func.now(), nullable=False - ) - updated_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False - ) - - user: Mapped["User"] = relationship(back_populates="api_credentials") diff --git a/backend/db/session.py b/backend/db/session.py deleted file mode 100644 index bed9004cde11681c5b46b8ae80ec48a1130e6600..0000000000000000000000000000000000000000 --- a/backend/db/session.py +++ /dev/null @@ -1,50 +0,0 @@ -import logging -from collections.abc import AsyncGenerator - -from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine - -from backend.core.config import get_settings - -logger = logging.getLogger(__name__) -settings = get_settings() - - -def _create_engine_safely(url: str): - try: - if url.startswith("sqlite"): - return create_async_engine( - url, - echo=False, - connect_args={"check_same_thread": False}, - ) - return create_async_engine( - url, - echo=False, - pool_pre_ping=True, - ) - except Exception as exc: - logger.error("Failed to create engine for %s, falling back to SQLite: %s", url, exc) - return create_async_engine( - "sqlite+aiosqlite:////tmp/autoforge.db", - echo=False, - connect_args={"check_same_thread": False}, - ) - - -engine = _create_engine_safely(settings.database_url) - -AsyncSessionLocal = async_sessionmaker( - engine, - class_=AsyncSession, - expire_on_commit=False, -) - - -async def get_db() -> AsyncGenerator[AsyncSession, None]: - async with AsyncSessionLocal() as session: - try: - yield session - await session.commit() - except Exception: - await session.rollback() - raise diff --git a/backend/events/__init__.py b/backend/events/__init__.py deleted file mode 100644 index 380b4a5502a9cdbbcc9ddbe3a0fbff31870c36c2..0000000000000000000000000000000000000000 --- a/backend/events/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Event publishing utilities.""" diff --git a/backend/events/publisher.py b/backend/events/publisher.py deleted file mode 100644 index 35c7465dd2d4528e767379ff64ee703dca713f74..0000000000000000000000000000000000000000 --- a/backend/events/publisher.py +++ /dev/null @@ -1,40 +0,0 @@ -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from backend.db.models import Event - - -class EventPublisher: - """Persists workflow events to PostgreSQL.""" - - def __init__(self, db: AsyncSession) -> None: - self.db = db - - async def publish( - self, - *, - workflow_db_id: int, - public_workflow_id: str, - event_type: str, - message: str, - task_id: int | None = None, - task_key: str | None = None, - metadata: dict | None = None, - ) -> Event: - event_metadata = {"workflow_id": public_workflow_id} - if task_key: - event_metadata["task_key"] = task_key - if metadata: - event_metadata.update(metadata) - - event = Event( - workflow_id=workflow_db_id, - task_id=task_id, - event_type=event_type, - message=message, - event_metadata=event_metadata, - ) - self.db.add(event) - await self.db.flush() - await self.db.refresh(event) - return event diff --git a/backend/execution/__init__.py b/backend/execution/__init__.py deleted file mode 100644 index 9f23aa3b734ca80ef24b9f140475825d6048df08..0000000000000000000000000000000000000000 --- a/backend/execution/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Docker sandbox execution — implemented in Phase 2+.""" diff --git a/backend/execution/docker_runner.py b/backend/execution/docker_runner.py deleted file mode 100644 index dea12efd9fad3c3488906e0518388326fd0312e4..0000000000000000000000000000000000000000 --- a/backend/execution/docker_runner.py +++ /dev/null @@ -1,265 +0,0 @@ -import asyncio -import os -import time -import uuid -from abc import ABC, abstractmethod -from pathlib import Path, PureWindowsPath - -from backend.core.config import get_settings -from backend.execution.sandbox import ExecutionResult - - -def _docker_volume_path(workspace: Path) -> str: - """Convert a host path to a Docker-compatible volume mount source.""" - resolved = workspace.resolve() - if os.name == "nt": - path = PureWindowsPath(resolved) - drive = path.drive.rstrip(":").lower() - rest = str(path.relative_to(path.anchor)).replace("\\", "/") - return f"/{drive}/{rest}" - return resolved.as_posix() - - -def _build_container_test_command(workspace: Path, *, network_disabled: bool) -> str: - requirements = workspace / "requirements.txt" - if ( - not network_disabled - and requirements.exists() - and requirements.read_text(encoding="utf-8").strip() - ): - return "pip install --no-cache-dir -r requirements.txt -q && python -m pytest -v" - return "python -m pytest -v" - - -class DockerRunnerBase(ABC): - @abstractmethod - async def run_tests( - self, - workspace: Path, - *, - timeout: int | None = None, - ) -> ExecutionResult: - """Run project tests inside an isolated container.""" - - @abstractmethod - async def cleanup(self, container_id: str | None = None) -> None: - """Remove container resources.""" - - -class DockerCliRunner(DockerRunnerBase): - """Execute generated projects via the Docker CLI — never on the host.""" - - def __init__( - self, - *, - docker_host: str | None = None, - image: str | None = None, - memory_limit: str | None = None, - cpu_limit: float | None = None, - network_disabled: bool | None = None, - ) -> None: - settings = get_settings() - self.docker_host = docker_host or settings.docker_host - self.image = image or settings.docker_image - self.memory_limit = memory_limit or settings.docker_memory_limit - self.cpu_limit = cpu_limit or settings.docker_cpu_limit - self.network_disabled = ( - network_disabled - if network_disabled is not None - else settings.docker_network_disabled - ) - self.default_timeout = settings.default_timeout - self._active_containers: set[str] = set() - - async def run_tests( - self, - workspace: Path, - *, - timeout: int | None = None, - ) -> ExecutionResult: - timeout = timeout or self.default_timeout - container_name = f"autoforge_{uuid.uuid4().hex[:12]}" - workspace_mount = _docker_volume_path(workspace) - test_command = _build_container_test_command(workspace, network_disabled=self.network_disabled) - - command = [ - "docker", - "run", - "--name", - container_name, - "--rm", - "--memory", - self.memory_limit, - "--cpus", - str(self.cpu_limit), - "-v", - f"{workspace_mount}:/workspace:ro", - "-w", - "/workspace", - ] - if self.network_disabled: - command.append("--network") - command.append("none") - - command.extend([self.image, "bash", "-lc", test_command]) - - start = time.perf_counter() - timed_out = False - stdout = "" - stderr = "" - exit_code = 1 - container_id = container_name - - try: - process = await asyncio.create_subprocess_exec( - *command, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) - self._active_containers.add(container_name) - try: - stdout_bytes, stderr_bytes = await asyncio.wait_for( - process.communicate(), - timeout=timeout, - ) - stdout = stdout_bytes.decode("utf-8", errors="replace") - stderr = stderr_bytes.decode("utf-8", errors="replace") - exit_code = process.returncode or 0 - except asyncio.TimeoutError: - timed_out = True - stderr = f"Execution timed out after {timeout} seconds" - exit_code = 124 - await self._force_remove_container(container_name) - finally: - self._active_containers.discard(container_name) - except FileNotFoundError as exc: - stderr = "Docker CLI not found. Generated code cannot run without Docker." - exit_code = 127 - return ExecutionResult( - stdout=stdout, - stderr=stderr, - exit_code=exit_code, - execution_time=time.perf_counter() - start, - timed_out=False, - container_id=container_id, - container_build_status="failed", - container_start_status="failed", - ) - - elapsed = time.perf_counter() - start - return ExecutionResult( - stdout=stdout, - stderr=stderr, - exit_code=exit_code, - execution_time=elapsed, - timed_out=timed_out, - container_id=container_id, - container_build_status="success", - container_start_status="started" if not timed_out else "timeout", - ) - - async def cleanup(self, container_id: str | None = None) -> None: - if container_id: - await self._force_remove_container(container_id) - return - for cid in list(self._active_containers): - await self._force_remove_container(cid) - - async def _force_remove_container(self, container_name: str) -> None: - process = await asyncio.create_subprocess_exec( - "docker", - "rm", - "-f", - container_name, - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.DEVNULL, - ) - await process.wait() - self._active_containers.discard(container_name) - - -class FakeDockerRunner(DockerRunnerBase): - """Test double that simulates container execution without Docker.""" - - def __init__( - self, - *, - exit_code: int = 0, - stdout: str = "===== 2 passed in 0.01s =====", - stderr: str = "", - timed_out: bool = False, - delay: float = 0, - should_fail_start: bool = False, - ) -> None: - self.exit_code = exit_code - self.stdout = stdout - self.stderr = stderr - self.timed_out = timed_out - self.delay = delay - self.should_fail_start = should_fail_start - self.fail_remaining = 0 - self.containers_started: list[str] = [] - self.cleaned_up: list[str] = [] - - def with_failures(self, count: int) -> "FakeDockerRunner": - self.fail_remaining = count - return self - - async def run_tests( - self, - workspace: Path, - *, - timeout: int | None = None, - ) -> ExecutionResult: - if self.should_fail_start: - return ExecutionResult( - stdout="", - stderr="Docker build failed", - exit_code=1, - execution_time=0, - container_build_status="failed", - container_start_status="failed", - ) - - if self.delay: - await asyncio.sleep(self.delay) - - container_id = f"fake_{uuid.uuid4().hex[:8]}" - self.containers_started.append(container_id) - - if self.fail_remaining > 0: - self.fail_remaining -= 1 - return ExecutionResult( - stdout="===== 1 failed in 0.01s =====", - stderr="AssertionError: test failed", - exit_code=1, - execution_time=self.delay, - container_id=container_id, - container_build_status="success", - container_start_status="started", - ) - - if self.timed_out: - return ExecutionResult( - stdout="", - stderr="Execution timed out", - exit_code=124, - execution_time=float(timeout or 1), - timed_out=True, - container_id=container_id, - container_build_status="success", - container_start_status="timeout", - ) - - return ExecutionResult( - stdout=self.stdout, - stderr=self.stderr, - exit_code=self.exit_code, - execution_time=self.delay, - container_id=container_id, - container_build_status="success", - container_start_status="started", - ) - - async def cleanup(self, container_id: str | None = None) -> None: - self.cleaned_up.append(container_id or "all") diff --git a/backend/execution/result_parser.py b/backend/execution/result_parser.py deleted file mode 100644 index 36abf4f40278f8533c355d8e1f8e80f446371d86..0000000000000000000000000000000000000000 --- a/backend/execution/result_parser.py +++ /dev/null @@ -1,40 +0,0 @@ -import re -from dataclasses import dataclass - - -@dataclass -class TestSummary: - passed: int - failed: int - total: int - - @property - def success(self) -> bool: - return self.failed == 0 and self.total > 0 - - -def parse_pytest_output(stdout: str, stderr: str) -> TestSummary: - combined = f"{stdout}\n{stderr}" - match = re.search(r"=+\s+(\d+) passed(?:,\s+(\d+) failed)?", combined) - if match: - passed = int(match.group(1)) - failed = int(match.group(2) or 0) - return TestSummary(passed=passed, failed=failed, total=passed + failed) - - failed_match = re.search(r"=+\s+(\d+) failed", combined) - if failed_match: - failed = int(failed_match.group(1)) - return TestSummary(passed=0, failed=failed, total=failed) - - return TestSummary(passed=0, failed=0, total=0) - - -def parse_verification_result(stdout: str, stderr: str, exit_code: int) -> dict: - summary = parse_pytest_output(stdout, stderr) - return { - "tests_passed": summary.passed, - "tests_failed": summary.failed, - "tests_total": summary.total, - "verification_passed": exit_code == 0 and summary.success, - "exit_code": exit_code, - } diff --git a/backend/execution/sandbox.py b/backend/execution/sandbox.py deleted file mode 100644 index 796c91eec0eb7497ab19ca466179d0a0d3c6a5fe..0000000000000000000000000000000000000000 --- a/backend/execution/sandbox.py +++ /dev/null @@ -1,38 +0,0 @@ -from dataclasses import dataclass -from pathlib import Path - - -@dataclass -class ExecutionResult: - stdout: str - stderr: str - exit_code: int - execution_time: float - timed_out: bool = False - container_id: str | None = None - container_build_status: str = "skipped" - container_start_status: str = "unknown" - - @property - def success(self) -> bool: - return self.exit_code == 0 and not self.timed_out - - -def write_project_files(workspace: Path, files: dict[str, dict[str, str]]) -> None: - workspace.mkdir(parents=True, exist_ok=True) - for rel_path, meta in files.items(): - target = workspace / rel_path - target.parent.mkdir(parents=True, exist_ok=True) - target.write_text(meta["content"], encoding="utf-8") - - -def cleanup_workspace(workspace: Path) -> None: - if not workspace.exists(): - return - for path in sorted(workspace.rglob("*"), reverse=True): - if path.is_file(): - path.unlink() - elif path.is_dir(): - path.rmdir() - if workspace.exists(): - workspace.rmdir() diff --git a/backend/execution/verification.py b/backend/execution/verification.py deleted file mode 100644 index 7e54981e508cd638a5488ddcda70afe14017c501..0000000000000000000000000000000000000000 --- a/backend/execution/verification.py +++ /dev/null @@ -1,79 +0,0 @@ -from pathlib import Path - -from backend.execution.docker_runner import DockerRunnerBase -from backend.execution.result_parser import parse_verification_result -from backend.execution.sandbox import ExecutionResult, write_project_files -from backend.orchestration.errors import TaskExecutionError -from backend.orchestration.task_context import TaskContext - - -class VerificationPipeline: - """Code → Static Analysis gate → Docker → Test Execution → Verification.""" - - def __init__(self, docker_runner: DockerRunnerBase) -> None: - self.docker_runner = docker_runner - - async def run(self, context: TaskContext) -> dict: - files = context.all_project_files - if not files: - raise TaskExecutionError( - "No project files available for verification", - error_type="RUNTIME_ERROR", - ) - - static_result = context.task_outputs_by_type.get("static_analysis") - if static_result and static_result.get("status") != "PASS": - raise TaskExecutionError( - "Static analysis did not pass; skipping execution", - error_type="STATIC_ANALYSIS_ERROR", - ) - - workspace = context.workspace_path / "verify" - write_project_files(workspace, files) - - execution: ExecutionResult | None = None - try: - execution = await self.docker_runner.run_tests(workspace) - verification = parse_verification_result( - execution.stdout, - execution.stderr, - execution.exit_code, - ) - - if execution.timed_out: - raise TaskExecutionError( - "Docker execution timed out", - error_type="TIMEOUT", - ) - - if not verification["verification_passed"]: - raise TaskExecutionError( - f"Verification failed: {verification['tests_failed']} test(s) failed", - error_type="TEST_FAILURE", - ) - - return { - "status": "PASS", - "stdout": execution.stdout, - "stderr": execution.stderr, - "exit_code": execution.exit_code, - "execution_time": execution.execution_time, - "timed_out": execution.timed_out, - "container_build_status": execution.container_build_status, - "container_start_status": execution.container_start_status, - "verification": verification, - } - finally: - await self.docker_runner.cleanup( - execution.container_id if execution else None - ) - - -async def run_verification(context: TaskContext) -> dict: - docker_runner = context.docker_runner - if docker_runner is None: - from backend.execution.docker_runner import DockerCliRunner - - docker_runner = DockerCliRunner() - pipeline = VerificationPipeline(docker_runner) - return await pipeline.run(context) diff --git a/backend/llm/__init__.py b/backend/llm/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/backend/llm/base.py b/backend/llm/base.py deleted file mode 100644 index 82cc054a14ac4e5e5150d65b39dfa1ab808ab93d..0000000000000000000000000000000000000000 --- a/backend/llm/base.py +++ /dev/null @@ -1,48 +0,0 @@ -import json -from abc import ABC, abstractmethod -from typing import Any - -from backend.llm.types import LLMCostEstimate, LLMResponse - - -class LLMProvider(ABC): - """Abstract LLM provider interface used by agents and orchestration.""" - - provider_name: str - - @abstractmethod - async def generate( - self, - prompt: str, - *, - system_prompt: str | None = None, - temperature: float = 0.2, - max_tokens: int | None = None, - ) -> LLMResponse: - """Generate a text completion from the provider.""" - - @abstractmethod - async def generate_structured( - self, - prompt: str, - *, - schema: dict[str, Any], - system_prompt: str | None = None, - temperature: float = 0.0, - ) -> dict[str, Any]: - """Generate structured JSON output validated against a schema description.""" - - @abstractmethod - def estimate_cost(self, input_tokens: int, output_tokens: int) -> LLMCostEstimate: - """Estimate API cost for a token usage pair.""" - - def _parse_json_content(self, content: str) -> dict[str, Any]: - text = content.strip() - if text.startswith("```"): - lines = text.splitlines() - if lines and lines[0].startswith("```"): - lines = lines[1:] - if lines and lines[-1].strip() == "```": - lines = lines[:-1] - text = "\n".join(lines).strip() - return json.loads(text) diff --git a/backend/llm/providers/__init__.py b/backend/llm/providers/__init__.py deleted file mode 100644 index d2d5f171929325346db6fda45102c78025a1914d..0000000000000000000000000000000000000000 --- a/backend/llm/providers/__init__.py +++ /dev/null @@ -1,20 +0,0 @@ -from backend.llm.base import LLMProvider -from backend.llm.providers.openai_provider import OpenAIProvider -from backend.llm.providers.gemini_provider import GeminiProvider - -SUPPORTED_PROVIDERS: dict[str, type[LLMProvider]] = { - "openai": OpenAIProvider, - "gemini": GeminiProvider, -} - - -def get_llm_provider(provider: str, api_key: str, model: str) -> LLMProvider: - provider_key = provider.lower().strip() - if provider_key not in SUPPORTED_PROVIDERS: - supported = ", ".join(sorted(SUPPORTED_PROVIDERS)) - raise ValueError(f"Unsupported LLM provider '{provider}'. Supported: {supported}") - return SUPPORTED_PROVIDERS[provider_key](api_key=api_key, model=model) - - -def list_supported_providers() -> list[str]: - return sorted(SUPPORTED_PROVIDERS.keys()) diff --git a/backend/llm/providers/gemini_provider.py b/backend/llm/providers/gemini_provider.py deleted file mode 100644 index 9e75f8c835c3bdfa776cdef963ba61a38e7f6a49..0000000000000000000000000000000000000000 --- a/backend/llm/providers/gemini_provider.py +++ /dev/null @@ -1,98 +0,0 @@ -import httpx - -from backend.llm.base import LLMProvider -from backend.llm.types import LLMCostEstimate, LLMResponse - -_GEMINI_PRICING: dict[str, dict[str, float]] = { - "gemini-3.1-pro": {"input": 1.25, "output": 5.00}, - "gemini-3.1-flash": {"input": 0.075, "output": 0.30}, -} - - -class GeminiProvider(LLMProvider): - provider_name = "gemini" - - def __init__(self, api_key: str, model: str) -> None: - self.api_key = api_key - self.model = model - self.base_url = "https://generativelanguage.googleapis.com/v1beta" - - async def generate( - self, - prompt: str, - *, - system_prompt: str | None = None, - temperature: float = 0.2, - max_tokens: int | None = None, - ) -> LLMResponse: - payload: dict = { - "contents": [{"role": "user", "parts": [{"text": prompt}]}], - "generationConfig": { - "temperature": temperature, - } - } - - if system_prompt: - payload["systemInstruction"] = {"parts": [{"text": system_prompt}]} - - if max_tokens is not None: - payload["generationConfig"]["maxOutputTokens"] = max_tokens - - async with httpx.AsyncClient(timeout=120.0) as client: - response = await client.post( - f"{self.base_url}/models/{self.model}:generateContent", - params={"key": self.api_key}, - headers={"Content-Type": "application/json"}, - json=payload, - ) - response.raise_for_status() - data = response.json() - - try: - choice = data["candidates"][0]["content"]["parts"][0]["text"] - except (KeyError, IndexError): - choice = "" - - usage = data.get("usageMetadata", {}) - return LLMResponse( - content=choice, - input_tokens=int(usage.get("promptTokenCount", 0)), - output_tokens=int(usage.get("candidatesTokenCount", 0)), - model=self.model, - provider=self.provider_name, - raw=data, - ) - - async def generate_structured( - self, - prompt: str, - *, - schema: dict, - system_prompt: str | None = None, - temperature: float = 0.0, - ) -> dict: - schema_instruction = ( - "Respond with valid JSON only that matches this schema description:\n" - f"{schema}" - ) - combined_system = ( - f"{system_prompt}\n\n{schema_instruction}" if system_prompt else schema_instruction - ) - response = await self.generate( - prompt, - system_prompt=combined_system, - temperature=temperature, - ) - return self._parse_json_content(response.content) - - def estimate_cost(self, input_tokens: int, output_tokens: int) -> LLMCostEstimate: - pricing = _GEMINI_PRICING.get(self.model, {"input": 1.0, "output": 3.0}) - input_cost = (input_tokens / 1_000_000) * pricing["input"] - output_cost = (output_tokens / 1_000_000) * pricing["output"] - return LLMCostEstimate( - input_tokens=input_tokens, - output_tokens=output_tokens, - estimated_cost_usd=round(input_cost + output_cost, 6), - provider=self.provider_name, - model=self.model, - ) diff --git a/backend/llm/providers/mock_provider.py b/backend/llm/providers/mock_provider.py deleted file mode 100644 index f43ddfa964c820b5a25e1fe02e9ff4690e81c131..0000000000000000000000000000000000000000 --- a/backend/llm/providers/mock_provider.py +++ /dev/null @@ -1,56 +0,0 @@ -from typing import Any - -from backend.llm.base import LLMProvider -from backend.llm.types import LLMCostEstimate, LLMResponse - - -class MockLLMProvider(LLMProvider): - """Test double for LLM provider interactions.""" - - provider_name = "mock" - - def __init__( - self, - *, - structured_response: dict[str, Any] | None = None, - text_response: str = "mock response", - model: str = "mock-model", - ) -> None: - self.structured_response = structured_response or {"tasks": []} - self.text_response = text_response - self.model = model - - async def generate( - self, - prompt: str, - *, - system_prompt: str | None = None, - temperature: float = 0.2, - max_tokens: int | None = None, - ) -> LLMResponse: - return LLMResponse( - content=self.text_response, - input_tokens=10, - output_tokens=20, - model=self.model, - provider=self.provider_name, - ) - - async def generate_structured( - self, - prompt: str, - *, - schema: dict[str, Any], - system_prompt: str | None = None, - temperature: float = 0.0, - ) -> dict[str, Any]: - return self.structured_response - - def estimate_cost(self, input_tokens: int, output_tokens: int) -> LLMCostEstimate: - return LLMCostEstimate( - input_tokens=input_tokens, - output_tokens=output_tokens, - estimated_cost_usd=0.0, - provider=self.provider_name, - model=self.model, - ) diff --git a/backend/llm/providers/openai_provider.py b/backend/llm/providers/openai_provider.py deleted file mode 100644 index fb73955e181652753b382dd89bcef68095558861..0000000000000000000000000000000000000000 --- a/backend/llm/providers/openai_provider.py +++ /dev/null @@ -1,99 +0,0 @@ -import httpx - -from backend.llm.base import LLMProvider -from backend.llm.types import LLMCostEstimate, LLMResponse - -# Approximate OpenAI pricing per 1M tokens (USD) — used for estimates only. -_OPENAI_PRICING: dict[str, dict[str, float]] = { - "gpt-4o": {"input": 2.50, "output": 10.00}, - "gpt-4o-mini": {"input": 0.15, "output": 0.60}, - "gpt-4.1": {"input": 2.00, "output": 8.00}, - "gpt-4.1-mini": {"input": 0.40, "output": 1.60}, -} - - -class OpenAIProvider(LLMProvider): - provider_name = "openai" - - def __init__(self, api_key: str, model: str, base_url: str = "https://api.openai.com/v1") -> None: - self.api_key = api_key - self.model = model - self.base_url = base_url.rstrip("/") - - async def generate( - self, - prompt: str, - *, - system_prompt: str | None = None, - temperature: float = 0.2, - max_tokens: int | None = None, - ) -> LLMResponse: - messages: list[dict[str, str]] = [] - if system_prompt: - messages.append({"role": "system", "content": system_prompt}) - messages.append({"role": "user", "content": prompt}) - - payload: dict = { - "model": self.model, - "messages": messages, - "temperature": temperature, - } - if max_tokens is not None: - payload["max_tokens"] = max_tokens - - async with httpx.AsyncClient(timeout=120.0) as client: - response = await client.post( - f"{self.base_url}/chat/completions", - headers={ - "Authorization": f"Bearer {self.api_key}", - "Content-Type": "application/json", - }, - json=payload, - ) - response.raise_for_status() - data = response.json() - - choice = data["choices"][0]["message"]["content"] - usage = data.get("usage", {}) - return LLMResponse( - content=choice, - input_tokens=int(usage.get("prompt_tokens", 0)), - output_tokens=int(usage.get("completion_tokens", 0)), - model=self.model, - provider=self.provider_name, - raw=data, - ) - - async def generate_structured( - self, - prompt: str, - *, - schema: dict, - system_prompt: str | None = None, - temperature: float = 0.0, - ) -> dict: - schema_instruction = ( - "Respond with valid JSON only that matches this schema description:\n" - f"{schema}" - ) - combined_system = ( - f"{system_prompt}\n\n{schema_instruction}" if system_prompt else schema_instruction - ) - response = await self.generate( - prompt, - system_prompt=combined_system, - temperature=temperature, - ) - return self._parse_json_content(response.content) - - def estimate_cost(self, input_tokens: int, output_tokens: int) -> LLMCostEstimate: - pricing = _OPENAI_PRICING.get(self.model, {"input": 1.0, "output": 3.0}) - input_cost = (input_tokens / 1_000_000) * pricing["input"] - output_cost = (output_tokens / 1_000_000) * pricing["output"] - return LLMCostEstimate( - input_tokens=input_tokens, - output_tokens=output_tokens, - estimated_cost_usd=round(input_cost + output_cost, 6), - provider=self.provider_name, - model=self.model, - ) diff --git a/backend/llm/types.py b/backend/llm/types.py deleted file mode 100644 index e83ec1399e4273d7b148191ee77c8c24e3025bca..0000000000000000000000000000000000000000 --- a/backend/llm/types.py +++ /dev/null @@ -1,21 +0,0 @@ -from dataclasses import dataclass -from typing import Any - - -@dataclass -class LLMResponse: - content: str - input_tokens: int - output_tokens: int - model: str - provider: str - raw: dict[str, Any] | None = None - - -@dataclass -class LLMCostEstimate: - input_tokens: int - output_tokens: int - estimated_cost_usd: float - provider: str - model: str diff --git a/backend/main.py b/backend/main.py deleted file mode 100644 index 97e447327183252d1dc173cc0fd58afb740474ab..0000000000000000000000000000000000000000 --- a/backend/main.py +++ /dev/null @@ -1,98 +0,0 @@ -import logging -from contextlib import asynccontextmanager - -from fastapi import FastAPI -from fastapi.middleware.cors import CORSMiddleware -from sqlalchemy import text - -from backend.api.auth import router as auth_router -from backend.api.providers import router as providers_router -from backend.api.settings import router as settings_router -from backend.api.workflows import router as workflows_router -from backend.api.schemas import HealthResponse -from backend.core.config import get_settings -from backend.db.models import Base -from backend.db.session import engine - -logger = logging.getLogger(__name__) - - -@asynccontextmanager -async def lifespan(app: FastAPI): - settings = get_settings() - logging.basicConfig(level=settings.log_level) - logger.info("Starting AutoForge backend") - - # Automatically initialize database schema for Hugging Face Spaces - async with engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - - yield - await engine.dispose() - logger.info("AutoForge backend stopped") - - -def create_app() -> FastAPI: - settings = get_settings() - app = FastAPI( - title="AutoForge API", - description="Autonomous DAG-based code generation and verification platform", - version="0.1.0", - lifespan=lifespan, - ) - - origins = [origin.strip() for origin in settings.cors_origins.split(",") if origin.strip()] - app.add_middleware( - CORSMiddleware, - allow_origins=origins, - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) - - from fastapi import APIRouter - from fastapi.staticfiles import StaticFiles - from fastapi.responses import FileResponse - import os - - api_router = APIRouter(prefix="/api") - api_router.include_router(auth_router) - api_router.include_router(workflows_router) - api_router.include_router(settings_router) - api_router.include_router(providers_router) - - @api_router.get("/health", response_model=HealthResponse, tags=["health"]) - async def health_check() -> HealthResponse: - db_status = "disconnected" - try: - async with engine.connect() as conn: - await conn.execute(text("SELECT 1")) - db_status = "connected" - except Exception as exc: - logger.warning("Database health check failed: %s", exc) - - return HealthResponse( - status="ok" if db_status == "connected" else "degraded", - database=db_status, - ) - - app.include_router(api_router) - - # Serve static frontend if it exists - frontend_dir = os.path.join(os.path.dirname(__file__), "..", "frontend", "dist") - if os.path.isdir(frontend_dir): - app.mount("/assets", StaticFiles(directory=os.path.join(frontend_dir, "assets")), name="assets") - - @app.get("/{full_path:path}") - async def serve_frontend(full_path: str): - # If a specific file is requested in root (like favicon.ico), try serving it - requested_file = os.path.join(frontend_dir, full_path) - if os.path.isfile(requested_file): - return FileResponse(requested_file) - # Otherwise, fall back to index.html for SPA routing - return FileResponse(os.path.join(frontend_dir, "index.html")) - - return app - - -app = create_app() diff --git a/backend/memory/__init__.py b/backend/memory/__init__.py deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/backend/memory/attempt_repository.py b/backend/memory/attempt_repository.py deleted file mode 100644 index 1db1e12884ad184bc921622d3c3acf67c2ad5cb3..0000000000000000000000000000000000000000 --- a/backend/memory/attempt_repository.py +++ /dev/null @@ -1,63 +0,0 @@ -from sqlalchemy import func, select -from sqlalchemy.ext.asyncio import AsyncSession - -from backend.db.models import Attempt - - -class AttemptRepository: - def __init__(self, db: AsyncSession) -> None: - self.db = db - - async def get_next_attempt_number(self, workflow_id: int) -> int: - result = await self.db.execute( - select(func.max(Attempt.attempt_number)).where(Attempt.workflow_id == workflow_id) - ) - current_max = result.scalar() - return int(current_max or 0) + 1 - - async def list_attempts_for_workflow(self, workflow_id: int) -> list[Attempt]: - result = await self.db.execute( - select(Attempt) - .where(Attempt.workflow_id == workflow_id) - .order_by(Attempt.attempt_number.asc(), Attempt.id.asc()) - ) - return list(result.scalars().all()) - - async def count_attempts(self, workflow_id: int) -> int: - result = await self.db.execute( - select(func.count()).select_from(Attempt).where(Attempt.workflow_id == workflow_id) - ) - return int(result.scalar() or 0) - - async def create_attempt( - self, - *, - workflow_id: int, - task_id: int, - attempt_number: int = 1, - generated_code: dict | None = None, - generated_tests: dict | None = None, - static_analysis_result: dict | None = None, - stdout: str | None = None, - stderr: str | None = None, - exit_code: int | None = None, - result: str | None = None, - error: dict | None = None, - ) -> Attempt: - attempt = Attempt( - workflow_id=workflow_id, - task_id=task_id, - attempt_number=attempt_number, - generated_code=generated_code, - generated_tests=generated_tests, - static_analysis_result=static_analysis_result, - stdout=stdout, - stderr=stderr, - exit_code=exit_code, - result=result, - error=error, - ) - self.db.add(attempt) - await self.db.flush() - await self.db.refresh(attempt) - return attempt diff --git a/backend/memory/task_repository.py b/backend/memory/task_repository.py deleted file mode 100644 index 815ed32de458852b6ea6109a22940672a51e83ea..0000000000000000000000000000000000000000 --- a/backend/memory/task_repository.py +++ /dev/null @@ -1,99 +0,0 @@ -from datetime import UTC, datetime - -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession - -from backend.db.models import Task, TaskStatus, Workflow, WorkflowStatus -from backend.orchestration.dag_schema import DAGPlan -from backend.orchestration.state_machine import validate_transition - - -class TaskRepository: - def __init__(self, db: AsyncSession) -> None: - self.db = db - - async def create_tasks_from_plan( - self, workflow: Workflow, plan: DAGPlan - ) -> list[Task]: - tasks: list[Task] = [] - for task_def in plan.tasks: - task = Task( - workflow_id=workflow.id, - task_key=task_def.id, - task_type=task_def.type.value, - status=TaskStatus.CREATED, - depends_on=task_def.depends_on, - ) - self.db.add(task) - tasks.append(task) - await self.db.flush() - for task in tasks: - await self.db.refresh(task) - return tasks - - async def get_tasks_for_workflow(self, workflow_db_id: int) -> list[Task]: - result = await self.db.execute( - select(Task) - .where(Task.workflow_id == workflow_db_id) - .order_by(Task.id.asc()) - ) - return list(result.scalars().all()) - - async def get_task_by_key(self, workflow_db_id: int, task_key: str) -> Task | None: - result = await self.db.execute( - select(Task).where( - Task.workflow_id == workflow_db_id, - Task.task_key == task_key, - ) - ) - return result.scalar_one_or_none() - - async def get_task_by_type(self, workflow_db_id: int, task_type: str) -> Task | None: - result = await self.db.execute( - select(Task).where( - Task.workflow_id == workflow_db_id, - Task.task_type == task_type, - ) - ) - return result.scalar_one_or_none() - - async def transition_task( - self, - task: Task, - target_status: TaskStatus | str, - *, - output: dict | None = None, - set_started: bool = False, - set_completed: bool = False, - ) -> Task: - validate_transition(task.status, target_status) - task.status = str(target_status) - if output is not None: - task.output = output - if set_started and task.started_at is None: - task.started_at = datetime.now(UTC) - if set_completed: - task.completed_at = datetime.now(UTC) - await self.db.flush() - await self.db.refresh(task) - return task - - async def mark_workflow_status( - self, - workflow: Workflow, - status: WorkflowStatus | str, - *, - completed: bool = False, - ) -> Workflow: - workflow.status = str(status) - if completed: - workflow.completed_at = datetime.now(UTC) - await self.db.flush() - await self.db.refresh(workflow) - return workflow - - async def save_planner_output(self, workflow: Workflow, planner_output: dict) -> Workflow: - workflow.planner_output = planner_output - await self.db.flush() - await self.db.refresh(workflow) - return workflow diff --git a/backend/memory/workflow_repository.py b/backend/memory/workflow_repository.py deleted file mode 100644 index 7be5d77cb231bac0ee8974d6739b38902eb83815..0000000000000000000000000000000000000000 --- a/backend/memory/workflow_repository.py +++ /dev/null @@ -1,84 +0,0 @@ -import uuid -from datetime import datetime - -from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.orm import selectinload - -from backend.db.models import Event, User, Workflow, WorkflowStatus - - -def generate_workflow_id() -> str: - return f"wf_{uuid.uuid4().hex[:12]}" - - -class WorkflowRepository: - def __init__(self, db: AsyncSession) -> None: - self.db = db - - async def create_workflow( - self, - *, - user_id: int, - user_prompt: str, - llm_provider: str | None = None, - llm_model: str | None = None, - ) -> Workflow: - workflow = Workflow( - workflow_id=generate_workflow_id(), - user_id=user_id, - user_prompt=user_prompt, - status=WorkflowStatus.CREATED, - llm_provider=llm_provider, - llm_model=llm_model, - ) - self.db.add(workflow) - await self.db.flush() - - event = Event( - workflow_id=workflow.id, - event_type="WORKFLOW_STARTED", - message="Workflow created", - event_metadata={"workflow_id": workflow.workflow_id}, - ) - self.db.add(event) - await self.db.flush() - await self.db.refresh(workflow) - return workflow - - async def get_workflow_by_public_id( - self, workflow_id: str, user_id: int - ) -> Workflow | None: - result = await self.db.execute( - select(Workflow) - .where(Workflow.workflow_id == workflow_id, Workflow.user_id == user_id) - .options(selectinload(Workflow.events)) - ) - return result.scalar_one_or_none() - - async def list_workflows_for_user(self, user_id: int) -> list[Workflow]: - result = await self.db.execute( - select(Workflow) - .where(Workflow.user_id == user_id) - .order_by(Workflow.created_at.desc()) - ) - return list(result.scalars().all()) - - async def get_user_by_email(self, email: str) -> User | None: - result = await self.db.execute(select(User).where(User.email == email)) - return result.scalar_one_or_none() - - async def get_user_by_username(self, username: str) -> User | None: - result = await self.db.execute(select(User).where(User.username == username)) - return result.scalar_one_or_none() - - async def get_user_by_id(self, user_id: int) -> User | None: - result = await self.db.execute(select(User).where(User.id == user_id)) - return result.scalar_one_or_none() - - async def create_user(self, *, email: str, username: str, password_hash: str) -> User: - user = User(email=email, username=username, password_hash=password_hash) - self.db.add(user) - await self.db.flush() - await self.db.refresh(user) - return user diff --git a/backend/orchestration/__init__.py b/backend/orchestration/__init__.py deleted file mode 100644 index 9b8c44efacd077328452e1e7f28222f8e7bd42ef..0000000000000000000000000000000000000000 --- a/backend/orchestration/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""DAG engine — validation, scheduling, execution, and workflow orchestration.""" diff --git a/backend/orchestration/dag_engine.py b/backend/orchestration/dag_engine.py deleted file mode 100644 index 9271cdcc4885612ff92dec3501bd5556f461120a..0000000000000000000000000000000000000000 --- a/backend/orchestration/dag_engine.py +++ /dev/null @@ -1,174 +0,0 @@ -import asyncio -from dataclasses import dataclass, field -from datetime import UTC, datetime -from typing import Any - -from backend.agents.task_handlers import get_task_handler -from backend.orchestration.dag_schema import DAGPlan, DAGTaskDefinition, TaskType -from backend.orchestration.errors import TaskExecutionError -from backend.orchestration.scheduler import get_ready_task_ids -from backend.orchestration.task_context import TaskContext - - -@dataclass -class TaskRunRecord: - task_key: str - task_type: TaskType - status: str - output: dict[str, Any] | None = None - error: dict[str, Any] | None = None - started_at: datetime | None = None - completed_at: datetime | None = None - - -@dataclass -class DAGExecutionResult: - success: bool - task_results: dict[str, TaskRunRecord] = field(default_factory=dict) - error: str | None = None - task_outputs_by_type: dict[str, dict[str, Any]] = field(default_factory=dict) - failed_task_type: str | None = None - failure_error: dict[str, Any] | None = None - - -class DAGExecutor: - """Executes a validated DAG with parallel execution of independent tasks.""" - - def __init__( - self, - *, - base_context: dict[str, Any] | None = None, - on_task_started: Any | None = None, - on_task_completed: Any | None = None, - on_task_failed: Any | None = None, - ) -> None: - self.base_context = base_context or {} - self.on_task_started = on_task_started - self.on_task_completed = on_task_completed - self.on_task_failed = on_task_failed - self.task_outputs: dict[str, dict[str, Any]] = {} - self.task_outputs_by_type: dict[str, dict[str, Any]] = {} - - async def execute(self, plan: DAGPlan) -> DAGExecutionResult: - completed: set[str] = set() - results: dict[str, TaskRunRecord] = { - task.id: TaskRunRecord(task_key=task.id, task_type=task.type, status="CREATED") - for task in plan.tasks - } - - while len(completed) < len(plan.tasks): - ready_ids = get_ready_task_ids(plan, completed) - if not ready_ids: - return DAGExecutionResult( - success=False, - task_results=results, - error="Deadlock detected: no ready tasks but DAG is incomplete", - ) - - layer_outcome = await self._execute_layer(plan, ready_ids, results) - if not layer_outcome.success: - return layer_outcome - - completed.update(ready_ids) - - return DAGExecutionResult( - success=True, - task_results=results, - task_outputs_by_type=dict(self.task_outputs_by_type), - ) - - async def _execute_layer( - self, - plan: DAGPlan, - ready_ids: list[str], - results: dict[str, TaskRunRecord], - ) -> DAGExecutionResult: - outcomes = await asyncio.gather( - *[self._run_task(plan, task_id, results) for task_id in ready_ids], - return_exceptions=True, - ) - - for outcome in outcomes: - if isinstance(outcome, DAGExecutionResult): - return outcome - if isinstance(outcome, Exception): - return DAGExecutionResult( - success=False, - task_results=results, - error=str(outcome), - ) - - return DAGExecutionResult(success=True, task_results=results) - - def _build_context(self, task_def: DAGTaskDefinition) -> TaskContext: - return TaskContext( - workflow_id=self.base_context["workflow_id"], - user_prompt=self.base_context["user_prompt"], - llm=self.base_context["llm"], - workspace_path=self.base_context["workspace_path"], - task_def=task_def, - task_outputs=dict(self.task_outputs), - task_outputs_by_type=dict(self.task_outputs_by_type), - docker_runner=self.base_context.get("docker_runner"), - use_stubs=self.base_context.get("use_stubs", False), - fail_task=self.base_context.get("fail_task"), - stub_delay=self.base_context.get("stub_delay", 0), - stub_execute_calls_docker=self.base_context.get("stub_execute_calls_docker", False), - ) - - async def _run_task( - self, - plan: DAGPlan, - task_id: str, - results: dict[str, TaskRunRecord], - ) -> None | DAGExecutionResult: - task_def = plan.task_map()[task_id] - record = results[task_id] - record.status = "RUNNING" - record.started_at = datetime.now(UTC) - - if self.on_task_started: - await self.on_task_started(task_def) - - context = self._build_context(task_def) - - try: - handler = get_task_handler(task_def.type) - output = await handler(context) - record.output = output - record.status = "SUCCESS" - record.completed_at = datetime.now(UTC) - self.task_outputs[task_id] = output - self.task_outputs_by_type[task_def.type.value] = output - if self.on_task_completed: - await self.on_task_completed(task_def, output) - except TaskExecutionError as exc: - record.status = "FAILED" - record.completed_at = datetime.now(UTC) - record.error = {"error_type": exc.error_type, "message": exc.message} - if self.on_task_failed: - await self.on_task_failed(task_def, record.error) - return DAGExecutionResult( - success=False, - task_results=results, - error=exc.message, - task_outputs_by_type=dict(self.task_outputs_by_type), - failed_task_type=task_def.type.value, - failure_error=record.error, - ) - except Exception as exc: - record.status = "FAILED" - record.completed_at = datetime.now(UTC) - record.error = {"error_type": "TASK_EXECUTION_ERROR", "message": str(exc)} - if self.on_task_failed: - await self.on_task_failed(task_def, record.error) - return DAGExecutionResult( - success=False, - task_results=results, - error=str(exc), - task_outputs_by_type=dict(self.task_outputs_by_type), - failed_task_type=task_def.type.value, - failure_error=record.error, - ) - - return None diff --git a/backend/orchestration/dag_schema.py b/backend/orchestration/dag_schema.py deleted file mode 100644 index 5b9b394c02b2d804e1a002c92a53ee71a38a05ad..0000000000000000000000000000000000000000 --- a/backend/orchestration/dag_schema.py +++ /dev/null @@ -1,42 +0,0 @@ -from enum import StrEnum - -from pydantic import BaseModel, Field, field_validator - - -class TaskType(StrEnum): - CODE_GENERATION = "code_generation" - TEST_GENERATION = "test_generation" - STATIC_ANALYSIS = "static_analysis" - EXECUTE = "execute" - - -VALID_TASK_TYPES = frozenset(TaskType) - - -class DAGTaskDefinition(BaseModel): - id: str = Field(min_length=1, max_length=64) - type: TaskType - depends_on: list[str] = Field(default_factory=list) - - @field_validator("id") - @classmethod - def normalize_id(cls, value: str) -> str: - normalized = value.strip() - if not normalized: - raise ValueError("Task id cannot be empty") - return normalized - - @field_validator("depends_on") - @classmethod - def normalize_dependencies(cls, value: list[str]) -> list[str]: - return [dep.strip() for dep in value if dep.strip()] - - -class DAGPlan(BaseModel): - tasks: list[DAGTaskDefinition] = Field(min_length=1) - - def task_ids(self) -> set[str]: - return {task.id for task in self.tasks} - - def task_map(self) -> dict[str, DAGTaskDefinition]: - return {task.id: task for task in self.tasks} diff --git a/backend/orchestration/dag_validator.py b/backend/orchestration/dag_validator.py deleted file mode 100644 index 42560f871ebd325e8ac337b73f47f5a8436eea42..0000000000000000000000000000000000000000 --- a/backend/orchestration/dag_validator.py +++ /dev/null @@ -1,85 +0,0 @@ -class DAGValidationError(Exception): - """Raised when a planner-produced DAG fails validation.""" - - def __init__(self, errors: list[str]) -> None: - self.errors = errors - message = "; ".join(errors) - super().__init__(message) - - -def validate_dag(plan: "DAGPlan") -> None: - """Validate DAG structure, references, and acyclicity.""" - from backend.orchestration.dag_schema import DAGPlan as DAGPlanModel - - if not isinstance(plan, DAGPlanModel): - raise TypeError("plan must be a DAGPlan instance") - - errors: list[str] = [] - task_ids = plan.task_ids() - - if len(task_ids) != len(plan.tasks): - errors.append("Duplicate task ids are not allowed") - - for task in plan.tasks: - for dependency in task.depends_on: - if dependency == task.id: - errors.append(f"Task '{task.id}' cannot depend on itself") - elif dependency not in task_ids: - errors.append( - f"Task '{task.id}' depends on unknown task '{dependency}'" - ) - - if errors: - raise DAGValidationError(errors) - - if _has_cycle(plan): - raise DAGValidationError(["DAG contains a cycle"]) - - -def _has_cycle(plan: "DAGPlan") -> bool: - graph = {task.id: list(task.depends_on) for task in plan.tasks} - visited: set[str] = set() - stack: set[str] = set() - - def visit(node: str) -> bool: - if node in stack: - return True - if node in visited: - return False - visited.add(node) - stack.add(node) - for dependency in graph[node]: - if visit(dependency): - return True - stack.remove(node) - return False - - return any(visit(task_id) for task_id in graph) - - -def topological_order(plan: "DAGPlan") -> list[str]: - """Return a valid topological ordering of task ids.""" - validate_dag(plan) - - in_degree = {task.id: 0 for task in plan.tasks} - dependents: dict[str, list[str]] = {task.id: [] for task in plan.tasks} - - for task in plan.tasks: - for dependency in task.depends_on: - in_degree[task.id] += 1 - dependents[dependency].append(task.id) - - ready = [task_id for task_id, degree in in_degree.items() if degree == 0] - order: list[str] = [] - - while ready: - current = ready.pop(0) - order.append(current) - for dependent in dependents[current]: - in_degree[dependent] -= 1 - if in_degree[dependent] == 0: - ready.append(dependent) - - if len(order) != len(plan.tasks): - raise DAGValidationError(["DAG contains a cycle"]) - return order diff --git a/backend/orchestration/errors.py b/backend/orchestration/errors.py deleted file mode 100644 index 96af299b53e062bde75ef70d1f1278d1bc8c6ff5..0000000000000000000000000000000000000000 --- a/backend/orchestration/errors.py +++ /dev/null @@ -1,15 +0,0 @@ -class TaskExecutionError(Exception): - """Raised when a DAG task handler fails.""" - - def __init__(self, message: str, *, error_type: str = "TASK_EXECUTION_ERROR") -> None: - super().__init__(message) - self.message = message - self.error_type = error_type - - -class PlannerError(Exception): - """Raised when the planner agent fails.""" - - def __init__(self, message: str, *, error_type: str = "PLANNER_ERROR") -> None: - super().__init__(message) - self.error_type = error_type diff --git a/backend/orchestration/failure_classifier.py b/backend/orchestration/failure_classifier.py deleted file mode 100644 index 1441b572581fa7ab7113c1001c25579fd8197e4f..0000000000000000000000000000000000000000 --- a/backend/orchestration/failure_classifier.py +++ /dev/null @@ -1,78 +0,0 @@ -from dataclasses import dataclass -from enum import StrEnum -from typing import Any - - -class FailureCategory(StrEnum): - SYNTAX_ERROR = "SYNTAX_ERROR" - MISSING_DEPENDENCY = "MISSING_DEPENDENCY" - IMPORT_ERROR = "IMPORT_ERROR" - RUNTIME_ERROR = "RUNTIME_ERROR" - TEST_FAILURE = "TEST_FAILURE" - STATIC_ANALYSIS_ERROR = "STATIC_ANALYSIS_ERROR" - TIMEOUT = "TIMEOUT" - DOCKER_BUILD_ERROR = "DOCKER_BUILD_ERROR" - DOCKER_START_ERROR = "DOCKER_START_ERROR" - CODE_GENERATION_ERROR = "CODE_GENERATION_ERROR" - TEST_GENERATION_ERROR = "TEST_GENERATION_ERROR" - UNKNOWN = "UNKNOWN" - - -@dataclass -class ClassifiedFailure: - category: FailureCategory - message: str - details: dict[str, Any] - error_type: str - - def to_dict(self) -> dict[str, Any]: - return { - "category": self.category.value, - "message": self.message, - "details": self.details, - "error_type": self.error_type, - } - - -_ERROR_TYPE_MAP: dict[str, FailureCategory] = { - "SYNTAX_ERROR": FailureCategory.SYNTAX_ERROR, - "STATIC_ANALYSIS_ERROR": FailureCategory.STATIC_ANALYSIS_ERROR, - "TEST_FAILURE": FailureCategory.TEST_FAILURE, - "TIMEOUT": FailureCategory.TIMEOUT, - "DOCKER_BUILD_ERROR": FailureCategory.DOCKER_BUILD_ERROR, - "DOCKER_START_ERROR": FailureCategory.DOCKER_START_ERROR, - "RUNTIME_ERROR": FailureCategory.RUNTIME_ERROR, - "CODE_GENERATION_ERROR": FailureCategory.CODE_GENERATION_ERROR, - "TEST_GENERATION_ERROR": FailureCategory.TEST_GENERATION_ERROR, - "IMPORT_ERROR": FailureCategory.IMPORT_ERROR, -} - - -def classify_failure(error: dict[str, Any]) -> ClassifiedFailure: - error_type = str(error.get("error_type", "UNKNOWN")) - message = str(error.get("message", "Unknown failure")) - details = dict(error.get("details") or {}) - - stdout = str(error.get("stdout") or details.get("stdout") or "") - stderr = str(error.get("stderr") or details.get("stderr") or "") - combined = f"{stdout}\n{stderr}".lower() - - category = _ERROR_TYPE_MAP.get(error_type, FailureCategory.UNKNOWN) - - if "modulenotfounderror" in combined or "no module named" in combined: - category = FailureCategory.MISSING_DEPENDENCY - elif "importerror" in combined: - category = FailureCategory.IMPORT_ERROR - elif "syntaxerror" in combined: - category = FailureCategory.SYNTAX_ERROR - elif "timeout" in message.lower() or error_type == "TIMEOUT": - category = FailureCategory.TIMEOUT - elif "test" in message.lower() and error_type in {"TEST_FAILURE", "TASK_EXECUTION_ERROR"}: - category = FailureCategory.TEST_FAILURE - - return ClassifiedFailure( - category=category, - message=message, - details=details, - error_type=error_type, - ) diff --git a/backend/orchestration/recovery_runner.py b/backend/orchestration/recovery_runner.py deleted file mode 100644 index 36919d9d14c2daff31a37eea75387e63fb469d85..0000000000000000000000000000000000000000 --- a/backend/orchestration/recovery_runner.py +++ /dev/null @@ -1,302 +0,0 @@ -from dataclasses import dataclass, field -from pathlib import Path -from typing import Any - -from sqlalchemy.ext.asyncio import AsyncSession - -from backend.agents.diagnosis_agent import DiagnosisAgent, stub_diagnose -from backend.agents.repair_agent import RepairAgent, stub_repair -from backend.analysis.static_analyzer import StaticAnalyzer -from backend.core.config import get_settings -from backend.db.models import Workflow -from backend.events.publisher import EventPublisher -from backend.execution.docker_runner import DockerRunnerBase -from backend.execution.verification import VerificationPipeline -from backend.llm.base import LLMProvider -from backend.memory.attempt_repository import AttemptRepository -from backend.orchestration.dag_schema import DAGTaskDefinition, TaskType -from backend.orchestration.errors import TaskExecutionError -from backend.orchestration.failure_classifier import classify_failure -from backend.orchestration.retry_manager import RetryManager -from backend.orchestration.task_context import TaskContext - - -@dataclass -class ProjectState: - code_files: dict[str, dict[str, str]] = field(default_factory=dict) - test_files: dict[str, dict[str, str]] = field(default_factory=dict) - static_analysis: dict[str, Any] | None = None - - @property - def all_files(self) -> dict[str, dict[str, str]]: - merged = dict(self.code_files) - merged.update(self.test_files) - return merged - - def apply_repair(self, files: dict[str, dict[str, str]]) -> None: - for path, meta in files.items(): - if path.startswith("tests/"): - self.test_files[path] = meta - else: - self.code_files[path] = meta - - def to_task_outputs(self) -> dict[str, dict[str, Any]]: - outputs: dict[str, dict[str, Any]] = {} - if self.code_files: - outputs["code_generation"] = {"files": self.code_files, "status": "generated"} - if self.test_files: - outputs["test_generation"] = {"files": self.test_files, "status": "generated"} - if self.static_analysis: - outputs["static_analysis"] = self.static_analysis - return outputs - - -@dataclass -class RecoveryResult: - success: bool - repair_count: int = 0 - final_error: str | None = None - project_state: ProjectState | None = None - - -class RecoveryRunner: - """FAIL → Diagnose → Repair → Static Analysis → Docker → Tests → retry.""" - - def __init__( - self, - db: AsyncSession, - events: EventPublisher, - attempts: AttemptRepository, - ) -> None: - self.db = db - self.events = events - self.attempts = attempts - self.analyzer = StaticAnalyzer() - - async def run( - self, - *, - workflow: Workflow, - llm: LLMProvider, - project_state: ProjectState, - failure_error: dict[str, Any], - execute_task_id: int, - workspace_path: Path, - docker_runner: DockerRunnerBase | None, - use_stubs: bool = False, - max_repairs: int | None = None, - ) -> RecoveryResult: - retry_manager = RetryManager(max_repairs=max_repairs) - classified = classify_failure(failure_error) - current_error = failure_error - stdout = str(failure_error.get("stdout") or "") - stderr = str(failure_error.get("stderr") or "") - - previous_attempts = await self.attempts.list_attempts_for_workflow(workflow.id) - - while retry_manager.can_retry(): - repair_number = retry_manager.record_retry() - attempt_number = await self.attempts.get_next_attempt_number(workflow.id) - - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="RETRY_STARTED", - message=f"Retry {repair_number} started", - task_id=execute_task_id, - metadata={ - "repair_number": repair_number, - "attempt_number": attempt_number, - "max_repairs": retry_manager.max_repairs, - }, - ) - - if use_stubs: - diagnosis = await stub_diagnose(classified) - else: - diagnosis = await DiagnosisAgent(llm).diagnose( - user_prompt=workflow.user_prompt, - failure=classified, - code_files=project_state.code_files, - test_files=project_state.test_files, - stdout=stdout, - stderr=stderr, - previous_attempts=[{"attempt_number": a.attempt_number, "result": a.result} for a in previous_attempts], - ) - - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="FAILURE_DIAGNOSED", - message=diagnosis.root_cause, - task_id=execute_task_id, - metadata=diagnosis.to_dict(), - ) - - if use_stubs: - repair = stub_repair(project_state.code_files, project_state.test_files) - else: - repair = await RepairAgent(llm).repair( - user_prompt=workflow.user_prompt, - diagnosis=diagnosis, - code_files=project_state.code_files, - test_files=project_state.test_files, - ) - - project_state.apply_repair(repair.files) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="REPAIR_GENERATED", - message=repair.repair_summary, - task_id=execute_task_id, - metadata={"repair_summary": repair.repair_summary, "attempt_number": attempt_number}, - ) - - static_result = self.analyzer.analyze_files(project_state.all_files) - project_state.static_analysis = static_result.to_dict() - - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="STATIC_ANALYSIS_STARTED", - message="Static analysis during recovery", - task_id=execute_task_id, - ) - - if static_result.status == "FAIL": - attempt_number = await self.attempts.get_next_attempt_number(workflow.id) - await self.attempts.create_attempt( - workflow_id=workflow.id, - task_id=execute_task_id, - attempt_number=attempt_number, - generated_code={"files": project_state.code_files}, - generated_tests={"files": project_state.test_files}, - static_analysis_result=project_state.static_analysis, - result="FAILED", - error={"message": "Static analysis failed during recovery"}, - ) - retry_manager.register_attempt(attempt_number) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="STATIC_ANALYSIS_COMPLETED", - message="Static analysis failed during recovery", - task_id=execute_task_id, - metadata=static_result.to_dict(), - ) - current_error = { - "error_type": "STATIC_ANALYSIS_ERROR", - "message": "Static analysis failed during recovery", - "details": static_result.to_dict(), - } - classified = classify_failure(current_error) - continue - - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="STATIC_ANALYSIS_COMPLETED", - message="Static analysis passed during recovery", - task_id=execute_task_id, - metadata=static_result.to_dict(), - ) - - context = TaskContext( - workflow_id=workflow.workflow_id, - user_prompt=workflow.user_prompt, - llm=llm, - workspace_path=workspace_path, - task_def=DAGTaskDefinition(id="D", type=TaskType.EXECUTE, depends_on=[]), - task_outputs_by_type=project_state.to_task_outputs(), - docker_runner=docker_runner, - use_stubs=use_stubs, - ) - - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="CONTAINER_STARTED", - message="Starting container for recovery verification", - task_id=execute_task_id, - ) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TEST_STARTED", - message="Running tests during recovery", - task_id=execute_task_id, - ) - - try: - if docker_runner is None: - from backend.execution.docker_runner import DockerCliRunner - - docker_runner = DockerCliRunner() - - verification = await VerificationPipeline(docker_runner).run(context) - attempt_number = await self.attempts.get_next_attempt_number(workflow.id) - await self.attempts.create_attempt( - workflow_id=workflow.id, - task_id=execute_task_id, - attempt_number=attempt_number, - generated_code={"files": project_state.code_files}, - generated_tests={"files": project_state.test_files}, - static_analysis_result=project_state.static_analysis, - stdout=verification.get("stdout"), - stderr=verification.get("stderr"), - exit_code=verification.get("exit_code"), - result="SUCCESS", - ) - retry_manager.register_attempt(attempt_number) - - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TEST_PASSED", - message="Tests passed after recovery", - task_id=execute_task_id, - metadata={"attempt_number": attempt_number, "repair_number": repair_number}, - ) - return RecoveryResult( - success=True, - repair_count=repair_number, - project_state=project_state, - ) - except TaskExecutionError as exc: - attempt_number = await self.attempts.get_next_attempt_number(workflow.id) - await self.attempts.create_attempt( - workflow_id=workflow.id, - task_id=execute_task_id, - attempt_number=attempt_number, - generated_code={"files": project_state.code_files}, - generated_tests={"files": project_state.test_files}, - static_analysis_result=project_state.static_analysis, - stdout=getattr(exc, "stdout", None), - stderr=exc.message, - exit_code=getattr(exc, "exit_code", 1), - result="FAILED", - error={"error_type": exc.error_type, "message": exc.message}, - ) - retry_manager.register_attempt(attempt_number) - - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TEST_FAILED", - message=exc.message, - task_id=execute_task_id, - metadata={"error_type": exc.error_type, "repair_number": repair_number}, - ) - - current_error = {"error_type": exc.error_type, "message": exc.message} - classified = classify_failure(current_error) - stdout = "" - stderr = exc.message - - return RecoveryResult( - success=False, - repair_count=retry_manager.repair_count, - final_error=current_error.get("message", "Recovery exhausted"), - project_state=project_state, - ) diff --git a/backend/orchestration/retry_manager.py b/backend/orchestration/retry_manager.py deleted file mode 100644 index 3c41a794787c5b1c91de8043b8075cea16b15f81..0000000000000000000000000000000000000000 --- a/backend/orchestration/retry_manager.py +++ /dev/null @@ -1,46 +0,0 @@ -from dataclasses import dataclass, field - -from backend.core.config import get_settings - - -@dataclass -class RetryState: - max_repairs: int - repair_count: int = 0 - attempt_numbers: list[int] = field(default_factory=list) - - @property - def total_attempts(self) -> int: - return len(self.attempt_numbers) - - def can_retry(self) -> bool: - return self.repair_count < self.max_repairs - - def record_repair(self) -> int: - self.repair_count += 1 - return self.repair_count - - -class RetryManager: - """Enforces maximum repair attempts for autonomous recovery.""" - - def __init__(self, max_repairs: int | None = None) -> None: - settings = get_settings() - self.state = RetryState(max_repairs=max_repairs or settings.max_retries) - - def can_retry(self) -> bool: - return self.state.can_retry() - - def record_retry(self) -> int: - return self.state.record_repair() - - @property - def repair_count(self) -> int: - return self.state.repair_count - - @property - def max_repairs(self) -> int: - return self.state.max_repairs - - def register_attempt(self, attempt_number: int) -> None: - self.state.attempt_numbers.append(attempt_number) diff --git a/backend/orchestration/scheduler.py b/backend/orchestration/scheduler.py deleted file mode 100644 index 66d339825fac1f3a3b86f009824608fc59df7fd8..0000000000000000000000000000000000000000 --- a/backend/orchestration/scheduler.py +++ /dev/null @@ -1,33 +0,0 @@ -from backend.orchestration.dag_schema import DAGPlan - - -def get_ready_task_ids(plan: DAGPlan, completed: set[str]) -> list[str]: - """Return task ids whose dependencies are all completed.""" - ready: list[str] = [] - for task in plan.tasks: - if task.id in completed: - continue - if all(dep in completed for dep in task.depends_on): - ready.append(task.id) - return ready - - -def resolve_execution_layers(plan: DAGPlan) -> list[list[str]]: - """Group tasks into layers that can execute in parallel.""" - completed: set[str] = set() - layers: list[list[str]] = [] - remaining = plan.task_ids() - - while remaining: - ready = [ - task_id - for task_id in remaining - if all(dep in completed for dep in plan.task_map()[task_id].depends_on) - ] - if not ready: - break - layers.append(ready) - completed.update(ready) - remaining -= set(ready) - - return layers diff --git a/backend/orchestration/state_machine.py b/backend/orchestration/state_machine.py deleted file mode 100644 index d614afcbfb8b6abc23ac6e283bad2786181526a4..0000000000000000000000000000000000000000 --- a/backend/orchestration/state_machine.py +++ /dev/null @@ -1,45 +0,0 @@ -from backend.db.models import TaskStatus - - -class InvalidStateTransitionError(Exception): - """Raised when a task state transition is not allowed.""" - - def __init__(self, current: str, target: str) -> None: - super().__init__(f"Invalid transition from {current} to {target}") - self.current = current - self.target = target - - -TERMINAL_STATES = frozenset( - {TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.CANCELLED} -) - -_VALID_TRANSITIONS: dict[TaskStatus, frozenset[TaskStatus]] = { - TaskStatus.CREATED: frozenset({TaskStatus.READY, TaskStatus.CANCELLED}), - TaskStatus.READY: frozenset({TaskStatus.RUNNING, TaskStatus.CANCELLED}), - TaskStatus.RUNNING: frozenset( - {TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.CANCELLED} - ), - TaskStatus.FAILED: frozenset({TaskStatus.RETRYING, TaskStatus.CANCELLED}), - TaskStatus.RETRYING: frozenset({TaskStatus.RUNNING, TaskStatus.CANCELLED}), - TaskStatus.SUCCESS: frozenset(), - TaskStatus.CANCELLED: frozenset(), -} - - -def can_transition(current: str | TaskStatus, target: str | TaskStatus) -> bool: - current_status = TaskStatus(current) - target_status = TaskStatus(target) - return target_status in _VALID_TRANSITIONS.get(current_status, frozenset()) - - -def validate_transition(current: str | TaskStatus, target: str | TaskStatus) -> TaskStatus: - current_status = TaskStatus(current) - target_status = TaskStatus(target) - if not can_transition(current_status, target_status): - raise InvalidStateTransitionError(str(current_status), str(target_status)) - return target_status - - -def is_terminal(status: str | TaskStatus) -> bool: - return TaskStatus(status) in TERMINAL_STATES diff --git a/backend/orchestration/task_context.py b/backend/orchestration/task_context.py deleted file mode 100644 index bfe0034518cfeafd83c47b0613a977222081a3c2..0000000000000000000000000000000000000000 --- a/backend/orchestration/task_context.py +++ /dev/null @@ -1,42 +0,0 @@ -from dataclasses import dataclass, field -from pathlib import Path -from typing import Any - -from backend.llm.base import LLMProvider -from backend.orchestration.dag_schema import DAGTaskDefinition, TaskType - - -@dataclass -class TaskContext: - workflow_id: str - user_prompt: str - llm: LLMProvider - workspace_path: Path - task_def: DAGTaskDefinition - task_outputs: dict[str, dict[str, Any]] = field(default_factory=dict) - task_outputs_by_type: dict[str, dict[str, Any]] = field(default_factory=dict) - docker_runner: Any | None = None - use_stubs: bool = False - fail_task: str | None = None - stub_delay: float = 0 - stub_execute_calls_docker: bool = False - - def get_dependency_output(self, task_key: str) -> dict[str, Any] | None: - return self.task_outputs.get(task_key) - - @property - def generated_files(self) -> dict[str, dict[str, str]]: - code_output = self.task_outputs_by_type.get(TaskType.CODE_GENERATION.value) or {} - return code_output.get("files", {}) - - @property - def test_files(self) -> dict[str, dict[str, str]]: - test_output = self.task_outputs_by_type.get(TaskType.TEST_GENERATION.value) or {} - return test_output.get("files", {}) - - @property - def all_project_files(self) -> dict[str, dict[str, str]]: - merged: dict[str, dict[str, str]] = {} - merged.update(self.generated_files) - merged.update(self.test_files) - return merged diff --git a/backend/orchestration/workflow_runner.py b/backend/orchestration/workflow_runner.py deleted file mode 100644 index 4f476e4290490899b5fd04dac503dcaab3816e68..0000000000000000000000000000000000000000 --- a/backend/orchestration/workflow_runner.py +++ /dev/null @@ -1,467 +0,0 @@ -from pathlib import Path -from typing import Any - -from sqlalchemy.ext.asyncio import AsyncSession - -from backend.agents.planner import PlannerAgent -from backend.core.config import get_settings -from backend.db.models import TaskStatus, Workflow, WorkflowStatus -from backend.events.publisher import EventPublisher -from backend.llm.base import LLMProvider -from backend.memory.attempt_repository import AttemptRepository -from backend.memory.task_repository import TaskRepository -from backend.orchestration.dag_engine import DAGExecutor -from backend.orchestration.dag_schema import DAGPlan, DAGTaskDefinition, TaskType -from backend.orchestration.dag_validator import DAGValidationError, validate_dag -from backend.orchestration.errors import PlannerError -from backend.orchestration.recovery_runner import ProjectState, RecoveryRunner - - -class WorkflowRunner: - """Orchestrates planning, validation, persistence, DAG execution, and recovery.""" - - def __init__(self, db: AsyncSession) -> None: - self.db = db - self.tasks = TaskRepository(db) - self.attempts = AttemptRepository(db) - self.events = EventPublisher(db) - - async def run( - self, - workflow: Workflow, - llm: LLMProvider, - *, - execution_context: dict[str, Any] | None = None, - plan: DAGPlan | None = None, - ) -> Workflow: - settings = get_settings() - ctx = dict(execution_context or {}) - workspace_path = Path(ctx.get("workspace_path") or settings.workspace_root) / workflow.workflow_id - workspace_path.mkdir(parents=True, exist_ok=True) - - base_context = { - "workflow_id": workflow.workflow_id, - "user_prompt": workflow.user_prompt, - "llm": llm, - "workspace_path": workspace_path, - "docker_runner": ctx.get("docker_runner"), - "use_stubs": ctx.get("use_stubs", False), - "fail_task": ctx.get("fail_task"), - "stub_delay": ctx.get("stub_delay", 0), - "stub_execute_calls_docker": ctx.get("stub_execute_calls_docker", False), - "max_repairs": ctx.get("max_repairs"), - } - - await self.tasks.mark_workflow_status(workflow, WorkflowStatus.RUNNING) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="WORKFLOW_STARTED", - message="Workflow execution started", - ) - - try: - dag_plan = plan or await self._plan(workflow, llm) - validate_dag(dag_plan) - except (PlannerError, DAGValidationError) as exc: - await self._fail_workflow( - workflow, - message=str(exc), - error_type=exc.__class__.__name__, - ) - return workflow - - await self.tasks.save_planner_output(workflow, dag_plan.model_dump()) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="PLANNER_COMPLETED", - message="Planner produced a valid DAG", - metadata={"task_count": len(dag_plan.tasks)}, - ) - - created_tasks = await self.tasks.create_tasks_from_plan(workflow, dag_plan) - for task in created_tasks: - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TASK_CREATED", - message=f"Task {task.task_key} created", - task_id=task.id, - task_key=task.task_key, - metadata={"task_type": task.task_type}, - ) - - for task in created_tasks: - if not task.depends_on: - await self.tasks.transition_task(task, TaskStatus.READY) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TASK_READY", - message=f"Task {task.task_key} is ready", - task_id=task.id, - task_key=task.task_key, - ) - - completed_keys: set[str] = set() - - async def on_task_started(task_def: DAGTaskDefinition) -> None: - task = await self.tasks.get_task_by_key(workflow.id, task_def.id) - if task is None: - return - if task.status == TaskStatus.CREATED: - await self.tasks.transition_task(task, TaskStatus.READY) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TASK_READY", - message=f"Task {task_def.id} is ready", - task_id=task.id, - task_key=task_def.id, - ) - await self.tasks.transition_task(task, TaskStatus.RUNNING, set_started=True) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TASK_STARTED", - message=f"Task {task_def.id} started", - task_id=task.id, - task_key=task_def.id, - ) - if task_def.type in (TaskType.CODE_GENERATION, TaskType.TEST_GENERATION): - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="LLM_CALLED", - message=f"Calling LLM for {task_def.type.value}", - task_id=task.id, - task_key=task_def.id, - ) - if task_def.type == TaskType.STATIC_ANALYSIS: - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="STATIC_ANALYSIS_STARTED", - message=f"Static analysis started for task {task_def.id}", - task_id=task.id, - task_key=task_def.id, - ) - if task_def.type == TaskType.EXECUTE: - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="CONTAINER_STARTED", - message="Starting Docker sandbox", - task_id=task.id, - task_key=task_def.id, - ) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TEST_STARTED", - message="Running tests in container", - task_id=task.id, - task_key=task_def.id, - ) - if task_def.type in {TaskType.CODE_GENERATION, TaskType.TEST_GENERATION}: - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="LLM_CALLED", - message=f"Calling LLM for {task_def.type.value}", - task_id=task.id, - task_key=task_def.id, - ) - elif task_def.type == TaskType.STATIC_ANALYSIS: - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="STATIC_ANALYSIS_STARTED", - message=f"Static analysis started for task {task_def.id}", - task_id=task.id, - task_key=task_def.id, - ) - elif task_def.type == TaskType.EXECUTE: - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="CONTAINER_STARTED", - message="Starting Docker sandbox", - task_id=task.id, - task_key=task_def.id, - ) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TEST_STARTED", - message="Running tests in container", - task_id=task.id, - task_key=task_def.id, - ) - - async def on_task_completed(task_def: DAGTaskDefinition, output: dict) -> None: - task = await self.tasks.get_task_by_key(workflow.id, task_def.id) - if task is None: - return - await self.tasks.transition_task( - task, - TaskStatus.SUCCESS, - output=output, - set_completed=True, - ) - completed_keys.add(task_def.id) - await self._persist_attempt(workflow.id, task, task_def.type, output, success=True) - event_type = _event_type_for_task(task_def.type.value) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type=event_type, - message=f"Task {task_def.id} completed", - task_id=task.id, - task_key=task_def.id, - metadata={"output_status": output.get("status")}, - ) - if task_def.type == TaskType.EXECUTE: - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TEST_PASSED", - message="All tests passed", - task_id=task.id, - task_key=task_def.id, - ) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TASK_COMPLETED", - message=f"Task {task_def.id} marked complete", - task_id=task.id, - task_key=task_def.id, - ) - await self._mark_dependents_ready(workflow, dag_plan, completed_keys) - - async def on_task_failed(task_def: DAGTaskDefinition, error: dict) -> None: - task = await self.tasks.get_task_by_key(workflow.id, task_def.id) - if task is None: - return - await self.tasks.transition_task( - task, TaskStatus.FAILED, output=error, set_completed=True - ) - await self._persist_attempt( - workflow.id, task, task_def.type, error, success=False - ) - event_type = "TEST_FAILED" if task_def.type == TaskType.EXECUTE else "TASK_FAILED" - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type=event_type, - message=error.get("message", "Task failed"), - task_id=task.id, - task_key=task_def.id, - metadata=error, - ) - - executor = DAGExecutor( - base_context=base_context, - on_task_started=on_task_started, - on_task_completed=on_task_completed, - on_task_failed=on_task_failed, - ) - result = await executor.execute(dag_plan) - - if result.success: - await self._complete_workflow(workflow) - return workflow - - if _is_recoverable(result): - project_state = _project_state_from_outputs(result.task_outputs_by_type) - execute_task = await self.tasks.get_task_by_type(workflow.id, TaskType.EXECUTE.value) - execute_task_id = execute_task.id if execute_task else 0 - failure_error = result.failure_error or { - "error_type": "WORKFLOW_EXECUTION_ERROR", - "message": result.error or "Workflow failed", - } - - recovery = await RecoveryRunner(self.db, self.events, self.attempts).run( - workflow=workflow, - llm=llm, - project_state=project_state, - failure_error=failure_error, - execute_task_id=execute_task_id, - workspace_path=workspace_path, - docker_runner=base_context.get("docker_runner"), - use_stubs=base_context.get("use_stubs", False), - max_repairs=base_context.get("max_repairs"), - ) - - if recovery.success: - if execute_task: - await self.tasks.transition_task( - execute_task, - TaskStatus.SUCCESS, - output={"status": "PASS", "recovered": True, "repair_count": recovery.repair_count}, - set_completed=True, - ) - await self._complete_workflow( - workflow, - metadata={"recovered": True, "repair_count": recovery.repair_count}, - ) - else: - await self._fail_workflow( - workflow, - message=recovery.final_error or "Recovery exhausted", - error_type="WORKFLOW_EXECUTION_ERROR", - metadata={"repair_count": recovery.repair_count}, - ) - return workflow - - await self._fail_workflow( - workflow, - message=result.error or "Workflow execution failed", - error_type="WORKFLOW_EXECUTION_ERROR", - ) - return workflow - - async def _complete_workflow( - self, - workflow: Workflow, - *, - metadata: dict | None = None, - ) -> None: - await self.tasks.mark_workflow_status( - workflow, WorkflowStatus.SUCCESS, completed=True - ) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="WORKFLOW_COMPLETED", - message="Workflow completed successfully", - metadata=metadata, - ) - - async def _persist_attempt( - self, - workflow_db_id: int, - task, - task_type: TaskType, - payload: dict, - *, - success: bool, - ) -> None: - attempt_number = await self.attempts.get_next_attempt_number(workflow_db_id) - kwargs: dict[str, Any] = { - "workflow_id": workflow_db_id, - "task_id": task.id, - "attempt_number": attempt_number, - "result": "SUCCESS" if success else "FAILED", - } - if task_type == TaskType.CODE_GENERATION: - kwargs["generated_code"] = payload if success else None - if not success: - kwargs["error"] = payload - elif task_type == TaskType.TEST_GENERATION: - kwargs["generated_tests"] = payload if success else None - if not success: - kwargs["error"] = payload - elif task_type == TaskType.STATIC_ANALYSIS: - kwargs["static_analysis_result"] = payload if success else None - if not success: - kwargs["error"] = payload - elif task_type == TaskType.EXECUTE: - kwargs["stdout"] = payload.get("stdout") - kwargs["stderr"] = payload.get("stderr") - kwargs["exit_code"] = payload.get("exit_code") - if success: - kwargs["generated_code"] = {"files": payload.get("files")} - else: - kwargs["error"] = payload - await self.attempts.create_attempt(**kwargs) - - async def _mark_dependents_ready( - self, - workflow: Workflow, - dag_plan: DAGPlan, - completed_keys: set[str], - ) -> None: - for task_def in dag_plan.tasks: - if task_def.id in completed_keys: - continue - if not all(dep in completed_keys for dep in task_def.depends_on): - continue - task = await self.tasks.get_task_by_key(workflow.id, task_def.id) - if task is None or task.status != TaskStatus.CREATED: - continue - await self.tasks.transition_task(task, TaskStatus.READY) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="TASK_READY", - message=f"Task {task_def.id} is ready", - task_id=task.id, - task_key=task_def.id, - ) - - async def _plan(self, workflow: Workflow, llm: LLMProvider) -> DAGPlan: - planner = PlannerAgent(llm) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="LLM_CALLED", - message="Calling planner LLM", - ) - return await planner.create_plan(workflow.user_prompt) - - async def _fail_workflow( - self, - workflow: Workflow, - *, - message: str, - error_type: str, - metadata: dict | None = None, - ) -> None: - await self.tasks.mark_workflow_status( - workflow, WorkflowStatus.FAILED, completed=True - ) - event_metadata = {"error_type": error_type} - if metadata: - event_metadata.update(metadata) - await self.events.publish( - workflow_db_id=workflow.id, - public_workflow_id=workflow.workflow_id, - event_type="WORKFLOW_FAILED", - message=message, - metadata=event_metadata, - ) - - -def _event_type_for_task(task_type: str) -> str: - mapping = { - "code_generation": "CODE_GENERATED", - "test_generation": "TESTS_GENERATED", - "static_analysis": "STATIC_ANALYSIS_COMPLETED", - "execute": "TASK_COMPLETED", - } - return mapping.get(task_type, "TASK_COMPLETED") - - -def _is_recoverable(result) -> bool: - if not result.task_outputs_by_type.get("code_generation"): - return False - recoverable_types = { - TaskType.EXECUTE.value, - TaskType.STATIC_ANALYSIS.value, - TaskType.TEST_GENERATION.value, - } - return result.failed_task_type in recoverable_types - - -def _project_state_from_outputs(outputs: dict[str, dict[str, Any]]) -> ProjectState: - code = outputs.get("code_generation", {}).get("files", {}) - tests = outputs.get("test_generation", {}).get("files", {}) - static_analysis = outputs.get("static_analysis") - return ProjectState( - code_files=dict(code), - test_files=dict(tests), - static_analysis=static_analysis, - ) diff --git a/backend/requirements.txt b/backend/requirements.txt deleted file mode 100644 index d41021d7d5e83ad5e0e70d76b1f8034546fed6ba..0000000000000000000000000000000000000000 --- a/backend/requirements.txt +++ /dev/null @@ -1,19 +0,0 @@ -fastapi>=0.115.0 -uvicorn[standard]>=0.32.0 -sqlalchemy[asyncio]>=2.0.36 -asyncpg>=0.30.0 -alembic>=1.14.0 -pydantic[email]>=2.10.0 -pydantic-settings>=2.6.0 -python-jose[cryptography]>=3.3.0 -passlib[bcrypt]>=1.7.4 -bcrypt>=4.0.0,<4.1.0 -cryptography>=43.0.0 -httpx>=0.28.0 -python-multipart>=0.0.17 - -# Dev / test -pytest>=8.3.0 -pytest-asyncio>=0.24.0 -pytest-cov>=6.0.0 -aiosqlite>=0.20.0