feat(notify): richer park/lifecycle messages — task description + phase + blocker

Park notifications were opaque ('Task 6c3c3202 parked — needs your attention'):
no idea what the task is, where it got to, or what's blocking it. Now each
message names the task DESCRIPTION (not just the short id), the PHASE it reached,
and the actual BLOCKER — _summarize_blocker() pulls the last review_verdict's
findings (the GPT-4.1 REQUEST_CHANGES text, collapsed + truncated), falling back
to 'no plan built' / 'revision cap hit'. Applies to parked + needs-more-input +
plan-ready emits. +1 test (description + phase + blocker present). 1148 passed.

agent-team/agent_team/coordinator.py

@@ -779,6 +779,19 @@ class Coordinator:
                 continue
             seen.add(thread_id)
             short = thread_id[:8]
+
+            # Read the settled state once so every message can name WHAT the task
+            # is (description), not just an opaque thread id.
+            try:
+                snap = self._graph.get_state(graph_mod.thread_config(thread_id))
+                values = getattr(snap, "values", {}) or {}
+            except Exception:  # noqa: BLE001
+                values = {}
+            desc = str(values.get("task") or "").strip() or "(no description)"
+            if len(desc) > 90:
+                desc = desc[:90] + "…"
+            label = f'"{desc}" (`{short}`)'
+
             try:
                 question = graph_mod.pending_question(self._graph, thread_id=thread_id)
             except Exception:  # noqa: BLE001 - never let a status check break the loop
@@ -804,23 +817,54 @@ class Coordinator:
                     _LOG.warning(
                         "failed to post follow-up question for %s", short, exc_info=True
                     )
-                self._emit(f"❓ Task {short}: needs more input — posted a question.")
+                self._emit(
+                    f"❓ {label} — needs more input. A new clarifying question was "
+                    "posted above; reply in its thread."
+                )
                 continue
 
-            # No pending question: the task settled. Distinguish parked vs done.
-            try:
-                snap = self._graph.get_state(graph_mod.thread_config(thread_id))
-                values = getattr(snap, "values", {}) or {}
-                status = values.get("status")
-            except Exception:  # noqa: BLE001
-                status = None
+            # No pending question: the task settled. Distinguish parked vs done,
+            # and say WHERE it got to and WHAT is blocking it.
+            status = values.get("status")
+            phase = str(values.get("current_phase") or "unknown")
             if status == TaskStatus.PARKED.value:
+                blocker = self._summarize_blocker(values)
                 self._emit(
-                    f"⚠️ Task {short}: parked — needs your attention "
-                    "(plan/review escalation). Re-assign or steer it to resume."
+                    f"⚠️ PARKED — {label}\n"
+                    f"• Reached phase: {phase}\n"
+                    f"• What's blocking it: {blocker}\n"
+                    "• Needs your review — the plan could not be auto-approved. "
+                    "Re-assign with more detail, or adjust the requirement to unblock."
                 )
             else:
-                self._emit(f"✅ Task {short}: plan ready for review.")
+                self._emit(f"✅ {label} — plan ready for review (phase: {phase}).")
+
+    @staticmethod
+    def _summarize_blocker(values: "dict[str, Any]") -> str:
+        """Human-readable reason a task parked, from the last review verdict.
+
+        Pulls the most recent ``review_verdicts`` entry's findings (the GPT-4.1
+        REQUEST_CHANGES text) — collapsed + truncated for a Slack line — so the
+        human sees WHAT stopped it, not just "escalation". Falls back to a generic
+        reason when there is no verdict (e.g. an unparseable-plan park before
+        review ever ran).
+        """
+        verdicts = values.get("review_verdicts") or []
+        if verdicts:
+            last = verdicts[-1]
+            if isinstance(last, dict):
+                findings = str(
+                    last.get("findings") or last.get("verdict") or ""
+                ).strip()
+                if findings:
+                    summary = " ".join(findings.split())
+                    return summary[:300] + ("…" if len(summary) > 300 else "")
+        if not values.get("plan"):
+            return "the planner could not produce a usable plan (no plan was built)."
+        return (
+            "the plan→review loop hit its revision cap (the reviewer kept "
+            "requesting changes without converging)."
+        )
 
     def tick(self) -> list[ResumeResult]:
         """One maintenance pass: deadline sweep + park policy, then drain (§3.3.1).

agent-team/tests/test_coordinator.py

@@ -1010,3 +1010,40 @@ def test_emit_swallows_notify_failure(db_path: Path) -> None:
     coord = _make_coordinator(db_path)
     coord._notify = _boom
     coord._emit("anything")  # must not raise
+
+
+def test_parked_message_includes_description_and_blocker(
+    db_path: Path, monkeypatch: Any
+) -> None:
+    from agent_team import coordinator as coord_mod
+    from agent_team.task_model import TaskStatus
+
+    msgs: list[str] = []
+    coord = _make_coordinator(db_path)
+    coord._notify = msgs.append
+    coord.setup()
+
+    monkeypatch.setattr(
+        coord_mod.graph_mod, "pending_question", lambda _g, *, thread_id: None
+    )
+
+    class _Snap:
+        values = {
+            "status": TaskStatus.PARKED.value,
+            "current_phase": "review",
+            "task": "add a smoke-test file in agent-team/tests",
+            "review_verdicts": [
+                {
+                    "verdict": "request_changes",
+                    "findings": "Missing a final review/commit phase; lint runs before tests.",
+                }
+            ],
+        }
+
+    monkeypatch.setattr(coord._graph, "get_state", lambda _cfg: _Snap())
+    coord._post_resume_followups([_resume_result("c0ffee01abcd")])
+    assert len(msgs) == 1
+    m = msgs[0]
+    assert "add a smoke-test file" in m  # WHAT the task is
+    assert "review" in m  # WHERE it got to
+    assert "Missing a final review/commit phase" in m  # WHY it's blocked