fix(intake): seed the task description into graph state (was silently dropped)

/new-task (and every intake: GitHub issue, /sh-assign-task) reached the clarifier
with NO description -> the clarifier asked 'no task description provided'. Root
cause: coordinator.start_task only LOGGED task_text (a P1-era decision when the
deterministic clarifier didn't consume a description), graph.start_task took no
task arg, and PipelineState/TaskRecord had no 'task' channel at all.

Fix: add a first-class 'task' field to PipelineState + TaskRecord (+ round-trip
in task_from_dict); graph.start_task seeds task into the initial invoke (persists
through intake_node's partial-state return into CLARIFY); coordinator.start_task
passes task=task_text. The clarifier already reads state['task'] via
_task_description, so it now sees the real description.

Test: start_task(task='build a login form') -> suspended CLARIFY state carries
task. 1143 passed, ruff clean.

agent-team/agent_team/coordinator.py

@@ -624,17 +624,15 @@ class Coordinator:
         :func:`agent_team.responder.notify_question` (ledger row OPEN first, then
         transport post). Returns the minted ``thread_id``.
 
-        **Intake-seed decision (P1).** ``agent_team.graph.start_task`` builds its
-        own INTAKE seed and accepts only ``thread_id`` / ``transport`` — it takes
-        no task-description argument, and a value pre-seeded onto the START
-        checkpoint via ``update_state`` is overwritten by its own seed invoke
-        (and a post-suspend ``update_state`` clears the pending interrupt, which
-        would break the human gate). So for P1 the ``task_text`` is intake
-        metadata held coordinator-side (logged) rather than written into
-        ``PipelineState``: the deterministic P1 clarifier does not consume a task
-        description anyway, and threading it into the graph state is a later phase
-        that extends the committed ``start_task`` seed contract. We keep it
-        minimal rather than reach past that contract or disturb the gate.
+        **Intake-seed (task description).** ``task_text`` is passed to
+        ``agent_team.graph.start_task`` as ``task=`` so it is written into the
+        seed ``PipelineState`` (the LLM clarifier reasons about it; an empty
+        description makes the clarifier ask for one). It is seeded directly into
+        the initial invoke — NOT via ``update_state`` (a pre-seed via
+        ``update_state`` would be overwritten by the seed invoke, and a
+        post-suspend ``update_state`` would clear the pending interrupt and break
+        the human gate). ``intake_node`` returns only a partial state
+        (status/phase), so the seeded ``task`` channel persists into CLARIFY.
         """
         if self._graph is None:
             raise RuntimeError("Coordinator.start_task called before setup()")
@@ -647,7 +645,9 @@ class Coordinator:
             transport_name,
             task_text[:200].replace("\n", "\\n").replace("\r", "\\r"),
         )
-        thread_id, _state = graph_mod.start_task(self._graph, transport=transport_name)
+        thread_id, _state = graph_mod.start_task(
+            self._graph, transport=transport_name, task=task_text
+        )
 
         question = graph_mod.pending_question(self._graph, thread_id=thread_id)
         if question is None:

agent-team/agent_team/graph.py

@@ -591,6 +591,7 @@ def start_task(
     *,
     thread_id: str | None = None,
     transport: str = "",
+    task: str = "",
 ) -> tuple[str, PipelineState]:
     """Start a new pipeline task and run it up to the first human gate (§3.3).
 
@@ -600,6 +601,13 @@ def start_task(
     the checkpointed snapshot after the suspend (its ``__interrupt__`` carries
     the pending question-set, surfaced by :func:`pending_question`).
 
+    ``task`` is the intake description (e.g. the Slack ``/new-task`` text or a
+    GitHub issue body). It is written into the seed ``PipelineState`` so the
+    clarifier can reason about it; ``intake_node`` returns only a partial state
+    (status/phase), so the seeded ``task`` channel persists into CLARIFY. An
+    empty ``task`` (the default) seeds no description — the clarifier then asks
+    for one.
+
     The graph MUST be compiled with a checkpointer for the suspend to persist;
     an uncheckpointed graph would run straight through without honouring the
     interrupt.
@@ -610,6 +618,7 @@ def start_task(
         thread_id=tid,
         status=TaskStatus.ACTIVE.value,
         current_phase=_phase_value(Phase.INTAKE),
+        task=task,
         qa_history=[],
         transport=transport,
         created_at=now,

agent-team/agent_team/task_model.py

@@ -87,6 +87,8 @@ class TaskRecord:
     thread_id: str
     status: TaskStatus
     current_phase: Phase
+    # Intake task description (mirrors PipelineState.task).
+    task: str = ""
     qa_history: list[Any] = field(default_factory=list)
     plan: dict[str, Any] | None = None
     review_verdicts: list[Any] = field(default_factory=list)
@@ -108,6 +110,10 @@ class PipelineState(TypedDict, total=False):
     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
     qa_history: list[Any]
     plan: dict[str, Any] | None
     review_verdicts: list[Any]
@@ -133,6 +139,7 @@ def task_from_dict(data: dict[str, Any]) -> TaskRecord:
         thread_id=data["thread_id"],
         status=TaskStatus(data["status"]),
         current_phase=Phase(data["current_phase"]),
+        task=data.get("task", ""),
         qa_history=list(data.get("qa_history", [])),
         plan=data.get("plan"),
         review_verdicts=list(data.get("review_verdicts", [])),

agent-team/tests/test_graph.py

@@ -136,6 +136,17 @@ def test_start_task_suspends_on_human_gate(compiled) -> None:
     assert payload["deadline"]
 
 
+def test_start_task_seeds_task_description_into_state(compiled) -> None:
+    # Regression: the intake description (Slack /new-task text, GitHub issue body)
+    # must reach the graph state so the clarifier can reason about it. It is
+    # seeded into the initial invoke and must persist through INTAKE into the
+    # suspended CLARIFY snapshot (intake_node returns only a partial state).
+    _thread_id, state = start_task(
+        compiled, transport="slack", task="build a login form"
+    )
+    assert state.get("task") == "build a login form"
+
+
 def test_pending_question_carries_foundation_questionset(compiled) -> None:
     thread_id, _ = start_task(compiled, transport="slack")
     payload = pending_question(compiled, thread_id=thread_id)