feat: initial bootstrap — structure, task_brief, blackboard, adapter bases, escalation, prompts

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

core/task_brief.py

@@ -0,0 +1,213 @@
+"""
+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})"
+        )