""" core/task_brief.py Dataclass-based schema for a TaskBrief — the fundamental unit of work passed between tiers in the-agency pipeline. """ from __future__ import annotations import uuid from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Optional # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _now_iso() -> str: """Return the current UTC time as an ISO-8601 string.""" return datetime.now(timezone.utc).isoformat() def _new_uuid() -> str: return str(uuid.uuid4()) # --------------------------------------------------------------------------- # TaskBrief # --------------------------------------------------------------------------- @dataclass class TaskBrief: """ Immutable-intent carrier that travels down (and sometimes back up) the T1→T5 tier hierarchy. Fields ------ brief_id : Unique identifier for this brief (UUID). run_id : Identifier for the top-level orchestration run. parent_brief_id : brief_id of the parent, None for the T1 root brief. tier : Tier level 1–5. role : Role key used to look up an agent personality file. goal_anchor : High-level goal set by T1; NEVER mutated by child tiers. workstream : Logical workstream name (e.g. "backend", "infra"). task : Concrete description of what this brief asks for. acceptance_criteria : Ordered list of pass/fail criteria. constraints : Hard constraints the agent must respect. context : Arbitrary key/value context bag. retry_budget : Maximum number of retry attempts allowed. retry_count : How many retries have been consumed so far. preferred_runtime : Execution runtime hint ("standard" | "coding_agent"). agent_personality : Optional path to an agent .md persona file. created_at : ISO-8601 creation timestamp (UTC). """ # Identity brief_id: str = field(default_factory=_new_uuid) run_id: str = field(default_factory=_new_uuid) parent_brief_id: Optional[str] = None # Tier / role tier: int = 1 role: str = "" # Goal — set once by T1, propagated unchanged goal_anchor: str = "" # Work description workstream: str = "" task: str = "" acceptance_criteria: list[str] = field(default_factory=list) constraints: list[str] = field(default_factory=list) context: dict[str, Any] = field(default_factory=dict) # Retry tracking retry_budget: int = 3 retry_count: int = 0 # Runtime / persona preferred_runtime: str = "standard" # "standard" | "coding_agent" agent_personality: Optional[str] = None # path to .md file # Metadata created_at: str = field(default_factory=_now_iso) # ------------------------------------------------------------------ # Validation # ------------------------------------------------------------------ def validate(self) -> None: """ Raise ValueError if any required field is missing or out of range. Call this before handing a brief to a runner or storing it. """ errors: list[str] = [] if not self.brief_id: errors.append("brief_id must not be empty") if not self.run_id: errors.append("run_id must not be empty") if self.tier not in range(1, 6): errors.append(f"tier must be 1–5, got {self.tier!r}") if not self.role: errors.append("role must not be empty") if not self.goal_anchor: errors.append("goal_anchor must not be empty") if not self.task: errors.append("task must not be empty") if self.retry_budget < 0: errors.append(f"retry_budget must be >= 0, got {self.retry_budget}") if self.retry_count < 0: errors.append(f"retry_count must be >= 0, got {self.retry_count}") if self.retry_count > self.retry_budget: errors.append( f"retry_count ({self.retry_count}) exceeds retry_budget ({self.retry_budget})" ) if self.preferred_runtime not in ("standard", "coding_agent"): errors.append( f"preferred_runtime must be 'standard' or 'coding_agent', got {self.preferred_runtime!r}" ) if errors: raise ValueError("TaskBrief validation failed:\n" + "\n".join(f" - {e}" for e in errors)) # ------------------------------------------------------------------ # Serialisation # ------------------------------------------------------------------ def to_dict(self) -> dict[str, Any]: """Serialise to a plain Python dict (JSON-safe).""" return { "brief_id": self.brief_id, "run_id": self.run_id, "parent_brief_id": self.parent_brief_id, "tier": self.tier, "role": self.role, "goal_anchor": self.goal_anchor, "workstream": self.workstream, "task": self.task, "acceptance_criteria": list(self.acceptance_criteria), "constraints": list(self.constraints), "context": dict(self.context), "retry_budget": self.retry_budget, "retry_count": self.retry_count, "preferred_runtime": self.preferred_runtime, "agent_personality": self.agent_personality, "created_at": self.created_at, } @classmethod def from_dict(cls, data: dict[str, Any]) -> "TaskBrief": """Deserialise from a plain dict. Unknown keys are silently ignored.""" known_fields = {f.name for f in cls.__dataclass_fields__.values()} # type: ignore[attr-defined] filtered = {k: v for k, v in data.items() if k in known_fields} return cls(**filtered) # ------------------------------------------------------------------ # Factory: child brief # ------------------------------------------------------------------ def make_child_brief( self, *, tier: int, role: str, task: str, workstream: str = "", acceptance_criteria: Optional[list[str]] = None, constraints: Optional[list[str]] = None, context: Optional[dict[str, Any]] = None, preferred_runtime: str = "standard", agent_personality: Optional[str] = None, retry_budget: int = 3, ) -> "TaskBrief": """ Create a child brief that inherits run_id and — critically — goal_anchor verbatim from this parent. The child's parent_brief_id is set to this brief's brief_id. """ return TaskBrief( # New identity brief_id=_new_uuid(), run_id=self.run_id, parent_brief_id=self.brief_id, # Tier / role tier=tier, role=role, # goal_anchor is ALWAYS copied unchanged from parent goal_anchor=self.goal_anchor, # Work specifics supplied by caller workstream=workstream or self.workstream, task=task, acceptance_criteria=acceptance_criteria or [], constraints=constraints or list(self.constraints), context=context or {}, # Runtime retry_budget=retry_budget, retry_count=0, preferred_runtime=preferred_runtime, agent_personality=agent_personality, ) # ------------------------------------------------------------------ # Repr # ------------------------------------------------------------------ def __repr__(self) -> str: return ( f"TaskBrief(brief_id={self.brief_id!r}, tier={self.tier}, " f"role={self.role!r}, workstream={self.workstream!r})" )