214 lines
7.8 KiB
Python
214 lines
7.8 KiB
Python
"""
|
||
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})"
|
||
)
|