Cgen

Prompt

Send back the complete code with all the fixes. Fix each of the listed errors one by one, making sure to actually correct them so that there are 0 errors remaining. Keep the original imports, since the files exist. Write out every single character; do not abbreviate anything. Fix every error. There must be exactly one file. Do not write anything else; just output the complete code, and it must not contain any comments. Never, under any circumstances, use simplified, substitute, dummy, simulated, or fake code. Write the entire file as complete, unabridged, production-ready code in a single code block. It must be 100% error-free, a complete, error-free file, and must be submitted as a downloadable file. These requirements are mandatory and must be strictly adhered to. If no list of errors is provided, you must find all the errors and fix them. If there were comments in the original code, delete them. And most importantly: YOU MUST NEVER SIMPLIFY! """ AgentOS – Code generation worker. Converts a software-oriented subtask into complete source files, a test suite, and a build verification step. All generation is done via the LLM, and output is validated by running tests in the code-execution sandbox. """ from __future__ import annotations import json import uuid from typing import Any, Dict, List, Optional from core.config import settings from core.logging_config import get_logger from core.storage import artifact_key, upload_text from schemas.worker import ResourceUsage, WorkerInput, WorkerResult from services.workers.base import BaseWorker logger = get_logger(__name__) _SPEC_SYSTEM_PROMPT = """You are a senior software architect. Given a software requirement, produce a comprehensive technical specification. Return JSON: { "title": str, "language": str, "framework": str or null, "file_tree": ["src/main.py", "tests/test_main.py", ...], "interfaces": [{"name": str, "signature": str, "description": str}], "dependencies": [{"name": str, "version": str, "purpose": str}], "database_schema": str or null, "test_plan": ["test case description", ...], "acceptance_criteria": ["..."], "architecture_notes": "..." }""" _CODE_SYSTEM_PROMPT = """You are an expert software engineer. Generate complete, production-ready source code for the following specification. Rules: 1. Every file must be complete – no ellipses, no placeholders, no TODO stubs. 2. Include all necessary imports. 3. Include docstrings for all public functions and classes. 4. Handle errors explicitly; do not use bare except. 5. Include comprehensive tests. Return JSON: { "files": [ {"path": "src/main.py", "content": "...", "description": "..."}, ... ], "build_command": "pip install -r requirements.txt && python -m pytest tests/", "notes": "..." }""" class CodeGenerationWorker(BaseWorker): """Code generation capability worker.""" capability_class = "code_generation" async def _run(self, worker_input: WorkerInput) -> WorkerResult: specification: str = worker_input.inputs.get( "specification", worker_input.objective ) language: str = worker_input.inputs.get("language", "python") framework: Optional[str] = worker_input.inputs.get("framework") acceptance_criteria: List[str] = worker_input.inputs.get( "acceptance_criteria", worker_input.acceptance_criteria, ) # ── Step 1: Generate technical specification ─────────────────────────── spec_content, spec_tokens = await self._llm_call( system_prompt=_SPEC_SYSTEM_PROMPT, user_prompt=( f"Requirement: {specification}\n" f"Language: {language}\n" f"Framework: {framework or 'none specified'}\n" f"Acceptance criteria: {json.dumps(acceptance_criteria)}" ), response_format={"type": "json_object"}, temperature=0.2, ) try: tech_spec: Dict[str, Any] = json.loads(spec_content) except json.JSONDecodeError: tech_spec = {"error": "JSON parse failed", "raw": spec_content[:500]} # ── Step 2: Generate complete source files ───────────────────────────── code_content, code_tokens = await self._llm_call( system_prompt=_CODE_SYSTEM_PROMPT, user_prompt=( f"Technical specification:\n{json.dumps(tech_spec, indent=2)}\n\n" f"Original requirement: {specification}" ), response_format={"type": "json_object"}, temperature=0.1, ) try: code_result: Dict[str, Any] = json.loads(code_content) except json.JSONDecodeError: return WorkerResult( subtask_id=worker_input.subtask_id, status="needs_repair", summary="Code generation returned non-JSON output.", error_code="JSON_PARSE_FAILED", error_message=code_content[:500], confidence_estimate=0.1, ) files: List[Dict[str, Any]] = code_result.get("files", []) if not files: return WorkerResult( subtask_id=worker_input.subtask_id, status="needs_repair", summary="No files generated.", error_code="NO_FILES_GENERATED", error_message="LLM returned empty files list.", confidence_estimate=0.0, ) # ── Step 3: Upload generated files to object storage ─────────────────── output_keys: List[str] = [] for file_spec in files: file_path: str = file_spec.get("path", f"output_{uuid.uuid4().hex[:8]}.txt") file_content: str = file_spec.get("content", "") if not file_content: continue key = artifact_key( str(worker_input.task_id), f"codegen_{worker_input.subtask_id}/{file_path}", ) await upload_text( bucket=settings.MINIO_ARTIFACTS_BUCKET, key=key, text=file_content, content_type="text/plain", ) output_keys.append(key) # Upload spec document spec_key = artifact_key( str(worker_input.task_id), f"codegen_{worker_input.subtask_id}/technical_spec.json", ) await upload_text( bucket=settings.MINIO_ARTIFACTS_BUCKET, key=spec_key, text=json.dumps( {"specification": tech_spec, "code_result": code_result}, default=str, indent=2, ), content_type="application/json", ) output_keys.append(spec_key) # ── Step 4: Verify generated files address acceptance criteria ───────── criteria_notes: List[str] = [] combined_code = "\n".join(f.get("content", "") for f in files) for i, criterion in enumerate(acceptance_criteria or []): keywords = [ w.lower() for w in criterion.split() if len(w) > 3 and w.isalpha() ] coverage = ( sum(1 for kw in keywords if kw in combined_code.lower()) / max(len(keywords), 1) if keywords else 1.0 ) notes_entry = ( f"Criterion {i+1}: {criterion!r} → coverage {coverage:.0%}" ) criteria_notes.append(notes_entry) return WorkerResult( subtask_id=worker_input.subtask_id, status="completed", summary=( f"Generated {len(files)} file(s) for {language} project. " f"Build command: {code_result.get('build_command', 'N/A')}" ), output_keys=output_keys, output_data={ "files": [ {"path": f.get("path"), "description": f.get("description", "")} for f in files ], "build_command": code_result.get("build_command", ""), "language": language, "framework": tech_spec.get("framework"), "file_count": len(files), "dependencies": tech_spec.get("dependencies", []), "test_plan": tech_spec.get("test_plan", []), }, validation_notes=[ f"Files generated: {len(files)}", f"Language: {language}", ] + criteria_notes, confidence_estimate=0.80, resource_usage=ResourceUsage(tokens_total=spec_tokens + code_tokens), )