"""Codegen domain - Pydantic schemas.""" from __future__ import annotations from datetime import datetime from typing import Any, Literal, Optional from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, model_validator class CodegenRunCreate(BaseModel): prompt_id: Optional[UUID] = None user_prompt: str = "" # Optional client speech-to-text to combine with text and/or attachments. voice_transcript: Optional[str] = None # UUIDs from POST …/prompt-media/upload (same project). attachment_ids: Optional[list[UUID]] = None # "agent": standard code generation (default) # "plan": read-only planning pass — generates a markdown plan, no code changes # "visual_edit": AI-assisted text replacement via element metadata run_mode: Literal["agent", "plan", "visual_edit"] = "agent" execution_target: Literal["cloud", "local"] = "cloud" # Visual edit only: DOM element descriptors captured by the frontend # Each element should contain: filePath, lineNumber, selector, textContent, etc. selected_elements: Optional[list[dict]] = None @model_validator(mode="after") def _require_some_user_content(self) -> "CodegenRunCreate": text = (self.user_prompt or "").strip() voice = (self.voice_transcript or "").strip() ids = self.attachment_ids or [] if not text and not voice and not ids: raise ValueError( "Provide user_prompt text, voice_transcript, and/or attachment_ids" ) return self class CodegenCreditEstimateResponse(BaseModel): minimum_credits: int maximum_credits: int prompt_credits: int attachment_credits: int class CodegenRunResponse(BaseModel): id: UUID project_id: UUID prompt_id: Optional[UUID] status: str run_mode: str plan: Optional[dict] plan_document: Optional[str] parent_plan_id: Optional[UUID] result_commit_sha: Optional[str] result_pr_url: Optional[str] result_summary: Optional[str] revision_id: Optional[UUID] error_message: Optional[str] agent_transcript: Optional[str] = None tokens_used: Optional[int] started_at: Optional[datetime] completed_at: Optional[datetime] created_at: datetime run_type: Optional[str] = None pending_user_inputs: Optional[dict] = None interruption_payload: Optional[dict[str, Any]] = None model_config = {"from_attributes": True} @classmethod def from_run(cls, run) -> "CodegenRunResponse": """Build response with explicit run_type (plan_only vs code_applied).""" data = cls.model_validate(run).model_dump() data["run_type"] = "code_applied" if run.revision_id else "plan_only" return cls(**data) class CodegenRunSummary(BaseModel): """Same as CodegenRunResponse but omits ``agent_transcript`` (can be very large). Used for list endpoints; load the full run via GET …/runs/{id} for replay text. """ id: UUID project_id: UUID prompt_id: Optional[UUID] = None status: str run_mode: str plan: Optional[dict] = None plan_document: Optional[str] = None parent_plan_id: Optional[UUID] = None result_commit_sha: Optional[str] = None result_pr_url: Optional[str] = None result_summary: Optional[str] = None revision_id: Optional[UUID] = None error_message: Optional[str] = None tokens_used: Optional[int] = None started_at: Optional[datetime] = None completed_at: Optional[datetime] = None created_at: datetime run_type: Optional[str] = None pending_user_inputs: Optional[dict] = None model_config = {"from_attributes": True} @classmethod def from_run(cls, run) -> "CodegenRunSummary": data = cls.model_validate(run).model_dump() data["run_type"] = "code_applied" if run.revision_id else "plan_only" return cls(**data) class SupplyUserInputsBody(BaseModel): """Values keyed by manifest item ``id`` (see ``pending_user_inputs`` on the codegen run).""" values: dict[str, str] = Field(default_factory=dict) class ResolveCapabilitiesBody(BaseModel): """Body for POST .../actions/{run}/provision -- extended beyond provision.json alone (see resolve_codegen_capabilities in workers/tasks/codegen.py) to also carry p2hs/feature-packs.json's install/compose entries, resolved by the same pre-push gate before the GitHub Actions runner's own push. `provision` is wrapped (not inlined at this body's top level) rather than reusing ProvisionManifest's own "extra=ignore" schema directly here, which would silently drop pack_installs/ pack_compositions if they were siblings of its own fields.""" provision: dict[str, Any] = Field(default_factory=dict) pack_installs: list[dict[str, Any]] = Field(default_factory=list) pack_compositions: list[dict[str, Any]] = Field(default_factory=list) class CodegenActionsCompleteBody(BaseModel): """Completion callback posted by the repo-local GitHub Actions runner.""" status: Literal["completed", "completed_with_l3", "failed", "canceled", "needs_continuation", "deploying", "blocked"] l3_candidate_ids: list[str] = Field(default_factory=list) l3_candidates: list[dict[str, Any]] = Field(default_factory=list) provider: Optional[str] = None commit_sha: Optional[str] = None branch: Optional[str] = None summary: Optional[str] = None transcript: Optional[str] = None files_changed: Optional[list[str]] = None files_changed_detail: Optional[list[dict[str, Any]]] = None tokens_used: Optional[int] = None pending_user_inputs: Optional[dict[str, Any]] = None pending_provision: Optional[dict[str, Any]] = None pending_email_templates: Optional[dict[str, Any]] = None # Resolved by resolveCapabilitiesBeforePush before this push -- only ever carries # OPTIONAL gaps (a required one means the runner never pushed and reported # status="blocked" instead, with no commit_sha here at all). Threaded through so # an optional gap still reaches the final run status/message for this execution # path instead of being silently discarded once the pre-push gate merely # returned "ready" rather than raising. pending_pack_install_outcomes: Optional[list[dict[str, Any]]] = None pending_pack_composition_outcomes: Optional[list[dict[str, Any]]] = None legacy_users: Optional[list[dict[str, Any]]] = None error_message: Optional[str] = None # Work-branch mode (see work_branch.py). Core only trusts these against the state it stored # itself: a report naming another branch, or a malformed sha, is ignored. work_branch: Optional[str] = Field(default=None, max_length=120) checkpoint: Optional[Literal["partial_work", "agent_output", "capabilities_provisioned", "promoted"]] = None checkpoint_sha: Optional[str] = Field(default=None, max_length=64) failure_class: Optional[str] = Field(default=None, max_length=40) retryable: Optional[bool] = None correction: Optional[dict[str, Any]] = None cursor_session_id: Optional[str] = None continuation_attempt: Optional[int] = None # Measured runner execution interval; excludes GitHub queue time. execution_seconds: Optional[int] = Field(default=None, ge=0, le=86400) class CloudCutoverUser(BaseModel): email: str = Field(min_length=3) role: str = "member" name: str = "" class CloudCutoverBody(BaseModel): users: list[CloudCutoverUser] = Field(default_factory=list) class CodegenActionsProgressBody(BaseModel): """Incremental output callback posted by the repo-local GitHub Actions runner.""" text: str stream: Literal["stdout", "stderr"] = "stdout" detail_format: Optional[str] = None class CapabilityApprovalResponse(BaseModel): """One security-sensitive capability the coding agent asked for, and the owner's decision.""" id: UUID name: str kind: str summary: str grants: list[dict[str, str]] status: str requested_run_id: Optional[UUID] = None created_at: datetime decided_at: Optional[datetime] = None model_config = ConfigDict(from_attributes=True)