"""Codegen domain - SQLAlchemy models.""" from __future__ import annotations from datetime import datetime from typing import Optional from uuid import UUID, uuid4 import sqlalchemy as sa from sqlalchemy import ForeignKey, Integer, String, Text from sqlalchemy.dialects.postgresql import JSONB, UUID from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy.sql import func from sqlalchemy.types import DateTime from src.core.database import Base # 001 + 004 + 020 enum values; create_type=False matches existing PostgreSQL type. CODEGEN_RUN_STATUS_ENUM = sa.Enum( "pending", "queued", "running", "planning", "generating", "applying", "validating", "preview_refreshing", "completed", "failed", "canceled", "plan_ready", # migration 020: plan mode terminal state (awaiting user approval) "awaiting_user_input", # legacy; migrated to user_input_required in 026 "clarification_required", # migration 026: AI questions before/during work "user_input_required", # migration 026: manifest / concrete values (replaces awaiting_user_input) "completed_with_limitations", # migration 129: an optional requirement was a primitive_gap "blocked", # migration 129: a required requirement was a primitive_gap or a pack install was blocked "completed_with_l3", # validated project-scoped extension; canonical Architecture remains unchanged name="p2hs_codegen_run_status", create_type=False, ) class CodegenRun(Base): __tablename__ = "codegen_runs" id: Mapped[UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid4 ) project_id: Mapped[UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, ) prompt_id: Mapped[Optional[UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("prompts.id"), nullable=True ) status: Mapped[str] = mapped_column( CODEGEN_RUN_STATUS_ENUM, nullable=False, server_default="pending", ) plan: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) result_commit_sha: Mapped[Optional[str]] = mapped_column( String(40), nullable=True ) result_pr_url: Mapped[Optional[str]] = mapped_column(Text, nullable=True) result_summary: Mapped[Optional[str]] = mapped_column(Text, nullable=True) revision_id: Mapped[Optional[UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("project_revisions.id"), nullable=True, ) error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True) tokens_used: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) created_by: Mapped[Optional[UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id"), nullable=True ) started_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True ) completed_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() ) # Worker resume: e.g. pushed = git complete, DB finalize pending. worker_checkpoint: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) checkpoint_payload: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) # --- migration 020: plan mode + visual edit + websocket --- # "agent" (default) | "plan" | "visual_edit" run_mode: Mapped[str] = mapped_column(String(20), nullable=False, server_default="agent") # Markdown plan document generated in plan mode (status=plan_ready) plan_document: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # FK to self: set on agent/visual_edit runs that were executed from an approved plan parent_plan_id: Mapped[Optional[UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("codegen_runs.id", ondelete="SET NULL"), nullable=True, ) # Visual edit metadata: array of selected DOM element descriptors from the frontend selected_elements: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True) # Captured stdout/stderr lines from repo-based agents (Cursor, Claude CLI) for UI replay + ops logs agent_transcript: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # Copy of p2hs/required-user-inputs.json after a successful repo-based run (dashboard collects values). pending_user_inputs: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) # Latest blocking interruption: clarification_required | user_input_required (normalized JSON). interruption_payload: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) # --- migration 082: desktop execution agent (Phase 0) --- # "cloud" (default: GitHub Actions / in-process executor) | "local" (paired desktop device) execution_target: Mapped[str] = mapped_column( String(20), nullable=False, server_default="cloud" ) device_id: Mapped[Optional[UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("devices.id", ondelete="SET NULL"), nullable=True, ) claimed_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True ) class RunChatMessage(Base): """Normalized chat row for a codegen run (user + assistant).""" __tablename__ = "run_chat_messages" id: Mapped[UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid4) project_id: Mapped[UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False ) session_id: Mapped[UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("prompt_sessions.id", ondelete="CASCADE"), nullable=False ) run_id: Mapped[UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("codegen_runs.id", ondelete="CASCADE"), nullable=False ) role: Mapped[str] = mapped_column(String(16), nullable=False) message_type: Mapped[str] = mapped_column(String(32), nullable=False, server_default="message") content: Mapped[str] = mapped_column(Text, nullable=False) payload: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) prompt_id: Mapped[Optional[UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("prompts.id", ondelete="SET NULL"), nullable=True ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() ) class ProjectCapabilityApproval(Base): """An owner's decision on one security-sensitive capability the coding agent declared. The agent only says WHAT it needs (see requirements_spec.py); Core derives how it is enforced. A declaration that grants anonymous access, any delete, sends email or accepts public form submissions, or installs a pack that is not in the project's plan, is held until a workspace admin approves it. The decision is keyed by a fingerprint of exactly what is granted, so it survives retries and later runs, and changes (and is asked again) only when the grant changes. status: pending | approved | rejected. kind: compose | pack_install. """ __tablename__ = "project_capability_approvals" __table_args__ = ( sa.UniqueConstraint("project_id", "fingerprint", name="uq_capability_approval_fingerprint"), ) id: Mapped[UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid4) project_id: Mapped[UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("projects.id", ondelete="CASCADE"), nullable=False, index=True ) fingerprint: Mapped[str] = mapped_column(String(64), nullable=False) kind: Mapped[str] = mapped_column(String(16), nullable=False) name: Mapped[str] = mapped_column(String(100), nullable=False) summary: Mapped[str] = mapped_column(Text, nullable=False) grants: Mapped[list] = mapped_column(JSONB, nullable=False, server_default=sa.text("'[]'::jsonb")) status: Mapped[str] = mapped_column(String(16), nullable=False, server_default="pending") requested_run_id: Mapped[Optional[UUID]] = mapped_column(UUID(as_uuid=True), nullable=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False ) decided_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) decided_by: Mapped[Optional[UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True )