Subtaskpy
Prompt
""" AgentOS – Subtask ORM model. A Subtask is one node in the directed-acyclic-graph execution plan for a Task. It carries its own lifecycle, dependency list, capability requirement, budget, and result reference. """ 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 SubtaskStatus if TYPE_CHECKING: from models.artifact import Artifact from models.task import Task class Subtask(Base): __tablename__ = "subtasks" __table_args__ = ( Index("ix_subtasks_task_id_status", "task_id", "status"), Index("ix_subtasks_task_id_order", "task_id", "execution_order"), ) id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 ) task_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False, index=True, ) # ── Graph positioning ───────────────────────────────────────────────────── # Human-readable label within the plan label: Mapped[str] = mapped_column(String(200), nullable=False) # Topological execution order (computed at planning time) execution_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0) # UUIDs of subtasks that must be COMPLETED before this one can start dependency_ids: Mapped[Optional[list]] = mapped_column( ARRAY(UUID(as_uuid=True)), nullable=True ) # Subtasks that can run in parallel with this one (informational) parallel_with_ids: Mapped[Optional[list]] = mapped_column( ARRAY(UUID(as_uuid=True)), nullable=True ) # ── Specification ───────────────────────────────────────────────────────── objective: Mapped[str] = mapped_column(Text, nullable=False) acceptance_criteria: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True) # JSON-serialized scoped input references (S3 keys, structured data) inputs: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) expected_outputs: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) # Which capability class handles this subtask capability_class: Mapped[str] = mapped_column(String(60), nullable=False) # Worker-specific configuration overrides worker_config: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) # ── Budget ──────────────────────────────────────────────────────────────── max_retries: Mapped[int] = mapped_column(Integer, nullable=False, default=3) max_time_budget_seconds: Mapped[int] = mapped_column( Integer, nullable=False, default=300 ) max_cost_budget_usd: Mapped[float] = mapped_column( Float, nullable=False, default=1.0 ) risk_level: Mapped[str] = mapped_column( String(20), nullable=False, default="low" ) # "low" | "medium" | "high" | "critical" # ── Status ──────────────────────────────────────────────────────────────── status: Mapped[str] = mapped_column( String(20), nullable=False, default=SubtaskStatus.PENDING.value, index=True, ) # ── Result ──────────────────────────────────────────────────────────────── # S3 key for the full structured worker result JSON result_key: Mapped[Optional[str]] = mapped_column(String(512), nullable=True) result_summary: Mapped[Optional[str]] = mapped_column(Text, nullable=True) confidence_estimate: Mapped[Optional[float]] = mapped_column(Float, nullable=True) validation_notes: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True) tool_call_records: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True) resource_usage: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) evidence_references: Mapped[Optional[list]] = mapped_column(JSONB, nullable=True) # ── Execution tracking ──────────────────────────────────────────────────── retry_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) repair_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) # Celery task id of the currently running worker job celery_task_id: Mapped[Optional[str]] = mapped_column(String(200), nullable=True) # Name of the worker type that handled this subtask worker_type: Mapped[Optional[str]] = mapped_column(String(60), nullable=True) # Correlation id for distributed tracing correlation_id: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) # ── 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 ) elapsed_seconds: Mapped[Optional[float]] = mapped_column(Float, nullable=True) # ── Error ───────────────────────────────────────────────────────────────── 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 ───────────────────────────────────────────────────────── task: Mapped["Task"] = relationship("Task", back_populates="subtasks", lazy="select") artifacts: Mapped[List["Artifact"]] = relationship( "Artifact", back_populates="subtask", cascade="all, delete-orphan", lazy="select", ) def __repr__(self) -> str: return ( f"<Subtask id={self.id} label={self.label!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!
