"""Core codegen composition for unattended agent build requests.""" import asyncio import os from pathlib import Path from types import SimpleNamespace import structlog from src.config import get_settings from src.core.exceptions import ValidationError from src.domain.agent_access.operations import MAX_ATTACHMENT_CHARS, MAX_ATTACHMENTS_TOTAL_CHARS from src.domain.codegen.credit_estimator import estimate_codegen_credits from src.domain.codegen.import_snapshot import snapshots_dir, write_snapshot from src.domain.codegen.repository import CodegenRunRepository from src.domain.codegen.service import CodegenService from src.domain.codegen.source_adoption import ( analyze_imported_source, build_source_adoption_plan, redact_source, ) from src.domain.instructions.repository import InstructionRepository from src.domain.manifest.repository import ManifestRunRepository from src.domain.manifest.service import ManifestService from src.domain.projects.repository import ( ProjectEnvVarRepository, ProjectRepository, ProjectSettingsRepository, ) from src.domain.prompt_media.repository import PromptMediaRepository logger = structlog.get_logger() _ADOPTION_LABELS = { "add": "Add this alongside what already exists in the project.", "replace": "Replace the relevant existing surface with this.", "merge": "Merge this into the existing project, reconciling with what's already there.", "inject": "Inject this into the existing project at the place the request names, leaving the rest as it is.", } # What "the project's own framework" means for each project type, so a conversion is # delivered in that stack instead of the attached files being copied in and served as they are. _FRAMEWORK_NAMES = { "nuxt_app": "Nuxt (Vue): pages/, layouts/, components/ and composables/", "react_app": "React: routes and components", } def _adoption_line(source_adoption: str | None, project_type: str | None) -> str: """Say what the adoption mode means for THIS project. "migrate" used to have no description at all (the agent was told to infer it). In a framework project, converting or replacing a design means rebuilding it in that framework with an identical rendered result; serving the attached files as static files is not a conversion. A static html project is the place for keeping pages as they are. An explicit instruction in the request still wins: this states the default.""" framework = _FRAMEWORK_NAMES.get(project_type or "") if source_adoption == "migrate": if framework: return ( f"Convert this existing app into this project's own framework ({framework}). Rebuild each " "page as that framework's routes, layouts and components so the RENDERED result is identical: " "same markup structure, copy, class names, styling, animation and motion. Bring the CSS in as " "it is, port the JavaScript behavior into the framework (composables and mounted hooks), and " "move page data into modules. Do not satisfy this by copying the attached files in and serving " "them as static files (for example from public/): that is not a conversion." ) return ( "Convert this existing app to run in this project, keeping the rendered result identical: same " "markup structure, copy, class names, styling, animation and motion." ) if source_adoption == "replace" and framework: return ( f"{_ADOPTION_LABELS['replace']} Implement it in this project's own framework ({framework}) with an " "identical rendered result, unless the request explicitly asks for the files to be served as they " "are (that belongs in a static html project)." ) return _ADOPTION_LABELS.get(source_adoption, "Adoption mode not specified — infer from the request text.") def _format_attachments( attachments, source_adoption: str | None = None, deployment_intent: str = "modify_only", project_type: str | None = None, ) -> str: """Render pasted design/build source as a clearly delimited, labeled block. Stays inside the user-role prompt (never the system prompt — see prompt_wrapper.py's build_system_prompt/build_agent_user_prompt split) so it carries the same trust tier as the rest of the request: reference material for codegen to reproduce, still subject to P2HS's own composition authority and platform contracts, never literal files to copy into a private backend. """ if not attachments: return "" adoption_line = _adoption_line(source_adoption, project_type) parts = [ "## Attached design/build source (from the requesting agent)", f"Adoption mode: {adoption_line}", ( f"Deployment intent: {deployment_intent}. Reproduce this design closely. " "It is reference material, not platform code — implement it through " "P2HS's existing App Block/Feature Pack/Cloud contracts, the same as " "any other request; do not copy backend/API calls from it verbatim." ), "Priority when the attached source conflicts with platform architecture:\n" "1. P2HS security/tenancy/platform invariants\n" "2. P2HS admitted contracts and composition semantics\n" "3. This request's explicit instructions\n" "4. The attached source's behavior and integrations\n" "5. The attached source's implementation details", "Concerns P2HS already owns (auth, records/persistence, file uploads, " "email, generic messaging, generic workflow triggers, payments) must be " "adapted to the matching P2HS capability/contract — never duplicated as a " "second backend, even if the attached source already implements one (its " "own database client, its own auth, its own workflow engine). Genuine " "third-party/customer-owned integrations that are not a P2HS capability " "(a customer's own ERP, a niche SaaS API, a proprietary internal service) " "must be preserved, not deleted and not invented into a P2HS capability " "that doesn't exist. Never preserve a hardcoded secret or API key from the " "attached source — extract it via required-user-inputs.json or platform " "env vars the same way any other request would.", ] for attachment in attachments: content = redact_source(attachment.content) fence = "````" if "```" in content else "```" parts.append(f"### {attachment.path}\n{fence}\n{content}\n{fence}") return "\n\n".join(parts) _HISTORY_FILES_SHOWN = 8 _ORIGINS = {"dashboard_import": "the dashboard"} def _chat_display(prompt: str, attachments, prompt_source: str | None = None) -> str: """The prompt as the dashboard shows it: redacted, with a line about attached source files.""" text = redact_source(prompt or "").strip() if not attachments: return text names = [a.path for a in attachments] shown = ", ".join(names[:_HISTORY_FILES_SHOWN]) more = f" and {len(names) - _HISTORY_FILES_SHOWN} more" if len(names) > _HISTORY_FILES_SHOWN else "" plural = "s" if len(names) != 1 else "" origin = _ORIGINS.get(prompt_source or "", "an AI connection") note = f"[Sent from {origin} with {len(names)} attached source file{plural}: {shown}{more}]" return text + chr(10) + chr(10) + note class AgentCodegenFacade: def __init__(self, db): self._db = db self.projects = ProjectRepository(db) self.instructions = InstructionRepository(db) self.service = CodegenService( codegen_repo=CodegenRunRepository(db), project_facade=self, project_settings_repo=ProjectSettingsRepository(db), env_var_repo=ProjectEnvVarRepository(db), manifest_service=ManifestService(db=db, run_repo=ManifestRunRepository(db)), ) async def get_project_type(self, project_id): project = await self.projects.get_by_id(project_id) return project.type if project else None async def get_project_instruction_content(self, project_id): return await self.instructions.get_current_content(project_id) async def get_project_repo_full_name(self, project_id): project = await self.projects.get_with_repo(project_id) return project.repo.repo_full_name if project and getattr(project, "repo", None) else None async def _resolve_attachment_ids(self, project_id, attachment_ids): """Validate binary attachment_ids belong to this project and fit the combined-byte-size budget before they reach CodegenService.create_run. Assets themselves were already materialized by confirm_asset_upload.""" if not attachment_ids: return [] assets = await PromptMediaRepository(self._db).list_by_ids_for_project( attachment_ids, project_id ) if len(assets) != len(set(attachment_ids)): raise ValidationError("One or more attachment_ids do not belong to this project") total_bytes = sum(a.byte_size for a in assets) max_total = get_settings().agent_asset_upload_total_bytes if total_bytes > max_total: raise ValidationError( f"Attached files exceed the {max_total // (1024 * 1024)} MiB combined limit" ) return [a.id for a in assets] async def _resolve_attachments(self, project_id, attachments): """Resolve each AttachmentInput to (path, content) text, reading asset_id-backed ones from disk (uploaded via request_asset_upload + confirm_asset_upload) instead of requiring the caller to retype large source files inline, where transcription risks silent corruption. Re-enforces the per-file/total character budget on the resolved text, since asset-backed content bypasses the schema's Field(max_length=...) (which only bounds the inline `content` field itself).""" if not attachments: return [] asset_ids = [a.asset_id for a in attachments if a.asset_id is not None] assets_by_id = {} if asset_ids: rows = await PromptMediaRepository(self._db).list_by_ids_for_project( asset_ids, project_id ) if len(rows) != len(set(asset_ids)): raise ValidationError( "One or more attachment asset_ids do not belong to this project" ) assets_by_id = {row.id: row for row in rows} settings = get_settings() upload_root = os.path.normpath(settings.prompt_media_upload_root) resolved = [] total_chars = 0 for attachment in attachments: if attachment.content is not None: text = attachment.content else: asset = assets_by_id[attachment.asset_id] rel = asset.storage_relative_path.replace("/", os.sep) abs_path = os.path.join(upload_root, rel) try: raw = await asyncio.to_thread(Path(abs_path).read_bytes) text = raw.decode("utf-8") except (OSError, UnicodeDecodeError) as exc: raise ValidationError( f"Attachment '{attachment.path}' could not be read as text" ) from exc if len(text) > MAX_ATTACHMENT_CHARS: raise ValidationError( f"Attachment '{attachment.path}' exceeds the " f"{MAX_ATTACHMENT_CHARS:,}-character per-file limit" ) total_chars += len(text) resolved.append(SimpleNamespace(path=attachment.path, content=text)) if total_chars > MAX_ATTACHMENTS_TOTAL_CHARS: raise ValidationError( f"Attachments exceed the {MAX_ATTACHMENTS_TOTAL_CHARS:,}-character total limit" ) return resolved async def _snapshot_attachments(self, project, grant, attachments, prompt_source): """Preserve the received files (see import_snapshot.py). Best-effort: the prompt still carries the source, so a disk problem must not fail the build.""" project_id = project.id try: return await asyncio.to_thread( write_snapshot, snapshots_dir(get_settings(), project.workspace_id, project_id), [(a.path, a.content) for a in attachments], source={ "kind": "mcp", "prompt_source": prompt_source, "grant_id": str(grant.id), }, ) except OSError as exc: logger.warning( "import_snapshot_write_failed", project_id=str(project_id), error_type=type(exc).__name__, ) return None async def _record_chat_prompt( self, project, grant, run, prompt, attachments, source_adoption, prompt_source ) -> None: """Show an MCP-started build in the project's prompt history, like a prompt typed in the builder. A run started from an AI connection used to have no Prompt, no chat row and no session, so the dashboard showed nothing, and (assistant replies are only written for a run that has a prompt) no replies either. This records what the agent actually asked for, redacted, plus a line saying how many source files came with it; the files themselves stay in the run and its import snapshot. Best-effort inside a savepoint: a history problem must never fail or roll back the build. """ from src.domain.codegen.run_chat_repository import RunChatMessageRepository from src.domain.instructions.repository import PromptRepository, PromptSessionRepository try: async with self._db.begin_nested(): sessions = PromptSessionRepository(self._db) session_id = getattr(project, "active_session_id", None) if session_id: existing = await sessions.get_by_id(session_id) if not existing or existing.project_id != project.id: session_id = None if not session_id: session_id = (await sessions.create(project.id, grant.created_by, None)).id await self.projects.update(project.id, active_session_id=session_id) display = _chat_display(prompt, attachments, prompt_source) row = await PromptRepository(self._db).create( session_id, display, grant.created_by, { "source": prompt_source or "mcp_agent", "grant_id": str(grant.id) if getattr(grant, "id", None) else None, "source_adoption": source_adoption, "attachment_paths": [a.path for a in attachments][:100], }, ) run.prompt_id = row.id await RunChatMessageRepository(self._db).create( project_id=project.id, session_id=session_id, run_id=run.id, role="user", content=display, message_type="message", prompt_id=row.id, ) except Exception as exc: logger.warning( "agent_prompt_history_failed", project_id=str(project.id), run_id=str(getattr(run, "id", "")), error_type=type(exc).__name__, ) async def start( self, grant, body, *, prompt_source=None, attachments=None, import_snapshot=None ): """Start a build from an actor (an AI connection's grant, or a person: ``id`` may be None). ``attachments`` (objects with ``path`` and ``content``) and ``import_snapshot`` (the manifest of the snapshot that holds them) let a caller that already stored the files, the dashboard's import, reuse everything after that point unchanged: the same adoption plan, prompt, history and dispatch. """ project = await self.projects.get_by_id(body.project_id) if attachments is None: attachments = await self._resolve_attachments( body.project_id, getattr(body, "attachments", None) or [] ) import_snapshot = ( await self._snapshot_attachments(project, grant, attachments, prompt_source) if attachments and project else None ) attachment_ids = await self._resolve_attachment_ids( body.project_id, getattr(body, "attachment_ids", None) or [] ) deployment_intent = getattr(body, "deployment_intent", "modify_only") analyses = [ analyze_imported_source([{"path": attachment.path, "content": attachment.content}]) for attachment in attachments ] existing_composition = getattr(project, "discovery_json", None) if project else None adoption_plan = build_source_adoption_plan( analyses=analyses, operation=getattr(body, "source_adoption", None), deployment_intent=deployment_intent, existing_composition=existing_composition if isinstance(existing_composition, dict) else None, ) attachments_block = _format_attachments( attachments, getattr(body, "source_adoption", None), deployment_intent, getattr(project, "type", None), ) if attachments: import json attachments_block += ( "\n\n## Imported source analysis and bounded adoption plan\n" + json.dumps( {"analysis": analyses, "plan": adoption_plan}, sort_keys=True, separators=(",", ":"), ) ) user_prompt = f"{body.prompt}\n\n{attachments_block}" if attachments_block else body.prompt estimate = estimate_codegen_credits( prompt=body.prompt, attachment_sizes=[len(a.content.encode()) for a in attachments], attachment_count=len(attachments), ) run = await self.service.create_run( project_id=body.project_id, user_prompt=user_prompt, created_by=grant.created_by, credit_estimate=estimate.__dict__, supersede_active=False, prompt_source=prompt_source, initiating_grant_id=getattr(grant, "id", None), attachment_ids=attachment_ids or None, generation_request={ "schema": "source_adoption_request.v1", "source_adoption": getattr(body, "source_adoption", None), "deployment_intent": deployment_intent, "adopt_attachment": getattr(body, "adopt_attachment", None), "source_analysis": analyses, "source_adoption_plan": adoption_plan, # Manifest only (paths, sizes, sha256); the files live on Core disk. "import_snapshot": import_snapshot, } if attachments else None, ) if project: await self._record_chat_prompt( project, grant, run, body.prompt, attachments, getattr(body, "source_adoption", None), prompt_source, ) return run