agents/contracts/Task.py

import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Literal

from contracts.base import BaseContract


class Atomicity(str, Enum):
    ATOMIC      = "atomic"        # single clear action, one owner, < 1 day
    NEEDS_SPLIT = "needs_split"   # still too big, recurse


class TaskKind(str, Enum):
    INVESTIGATE = "investigate"   # read code, reproduce, research
    DESIGN      = "design"        # API/schema decisions
    IMPLEMENT   = "implement"     # write code
    MIGRATE     = "migrate"       # DB / data changes
    TEST        = "test"          # unit/integration/e2e
    DOCS        = "docs"          # docs, changelog
    REVIEW      = "review"        # code review, verification
    OPS         = "ops"           # deploy, config, feature flag


@dataclass
class Task(BaseContract):
    id: str
    title: str
    kind: TaskKind
    description: str
    agent: str
    estimate_p50: int
    estimate_p90: int
    risk: Literal["low", "medium", "high"]
    acceptance_criteria: list[str] = field(default_factory=list)
    depends_on: list[str] = field(default_factory=list)
    atomicity: Atomicity = Atomicity.ATOMIC
    estimate_minutes: Optional[float] = None
    files_hint: list[str] = field(default_factory=list)   # probable touch points
    open_questions: list[str] = field(default_factory=list)

    @staticmethod
    def new(**kw) -> "Task":
        return Task(id=str(uuid.uuid4())[:8], **kw)

    def render_dbg(self) -> str:
        lines = [f"# Task: {self.description}"]
        deps = f" (after {', '.join(self.depends_on)})" if self.depends_on else ""
        lines.append(f"- [{self.kind.value}] {self.id}: {self.title}{deps}")
        for ac in self.acceptance_criteria:
            lines.append(f"    ✓ {ac}")
        for q in self.open_questions:
            lines.append(f"    ? {q}")

        return "\n".join(lines)

    def to_prompt_text(self) -> str:
        lines = [
            f"# Task: {self.description}",
            f"# Before start you must ensure that another tasks have been done: {', '.join(self.depends_on)}" if self.depends_on else "",
            f"# Acceptance criteria: \n{"\n - ".join(self.acceptance_criteria)}"
        ]

        return "\n".join(lines)

    def is_filled(self) -> bool:
        return len(self.open_questions) == 0 and self.atomicity == Atomicity.ATOMIC