"""Durable, idempotent submission of agent mutations.""" import hashlib import json from typing import Any, Literal from uuid import UUID, uuid4 from pydantic import Field, model_validator from src.core.exceptions import ConflictError, NotFoundError from src.domain.agent_access.models import AgentOperation from src.domain.agent_access.rate_limit import enforce_tool_rate_limit from src.domain.agent_access.schemas import StrictModel from src.domain.agent_access.service import prompt_source_for, utcnow class CreateProject(StrictModel): request_key: str = Field(min_length=8, max_length=128) # Required only when the connection covers several workspaces (see list_workspaces). workspace_id: UUID | None = None name: str = Field(min_length=1, max_length=100) project_type: Literal["html", "react_app", "nuxt_app"] = "nuxt_app" description: str = Field(default="", max_length=2000) # Authoritative attachment bounds. The MCP Worker and the Dashboard proxy each # raise their own request-body ceiling to fit this budget plus JSON escaping and # envelope overhead (see P2HS-MCP/src/index.ts, P2HS-Dash server/api/agent-access); # this is the one place that decides what "too large" actually means. # Sized off a real multi-page static-site export (29 files, ~646,000 chars) that # used to force splitting across builds; keep this in sync with both mirrors above. MAX_ATTACHMENTS = 60 MAX_ATTACHMENT_CHARS = 150_000 MAX_ATTACHMENTS_TOTAL_CHARS = 1_000_000 # Binary media (photos/video) referenced by id -- see asset_upload.py. Separate, # smaller cap than MAX_ATTACHMENTS: each one is a real uploaded file, not a # pasted text snippet, so the practical count per build is much lower. MAX_ATTACHMENT_IDS = 20 class AttachmentInput(StrictModel): """Pasted design/build source (e.g. an HTML/CSS/React export from another AI tool) — text only. Never a request to fetch/execute anything; codegen treats it as untrusted design reference material, same trust tier as the prompt. Exactly one of content or asset_id: content is for typing a file's text directly into the request; asset_id is for a file already uploaded via request_asset_upload + confirm_asset_upload (see agent_facade.py, which resolves it to text) -- for large source files, retyping content by hand risks silent corruption that uploading the exact bytes avoids. """ path: str = Field(min_length=1, max_length=200) content: str | None = Field(default=None, max_length=MAX_ATTACHMENT_CHARS) asset_id: UUID | None = None @model_validator(mode="after") def _exactly_one_content_source(self): if (self.content is None) == (self.asset_id is None): raise ValueError("Provide exactly one of content or asset_id") if self.content is not None and not self.content: raise ValueError("content must not be empty") return self class StartBuild(StrictModel): request_key: str = Field(min_length=8, max_length=128) project_id: UUID prompt: str = Field(min_length=1, max_length=32_000) attachments: list[AttachmentInput] = Field(default_factory=list, max_length=MAX_ATTACHMENTS) # Binary media (photos/video), uploaded beforehand via request_asset_upload + # confirm_asset_upload and referenced here by id -- see agent_facade.py, which # validates project ownership and the combined-byte-size cap before this reaches # CodegenService.create_run's own attachment_ids parameter. attachment_ids: list[UUID] = Field(default_factory=list, max_length=MAX_ATTACHMENT_IDS) # Explicit signal for how attached source relates to the project, so codegen # doesn't have to infer intent from prose alone. None when there are no # attachments, or when the caller doesn't know/care which mode applies. source_adoption: Literal["add", "merge", "inject", "replace", "migrate"] | None = None deployment_intent: Literal["modify_only", "modify_and_deploy"] = "modify_only" adopt_attachment: dict[str, Any] | None = None @model_validator(mode="after") def _bound_total_attachment_size(self): # Only bounds the inline (content=) attachments here -- asset_id-backed # ones have no text yet at this point (see AttachmentInput). The full # combined total, including resolved asset text, is re-checked in # agent_facade.py once those are read off disk. total = sum(len(a.content) for a in self.attachments if a.content is not None) if total > MAX_ATTACHMENTS_TOTAL_CHARS: raise ValueError( f"Attachments exceed the {MAX_ATTACHMENTS_TOTAL_CHARS:,}-character total limit" ) return self def ensure_ui_source(self) -> None: """Refuse a "migrate" that carries no UI markup, before anything is queued or charged. Only the paths matter, so this holds for content= and asset_id= (zip) attachments alike. Raised as Core's own ValidationError with a machine code rather than as a pydantic error: the router reduces a schema error to a bare "Invalid tool arguments", and the calling agent needs to be told what to send. The code is on the public allowlist in the Dashboard proxy.""" from src.core.exceptions import ValidationError as CoreValidationError from src.domain.codegen.source_adoption import has_ui_markup, missing_ui_source_message if self.source_adoption != "migrate": return paths = [a.path for a in self.attachments] if not has_ui_markup(paths): raise CoreValidationError(missing_ui_source_message(paths), error="ui_source_required") class GetOperation(StrictModel): request_key: str = Field(min_length=8, max_length=128) class RequestAssetUpload(StrictModel): """Ask for a short-lived direct-upload URL for one binary file (photo/video). No request_key: nothing durable is written until confirm_asset_upload, so a plain retry is harmless -- unlike create_project/start_build/purchases, this doesn't go through AgentOperationService's idempotent mutation queue. """ project_id: UUID filename: str = Field(min_length=1, max_length=200) content_type: str = Field(min_length=1, max_length=100) byte_size: int = Field(gt=0) class ConfirmAssetUpload(StrictModel): """Materialize a PromptMediaAsset from a file the caller has already PUT to the URL returned by request_asset_upload. Idempotent by asset_id existence. filename must match what was passed to request_asset_upload -- it's part of the R2 object key, and confirm_asset_upload only carries asset_id/project_id (not the whole upload_url) so the key is rebuilt from these three values. """ project_id: UUID asset_id: UUID filename: str = Field(min_length=1, max_length=200) class AgentOperationService: def __init__(self, access): self.access = access self.repo = access.repo @staticmethod def view(operation): return { "operation_id": str(operation.id), "request_key": operation.request_key, "action": operation.action, "status": operation.status, "result": operation.result, } async def submit(self, principal, action, body): scope = { "create_project": "projects.create", "start_build": "builds.run", "create_credit_checkout": "credits.checkout", "purchase_credits": "credits.purchase", }[action] workspace_id = await self.access.require_scope( principal, scope, getattr(body, "project_id", None), getattr(body, "workspace_id", None), ) if "workspace_id" in type(body).model_fields: # Pin the resolved workspace so the same request_key hashes identically # whether the caller named the workspace or it was implied. body = body.model_copy(update={"workspace_id": workspace_id}) # Serialize retries for this grant and recheck revocation under the same lock. current = await self.repo.grant(principal.id, lock=True) self.access.require_active(current) arguments = body.model_dump(mode="json", exclude={"request_key"}) digest = hashlib.sha256( json.dumps( {"action": action, "arguments": arguments}, sort_keys=True, separators=(",", ":") ).encode() ).hexdigest() previous = await self.repo.operation(principal.id, body.request_key) if previous: if previous.request_hash != digest: raise ConflictError("Request key was already used with different arguments") return self.view(previous) # Only a genuinely new submission consumes the rate limit: a retry of an # already-accepted request_key returned above without reaching this point. await enforce_tool_rate_limit( self.access.redis, grant_id=principal.id, workspace_id=workspace_id, tool_name=action, ) operation = await self.repo.add( AgentOperation( id=uuid4(), grant_id=principal.id, workspace_id=workspace_id, request_key=body.request_key, action=action, request_hash=digest, arguments=arguments, prompt_source=prompt_source_for(self.access.settings, principal), status="queued", result={}, created_at=utcnow(), updated_at=utcnow(), ) ) return self.view(operation) async def get(self, principal, request_key): operation = await self.repo.operation(principal.id, request_key) if not operation: raise NotFoundError("Operation not found") scope = { "create_project": "projects.create", "start_build": "builds.run", "create_credit_checkout": "credits.checkout", "purchase_credits": "credits.purchase", }[operation.action] project_id = operation.arguments.get("project_id") # Also proves the operation's workspace is still part of this grant. await self.access.require_scope( principal, scope, UUID(project_id) if project_id else None, operation.workspace_id, ) return self.view(operation)