Task

Prompt

""" AgentOS – Task ORM model. A Task is the top-level durable record for one user request. It owns the lifecycle state machine and links to all subtasks, events, artifacts, and the execution plan. """ from __future__ import annotations import uuid from datetime import datetime from typing import TYPE_CHECKING, List, Optional from sqlalchemy import ( Boolean, DateTime, Float, ForeignKey, Index, Integer, String, Text, func, ) from sqlalchemy.dialects.postgresql import ARRAY, JSONB, UUID from sqlalchemy.orm import Mapped, mapped_column, relationship from core.database import Base from core.state_machine import TaskStatus if TYPE_CHECKING: from models.artifact import Artifact from models.event import TaskEvent from models.plan import TaskPlan from models.subtask import Subtask from models.user import User, Workspace class Task(Base): __tablename__ = "tasks" __table_args__ = ( Index("ix_tasks_user_id_created_at", "user_id", "created_at"), Index("ix_tasks_workspace_id_status", "workspace_id", "status"), ) id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 ) # Human-readable short identifier (ULID-style prefix, generated at intake) short_id: Mapped[str] = mapped_column( String(26), nullable=False, unique=True, index=True ) user_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True, ) workspace_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("workspaces.id", ondelete="CASCADE"), nullable=False, index=True, ) # ── Core fields ─────────────────────────────────────────────────────────── title: Mapped[str] = mapped_column(String(500), nullable=False) description: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # Task lifecycle status status: Mapped[str] = mapped_column( String(30), nullable=False, default=TaskStatus.CREATED.value, index=True, ) # Classification output task_category: Mapped[Optional[str]] = mapped_column(String(60), nullable=True) confidence_score: Mapped[Optional[float]] = mapped_column(Float, nullable=True) risk_level: Mapped[Optional[str]] = mapped_column(String(20), nullable=True) required_capabilities: Mapped[Optional[list]] = mapped_column( ARRAY(String), nullable=True ) estimated_cost_usd: Mapped[Optional[float]] = mapped_column(Float, nullable=True) estimated_latency_seconds: Mapped[Optional[int]] = mapped_column( Integer, nullable=True ) execution_tier: Mapped[Optional[str]] = mapped_column(String(30), nullable=True) # Input references (S3 keys of uploaded files, URLs, etc.) input_references: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) # Requested output formats preferred_output_formats: Mapped[Optional[list]] = mapped_column( ARRAY(String), nullable=True ) # Budget constraints max_cost_usd: Mapped[Optional[float]] = mapped_column(Float, nullable=True) max_latency_seconds: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) max_retries: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) max_repair_attempts: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) # Policy flags requires_approval: Mapped[bool] = mapped_column( Boolean, nullable=False, default=False ) approval_reason: Mapped[Optional[str]] = mapped_column(Text, nullable=True) approved_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True ) approved_by_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="SET NULL"), nullable=True ) # Cost tracking actual_cost_usd: Mapped[Optional[float]] = mapped_column(Float, nullable=True) total_tokens_used: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) # Execution context (checkpoint state for restart-safe recovery) execution_context: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) # Final output final_output_summary: Mapped[Optional[str]] = mapped_column(Text, nullable=True) final_manifest: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) known_limitations: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True) # Correlation identifier for distributed tracing correlation_id: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) # Retry tracking retry_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) repair_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) # Timestamps created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now(), ) queued_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), 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 ) failed_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True ) # Error information error_code: Mapped[Optional[str]] = mapped_column(String(60), nullable=True) error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True) error_details: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) # ── Relationships ───────────────────────────────────────────────────────── user: Mapped[Optional["User"]] = relationship( "User", back_populates="tasks", foreign_keys=[user_id], lazy="select" ) workspace: Mapped["Workspace"] = relationship( "Workspace", back_populates="tasks", lazy="select" ) subtasks: Mapped[List["Subtask"]] = relationship( "Subtask", back_populates="task", cascade="all, delete-orphan", order_by="Subtask.created_at", lazy="select", ) events: Mapped[List["TaskEvent"]] = relationship( "TaskEvent", back_populates="task", cascade="all, delete-orphan", order_by="TaskEvent.occurred_at", lazy="select", ) artifacts: Mapped[List["Artifact"]] = relationship( "Artifact", back_populates="task", cascade="all, delete-orphan", order_by="Artifact.created_at", lazy="select", ) plan: Mapped[Optional["TaskPlan"]] = relationship( "TaskPlan", back_populates="task", uselist=False, cascade="all, delete-orphan", lazy="select", ) def __repr__(self) -> str: return f"<Task id={self.id} short_id={self.short_id!r} status={self.status!r}>" 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!