orchestrator/agent-team/agent_team/task_model.py

"""Task-record / thread model + LangGraph graph-state schema (design §3.3).

A task is a long-lived, resumable record (a LangGraph thread). This module is
the pure model layer — no I/O — defining:

* :class:`TaskStatus` / :class:`Phase` — task lifecycle enums.
* :class:`TaskRecord` — the durable task record (§3.3 "the task record holds:
  status, current phase, the full Q&A history, the plan, review verdicts, the
  candidate diff, and CI results").
* :class:`PipelineState` — a ``TypedDict`` used as the LangGraph graph state
  schema; its keys mirror :class:`TaskRecord` fields.
* :func:`new_thread_id` — uuid thread-id minting.
* JSON serialization helpers (:func:`task_to_dict` / :func:`task_from_dict` /
  :func:`task_to_json` / :func:`task_from_json`).

The signatures here are CONTRACTS leaf builders import verbatim.
"""

from __future__ import annotations

import json
import uuid
from dataclasses import asdict, dataclass, field
from enum import Enum
from typing import Any, TypedDict

__all__ = [
    "Phase",
    "PipelineState",
    "TaskRecord",
    "TaskStatus",
    "new_thread_id",
    "task_from_dict",
    "task_from_json",
    "task_to_dict",
    "task_to_json",
]


class TaskStatus(Enum):
    """Top-level task lifecycle status.

    ``ACTIVE`` — progressing through stages. ``WAITING_HUMAN`` — suspended on a
    LangGraph ``interrupt()`` awaiting Adam's answer. ``PARKED`` — stalled
    (no answer in window, N failed build loops, or budget contention) and
    ALARM-ed rather than spinning (§3.3, §6.6). ``DONE`` — draft PR + report
    produced. ``FAILED`` — terminal failure.
    """

    ACTIVE = "active"
    WAITING_HUMAN = "waiting_human"
    PARKED = "parked"
    DONE = "done"
    FAILED = "failed"


class Phase(Enum):
    """Pipeline phase the task is currently in (§3.3)."""

    INTAKE = "intake"
    CLARIFY = "clarify"
    PLAN = "plan"
    REVIEW = "review"
    BUILD = "build"
    VERIFY = "verify"
    # Direct-Confluence task lane (task_kind == "confluence"): draft the page
    # change, gate it for owner approve / request_changes, then write it. The
    # terminals reuse the shared DONE / PARKED phases.
    CONF_DRAFT = "conf_draft"
    CONF_GATE = "conf_gate"
    CONF_WRITE = "conf_write"
    PARKED = "parked"
    DONE = "done"


def new_thread_id() -> str:
    """Mint a fresh unique ``thread_id`` (uuid4 hex)."""
    return uuid.uuid4().hex


@dataclass
class TaskRecord:
    """The durable per-task record (§3.3, §3.3.1).

    Mirrors the LangGraph thread state; the SQLite checkpointer persists the
    graph state while this record is the logical view the coordinator reasons
    over. ``qa_history`` is the full clarifier Q&A; ``review_verdicts`` the
    adversarial review outcomes; ``candidate_diff`` + ``diff_hash`` the builder
    output and its ledger-recorded hash (§3.3.2); ``ci_results`` the
    authenticated CI conclusion the verifier reads.
    """

    thread_id: str
    status: TaskStatus
    current_phase: Phase
    # Intake task description (mirrors PipelineState.task).
    task: str = ""
    # Slack root-message ts for one-thread-per-task (mirrors
    # PipelineState.slack_thread_ts). Empty for non-/new-task origins.
    slack_thread_ts: str = ""
    qa_history: list[Any] = field(default_factory=list)
    plan: dict[str, Any] | None = None
    review_verdicts: list[Any] = field(default_factory=list)
    candidate_diff: str | None = None
    diff_hash: str | None = None
    ci_results: dict[str, Any] | None = None
    # P3 box-side build->dispatch->verify plumbing (§3.3.2). Set by the DISPATCH
    # node when it triggers the apply/verify CI run: ``run_id`` is the GitHub
    # Actions run id the verifier's read-only fetcher polls + the pure-code gate
    # binds its verdict to; ``ci_correlation_tag`` is the per-dispatch nonce
    # carried as a workflow input so the run_id poll matches THIS task's exact
    # run (anti-race / anti-replay); ``dispatched_at`` bounds the CI-watch
    # timeout. All None until a task reaches DISPATCH on the live P3 path.
    run_id: str | None = None
    ci_correlation_tag: str | None = None
    dispatched_at: str | None = None
    # Count of build<->verify loops already consumed for THIS task (§3.3 #6).
    # Durable per-task state (NOT the shared wiring-time VerifierConfig): the
    # verifier reads it from state, increments on a recoverable gate FAIL, and
    # parks once it reaches VerifierConfig.max_build_loops so a perpetually-
    # failing task can never loop BUILD->DISPATCH->VERIFY forever (LOGIC-RACE-01).
    build_loops: int = 0
    # Count of plan-review human-decision gate visits already consumed for THIS
    # task (Phase B2a). The graph routes a review-cap dead-end to the plan gate,
    # which interrupts for an owner approve / request_changes / abandon decision.
    # A human ``request_changes`` re-enters plan<->review, which can hit the cap
    # and gate AGAIN; this counter is the combined ceiling on human-driven gate
    # loops (mirrors PipelineState.plan_gate_visits) so the loop always
    # terminates: once it reaches ``MAX_PLAN_GATE_VISITS`` the gate stops
    # offering request_changes and the task goes terminal PARKED.
    plan_gate_visits: int = 0
    transport: str = ""
    created_at: str | None = None
    updated_at: str | None = None
    # Set when a pipeline node raised during resume and the coordinator failed
    # the task (mirrors PipelineState.failure_reason). Empty on a healthy task.
    failure_reason: str = ""
    # Direct-Confluence lane state (mirrors the PipelineState keys below). All
    # defaulted so an existing record (no Confluence lane) round-trips unchanged.
    # ``task_kind`` "" keeps the default clarify->plan->review path; "confluence"
    # routes through CONF_DRAFT->CONF_GATE->CONF_WRITE. ``confluence_draft`` is the
    # drafted page change awaiting the gate; ``confluence_feedback`` the owner's
    # request_changes notes the redraft reads; ``confluence_gate_visits`` the
    # ceiling on draft<->gate revision loops (mirrors plan_gate_visits);
    # ``confluence_result`` the written-page outcome (id, version, url).
    task_kind: str = ""
    confluence_draft: dict[str, Any] | None = None
    confluence_feedback: str = ""
    confluence_gate_visits: int = 0
    confluence_result: dict[str, Any] | None = None


class PipelineState(TypedDict, total=False):
    """LangGraph graph-state schema; keys mirror :class:`TaskRecord` (§3.3).

    Used as the graph's state type. ``total=False`` so a node may write a
    subset of keys per checkpoint transition.
    """

    thread_id: str
    status: str
    current_phase: str
    # The intake task description (Slack /new-task text, GitHub issue body, etc.).
    # Seeded by graph.start_task and read by the clarifier/planner; a first-class
    # channel so the seeded value persists across node transitions.
    task: str
    # The Slack root-message ``ts`` for a /new-task task (the "📥 Task received"
    # ack post). All of the task's clarifier questions and lifecycle milestone
    # notifications thread under this ``ts`` so one task maps to one Slack thread.
    # Empty/absent for a task that did not originate from /new-task (no root post),
    # in which case posts are top-level exactly as before.
    slack_thread_ts: str
    qa_history: list[Any]
    plan: dict[str, Any] | None
    review_verdicts: list[Any]
    candidate_diff: str | None
    diff_hash: str | None
    ci_results: dict[str, Any] | None
    # P3 box-side build->dispatch->verify plumbing (mirrors TaskRecord). ``run_id``
    # is the dispatched apply/verify Actions run id the verifier fetches + the
    # gate binds to; ``ci_correlation_tag`` is the per-dispatch nonce carried as a
    # workflow input so the poll matches THIS task's run; ``dispatched_at`` bounds
    # the CI-watch timeout.
    run_id: str | None
    ci_correlation_tag: str | None
    dispatched_at: str | None
    # Build<->verify loops already consumed for THIS task (mirrors TaskRecord).
    # The verifier threads it through DURABLE state — increments on a recoverable
    # gate FAIL, parks at VerifierConfig.max_build_loops — so the build-loop
    # budget is real (LOGIC-RACE-01: it was previously read from the shared
    # wiring-time config and never advanced).
    build_loops: int
    # Plan-review human-decision gate visits consumed for THIS task (mirrors
    # TaskRecord.plan_gate_visits; Phase B2a). The combined ceiling on
    # human-driven plan<->review loops: once it reaches MAX_PLAN_GATE_VISITS the
    # gate stops offering request_changes and the task goes terminal PARKED, so
    # the human-in-the-loop revision cycle can never spin forever.
    plan_gate_visits: int
    transport: str
    created_at: str | None
    updated_at: str | None
    # Set when the coordinator terminally fails a task because a pipeline node
    # raised during resume (see ``Coordinator._fail_resumed_task``). Carries a
    # short "ExcType: message" so the failure notification can say what broke.
    # Absent on a healthy task.
    failure_reason: str
    # Intake discriminator routing a task into a non-default lane. "" (default,
    # absent) keeps the existing clarify->plan->review path; "confluence" routes
    # a direct Confluence-documentation task through CONF_DRAFT->CONF_GATE->
    # CONF_WRITE.
    task_kind: str
    # The drafted Confluence page change awaiting the gate (title, body, space,
    # parent, target page id, etc.). None until CONF_DRAFT produces it.
    confluence_draft: dict | None
    # Free-text owner feedback captured at CONF_GATE on a request_changes
    # decision; CONF_DRAFT reads it to revise the draft. Empty when none.
    confluence_feedback: str
    # Count of CONF_GATE human-decision visits consumed for THIS task. The
    # combined ceiling on confluence draft<->gate revision loops so a
    # perpetually-revised page can never spin forever (mirrors plan_gate_visits).
    confluence_gate_visits: int
    # The result of the CONF_WRITE Confluence API call (page id, version, url).
    # None until the page has been written.
    confluence_result: dict | None


def task_to_dict(record: TaskRecord) -> dict[str, Any]:
    """Serialize a :class:`TaskRecord` to a JSON-safe dict (enums -> values)."""
    data = asdict(record)
    data["status"] = record.status.value
    data["current_phase"] = record.current_phase.value
    return data


def task_from_dict(data: dict[str, Any]) -> TaskRecord:
    """Rebuild a :class:`TaskRecord` from a :func:`task_to_dict` dict."""
    return TaskRecord(
        thread_id=data["thread_id"],
        status=TaskStatus(data["status"]),
        current_phase=Phase(data["current_phase"]),
        task=data.get("task", ""),
        slack_thread_ts=data.get("slack_thread_ts", ""),
        qa_history=list(data.get("qa_history", [])),
        plan=data.get("plan"),
        review_verdicts=list(data.get("review_verdicts", [])),
        candidate_diff=data.get("candidate_diff"),
        diff_hash=data.get("diff_hash"),
        ci_results=data.get("ci_results"),
        run_id=data.get("run_id"),
        ci_correlation_tag=data.get("ci_correlation_tag"),
        dispatched_at=data.get("dispatched_at"),
        build_loops=data.get("build_loops", 0),
        plan_gate_visits=data.get("plan_gate_visits", 0),
        transport=data.get("transport", ""),
        created_at=data.get("created_at"),
        updated_at=data.get("updated_at"),
        failure_reason=data.get("failure_reason", ""),
        task_kind=data.get("task_kind", ""),
        confluence_draft=data.get("confluence_draft"),
        confluence_feedback=data.get("confluence_feedback", ""),
        confluence_gate_visits=data.get("confluence_gate_visits", 0),
        confluence_result=data.get("confluence_result"),
    )


def task_to_json(record: TaskRecord) -> str:
    """Serialize a :class:`TaskRecord` to a JSON string."""
    return json.dumps(task_to_dict(record), sort_keys=True)


def task_from_json(payload: str | bytes) -> TaskRecord:
    """Deserialize a :class:`TaskRecord` from a JSON string/bytes."""
    return task_from_dict(json.loads(payload))