feat(agent-team): Slack decision surface for the plan-review gate (Phase B3)

Turn a human's Slack interaction at the plan gate into a structured decision the
graph can route, with the kind-aware mapping that closes a silent-FAIL hazard.

- KIND-AWARE NORMALIZATION (load-bearing): map_plan_decision() in slack_adapter
  maps a reply to {"decision","notes"} — approve ∈ {approve,approved,yes,ok,lgtm,
  ship}; abandon ∈ {abandon,reject,cancel,stop,kill}; EVERYTHING ELSE →
  request_changes with the full reply as notes (never accidental abandon). Wired
  in the listener's _resolve_payload for plan_decision rows ONLY (clarify passes
  through). Without this, arbitrary change-notes hit the graph's
  unrecognized-verb→FAILED path and silently fail the task. Anti-FAIL tests
  assert prose → request_changes (!= abandon) at both the mapper and the
  end-to-end listener seam; a regression test guards clarify pass-through.
  New find_open_question_kind_by_channel_ref (anti-replay, status='open') powers
  the thread-reply fallback's kind lookup.
- BUTTONS + MODAL: build_plan_decision_blocks() renders Approve (primary) /
  Request changes / Abandon (danger+confirm); question_id double-anchored in
  message metadata AND each button value ("<verb>:<question_id>"). Approve/abandon
  submit via the existing @app.action(.*); request_changes has a dedicated
  handler that AUTHORIZES before views_open (proven by test) and opens a notes
  modal (private_metadata carries the id) → view_submission → request_changes +
  notes. Free-text reply stays the always-available equal path. AUTHZ-01 ordering
  preserved.
- No manifest change (views.open needs no extra scope).

1454 passed (1412 + 42).

agent-team/agent_team/coordinator.py

@@ -1263,7 +1263,11 @@ class Coordinator:
         transport_name = str(question.get("transport") or "")
 
         # 1. Durable ledger row first (open, kind='plan_decision'), guarded by the
-        #    single-open-gate invariant.
+        #    single-open-gate invariant. The row's gate post (via the transport's
+        #    post_question) carries the three decision BUTTONS over the same
+        #    presentation body the lifecycle milestone shows (B3), so the buttons
+        #    land on the decision message itself.
+        body = self._plan_decision_presentation(question, label=label)
         try:
             conn = connect(self._db_path)
             try:
@@ -1275,6 +1279,7 @@ class Coordinator:
                     transport_name=transport_name,
                     deadline=deadline,
                     root_ts=root_ts,
+                    presentation=body,
                 )
             finally:
                 conn.close()
@@ -1290,9 +1295,11 @@ class Coordinator:
             # opener; just stop here.
             return
 
-        # 2. Present the plan + findings + decision instructions, threaded under
+        # 2. Emit the human-readable presentation as a lifecycle milestone too
-        #    the task root.
+        #    (B2b), threaded under the task root, so the decision is visible to
-        body = self._plan_decision_presentation(question, label=label)
+        #    sinks that don't render Block Kit (and to the dashboard/notify
+        #    trail). The buttons live on the transport gate post above; this is
+        #    the always-available text path.
         self._emit(body, thread_ts=root_ts)
 
     def _plan_decision_presentation(
@@ -1315,8 +1322,8 @@ class Coordinator:
             findings = findings[:600] + "…"
 
         instructions = (
-            "• Decide: reply *approve* / *request changes <notes>* / *abandon* "
+            "• Decide: use the buttons below, OR reply *approve* / "
-            "in this thread."
+            "*request changes <notes>* / *abandon* in this thread."
         )
         header = f"🧭 {label} — plan needs your decision (review could not approve it)."
 
@@ -1348,6 +1355,7 @@ class Coordinator:
         transport_name: str,
         deadline: str,
         root_ts: str | None,
+        presentation: str = "",
     ) -> bool:
         """Open a ``kind='plan_decision'`` ledger row, enforcing one-open-gate (B-2).
 
@@ -1400,7 +1408,10 @@ class Coordinator:
                 question_id=question_id,
                 turn=turn,
                 question_set=self._plan_decision_question_set(
-                    thread_id=thread_id, question_id=question_id, turn=turn
+                    thread_id=thread_id,
+                    question_id=question_id,
+                    turn=turn,
+                    presentation=presentation,
                 ),
                 deadline=deadline,
                 **post_kwargs,
@@ -1423,15 +1434,17 @@ class Coordinator:
 
     @staticmethod
     def _plan_decision_question_set(
-        *, thread_id: str, question_id: str, turn: int
+        *, thread_id: str, question_id: str, turn: int, presentation: str = ""
     ) -> "QuestionSet":
-        """Build a minimal QuestionSet for the gate post (answer-mapping only).
+        """Build the QuestionSet for the gate post (decision surface, B3).
 
         The transport's ``post_question`` requires a ``question_set`` so the
-        inbound answer can map back to ``question_id``; the gate's human-readable
+        inbound answer can map back to ``question_id``. The ``context`` carries
-        presentation is posted via the lifecycle ``_emit`` sink, so this set only
+        ``kind == 'plan_decision'`` (so the Slack transport renders the three
-        needs to carry the identity (the decision-instruction wording lives in the
+        decision buttons instead of generic question blocks) and the human-
-        presentation message). The single question text is a terse decision
+        readable ``presentation`` body (so the buttons render over the plan +
+        findings + instructions, and the message text fallback shows the same to
+        non-interactive clients). The single question text is a terse decision
         prompt for transports that render the set directly.
         """
         from agent_team.transport.base import QuestionSet  # noqa: PLC0415
@@ -1441,7 +1454,10 @@ class Coordinator:
             question_id=question_id,
             turn=turn,
             questions=["Approve, request changes, or abandon this plan?"],
-            context={"kind": graph_mod.PLAN_DECISION_KIND},
+            context={
+                "kind": graph_mod.PLAN_DECISION_KIND,
+                "presentation": presentation,
+            },
         )
 
     @staticmethod

agent-team/agent_team/db/schema.py

@@ -43,6 +43,7 @@ __all__ = [
     "delete_issue_ingested",
     "expire_question",
     "find_open_question_by_channel_ref",
+    "find_open_question_kind_by_channel_ref",
     "init_db",
     "issue_already_ingested",
     "migrate",
@@ -579,6 +580,32 @@ def find_open_question_by_channel_ref(
     return None if row is None else str(row["question_id"])
 
 
+def find_open_question_kind_by_channel_ref(
+    conn: sqlite3.Connection,
+    channel_ref: str,
+) -> tuple[str, str] | None:
+    """Map a transport ``channel_ref`` to its open ``(question_id, kind)``.
+
+    Like :func:`find_open_question_by_channel_ref` but also returns the ``kind``
+    column so the caller can route ``plan_decision`` rows through the
+    kind-aware decision normalizer without a second query.
+
+    Returns ``(question_id, kind)`` for the matching open row, or ``None`` if
+    ``channel_ref`` is empty or matches no open row. The lookup is constrained
+    to ``status='open'`` (anti-replay) — same semantics as the parent function.
+    """
+    if not channel_ref:
+        return None
+    row = conn.execute(
+        "SELECT question_id, kind FROM pending_questions "
+        "WHERE channel_ref=? AND status='open'",
+        (channel_ref,),
+    ).fetchone()
+    if row is None:
+        return None
+    return str(row["question_id"]), str(row["kind"])
+
+
 def supersede_question(
     conn: sqlite3.Connection,
     *,

agent-team/agent_team/transport/slack_adapter.py

@@ -35,12 +35,21 @@ from agent_team.transport.base import (
 
 __all__ = [
     "CALLBACK_ID_PREFIX",
+    "PLAN_DECISION_ABANDON_ACTION",
+    "PLAN_DECISION_ABANDON_VERBS",
+    "PLAN_DECISION_APPROVE_ACTION",
+    "PLAN_DECISION_APPROVE_VERBS",
+    "PLAN_DECISION_KIND",
+    "PLAN_DECISION_REQUEST_CHANGES_ACTION",
     "VIA_SLACK",
     "SlackPostError",
     "SlackPoster",
     "SlackTransport",
     "build_callback_id",
+    "build_plan_decision_blocks",
     "build_question_blocks",
+    "build_request_changes_modal",
+    "map_plan_decision",
     "parse_callback_id",
 ]
 
@@ -54,6 +63,67 @@ VIA_SLACK = "slack"
 # question) from the base module.
 CALLBACK_ID_PREFIX = "shq"
 
+# The ``pending_questions.kind`` discriminator for the plan-review decision gate.
+# Mirrors ``agent_team.graph.PLAN_DECISION_KIND`` and the DB CHECK constraint;
+# duplicated here (as a plain string constant) so the transport/listener layer
+# can route kind-aware decisions WITHOUT importing the graph module.
+PLAN_DECISION_KIND = "plan_decision"
+
+# Plan-decision verb allowlists (B3). A reply / button-value to a
+# ``kind == 'plan_decision'`` question is normalized (lowercase + strip) and
+# matched against these allowlists. Anything ELSE — arbitrary change-request
+# prose — maps to ``request_changes`` carrying the FULL original reply as
+# ``notes`` (the safe default; NEVER an accidental approve or abandon). See
+# :func:`map_plan_decision`.
+PLAN_DECISION_APPROVE_VERBS: frozenset[str] = frozenset(
+    {"approve", "approved", "yes", "ok", "lgtm", "ship"}
+)
+PLAN_DECISION_ABANDON_VERBS: frozenset[str] = frozenset(
+    {"abandon", "reject", "cancel", "stop", "kill"}
+)
+
+# Block Kit action_id namespace for the three plan-gate buttons. Each carries
+# the decision verb; ``request_changes`` opens a modal for free-form notes.
+PLAN_DECISION_ACTION_PREFIX = "plan_decision"
+PLAN_DECISION_APPROVE_ACTION = f"{PLAN_DECISION_ACTION_PREFIX}:approve"
+PLAN_DECISION_REQUEST_CHANGES_ACTION = f"{PLAN_DECISION_ACTION_PREFIX}:request_changes"
+PLAN_DECISION_ABANDON_ACTION = f"{PLAN_DECISION_ACTION_PREFIX}:abandon"
+
+
+def map_plan_decision(raw_answer: Any) -> dict[str, str]:
+    """Map a raw plan-gate reply to a structured ``{"decision", "notes"}`` dict.
+
+    THE LOAD-BEARING B3 SAFETY MAPPING. The graph's ``_parse_decision`` maps any
+    unrecognized verb to ``abandon`` → terminal FAILED, so free-text change
+    notes (e.g. "use pytest fixtures instead") would silently FAIL the task if
+    they reached the graph unmapped. This normalizes BEFORE the graph sees it:
+
+    * normalize the text (``str`` → lowercase → strip);
+    * **approve** iff in :data:`PLAN_DECISION_APPROVE_VERBS`
+      (``approve/approved/yes/ok/lgtm/ship``), with ``notes=""``;
+    * **abandon** iff in :data:`PLAN_DECISION_ABANDON_VERBS`
+      (``abandon/reject/cancel/stop/kill``), with ``notes=""``;
+    * **everything else → ``request_changes`` with the FULL original reply as
+      ``notes``** — the safe default. Arbitrary prose is a change request, never
+      an accidental approve or abandon. Empty / whitespace-only input also maps
+      to ``request_changes`` (with empty notes): a blank reply is treated as a
+      benign no-op change request, never a destructive abandon.
+
+    The answer is opaque DATA throughout — never executed or interpreted beyond
+    this verb match. Returns the dict the graph's ``_parse_decision`` consumes.
+    """
+    original = "" if raw_answer is None else str(raw_answer)
+    verb = original.strip().lower()
+    if verb in PLAN_DECISION_APPROVE_VERBS:
+        return {"decision": "approve", "notes": ""}
+    if verb in PLAN_DECISION_ABANDON_VERBS:
+        return {"decision": "abandon", "notes": ""}
+    # Everything else (including empty/whitespace) → request_changes, carrying
+    # the full ORIGINAL reply (not the lowercased form) so the notes preserve
+    # the human's exact wording. Never an accidental abandon.
+    return {"decision": "request_changes", "notes": original.strip()}
+
+
 # Type of the network seam: given the rendered Slack message kwargs, perform the
 # ``chat.postMessage`` and return the response payload. The only field this
 # adapter requires from the response is the message ``ts`` (the ``channel_ref``).
@@ -163,6 +233,121 @@ def build_question_blocks(
     return blocks
 
 
+def build_plan_decision_blocks(
+    question_id: str, body_text: str
+) -> list[dict[str, Any]]:
+    """Render the plan-gate decision surface as Block Kit blocks (B3).
+
+    A section carrying the human-readable ``body_text`` (the plan summary +
+    review findings + decision instructions assembled by the coordinator) plus
+    an ``actions`` block with the three decision buttons:
+
+    * **Approve** (``action_id="plan_decision:approve"``, ``style="primary"``);
+    * **Request changes** (``action_id="plan_decision:request_changes"``) — its
+      dedicated handler opens a modal for free-form notes;
+    * **Abandon** (``action_id="plan_decision:abandon"``, ``style="danger"``,
+      guarded by a ``confirm`` dialog so it is never a single-click mistake).
+
+    Each button's ``value`` encodes ``"<verb>:<question_id>"`` so an inbound
+    ``block_actions`` payload can recover BOTH the verb and the ledger
+    ``question_id`` even if the message metadata is absent. The ``question_id``
+    is ALSO carried in the message ``metadata.event_payload`` by the caller
+    (mirroring the clarifier), so recovery is double-anchored. The section text
+    is the always-visible fallback for non-interactive clients (a free-text
+    thread reply remains an equal path to all three decisions).
+
+    Returns a plain JSON-serializable list (no Slack SDK types).
+    """
+    section_text = body_text if body_text else "Plan needs your decision."
+    # Slack section text caps at ~3000 chars; keep a margin.
+    if len(section_text) > 2900:
+        section_text = section_text[:2900].rstrip() + "…"
+    return [
+        {"type": "section", "text": {"type": "mrkdwn", "text": section_text}},
+        {
+            "type": "actions",
+            "elements": [
+                {
+                    "type": "button",
+                    "action_id": PLAN_DECISION_APPROVE_ACTION,
+                    "text": {"type": "plain_text", "text": "Approve"},
+                    "style": "primary",
+                    "value": f"approve:{question_id}",
+                },
+                {
+                    "type": "button",
+                    "action_id": PLAN_DECISION_REQUEST_CHANGES_ACTION,
+                    "text": {"type": "plain_text", "text": "Request changes"},
+                    "value": f"request_changes:{question_id}",
+                },
+                {
+                    "type": "button",
+                    "action_id": PLAN_DECISION_ABANDON_ACTION,
+                    "text": {"type": "plain_text", "text": "Abandon"},
+                    "style": "danger",
+                    "value": f"abandon:{question_id}",
+                    "confirm": {
+                        "title": {"type": "plain_text", "text": "Abandon this task?"},
+                        "text": {
+                            "type": "mrkdwn",
+                            "text": (
+                                "This terminally FAILS the task. The plan is "
+                                "discarded and the pipeline stops."
+                            ),
+                        },
+                        "confirm": {"type": "plain_text", "text": "Abandon"},
+                        "deny": {"type": "plain_text", "text": "Keep"},
+                        "style": "danger",
+                    },
+                },
+            ],
+        },
+    ]
+
+
+# Block / action ids for the request-changes modal input, so the inbound
+# ``view_submission`` extraction can find the notes value deterministically.
+REQUEST_CHANGES_MODAL_CALLBACK_ID = (
+    f"{PLAN_DECISION_ACTION_PREFIX}:request_changes_modal"
+)
+REQUEST_CHANGES_NOTES_BLOCK_ID = "plan_decision_notes_block"
+REQUEST_CHANGES_NOTES_ACTION_ID = "plan_decision_notes_input"
+
+
+def build_request_changes_modal(question_id: str) -> dict[str, Any]:
+    """Build the "Request changes" notes modal (views.open view, B3).
+
+    One required multiline ``plain_text_input`` ("What should change?"). The
+    ledger ``question_id`` round-trips through ``private_metadata`` as
+    ``"request_changes:<question_id>"`` so the eventual ``view_submission``
+    recovers it (mirroring the button-value encoding). On submit, the input text
+    becomes the ``request_changes`` notes via the kind-aware normalizer.
+
+    Returns a plain JSON-serializable view dict (no Slack SDK types) so the
+    caller passes it straight to ``client.views_open(trigger_id=..., view=...)``.
+    """
+    return {
+        "type": "modal",
+        "callback_id": REQUEST_CHANGES_MODAL_CALLBACK_ID,
+        "private_metadata": f"request_changes:{question_id}",
+        "title": {"type": "plain_text", "text": "Request changes"},
+        "submit": {"type": "plain_text", "text": "Send"},
+        "close": {"type": "plain_text", "text": "Cancel"},
+        "blocks": [
+            {
+                "type": "input",
+                "block_id": REQUEST_CHANGES_NOTES_BLOCK_ID,
+                "label": {"type": "plain_text", "text": "What should change?"},
+                "element": {
+                    "type": "plain_text_input",
+                    "action_id": REQUEST_CHANGES_NOTES_ACTION_ID,
+                    "multiline": True,
+                },
+            }
+        ],
+    }
+
+
 def _join_answer_actions(actions: Sequence[Mapping[str, Any]]) -> Any:
     """Reduce one-or-more interactive actions to a single answer value.
 
@@ -256,14 +441,29 @@ class SlackTransport(Transport):
         durable ``channel_ref`` (it uses the root ``thread_ts`` when threading so
         an inbound reply's ``thread_ts`` maps back to this question).
         """
-        blocks = build_question_blocks(question_set, deadline)
+        # Plan-review gate (B3): when the question_set is a plan decision, render
+        # the three decision buttons (Approve / Request changes / Abandon) over
+        # the presentation text instead of the generic question blocks. The
+        # presentation body is passed through the question_set ``context`` (under
+        # ``presentation``) by the coordinator; the buttons carry the recoverable
+        # question_id (and the message metadata double-anchors it). A free-text
+        # thread reply remains an equal path to all three decisions.
+        if question_set.context.get("kind") == PLAN_DECISION_KIND:
+            body_text = str(question_set.context.get("presentation") or "")
+            blocks = build_plan_decision_blocks(question_id, body_text)
+            fallback_text = body_text or (
+                f"Plan decision needed on task {thread_id} (turn {turn})."
+            )
+        else:
+            blocks = build_question_blocks(question_set, deadline)
+            fallback_text = (
+                f"Agent-team needs input on task {thread_id} "
+                f"(turn {turn}); reply by {deadline}."
+            )
         message: dict[str, Any] = {
             "channel": self.channel,
             "callback_id": build_callback_id(question_id),
-            "text": (
+            "text": fallback_text,
-                f"Agent-team needs input on task {thread_id} "
-                f"(turn {turn}); reply by {deadline}."
-            ),
             "blocks": blocks,
             "metadata": {
                 "event_type": "agent_team_question",
@@ -343,8 +543,23 @@ def _extract_question_id(raw: Mapping[str, Any]) -> str:
 
     # Interactive payloads nest the callback metadata under ``view`` / ``message``.
     view = raw.get("view")
-    if isinstance(view, Mapping) and view.get("callback_id"):
+    if isinstance(view, Mapping):
-        return parse_callback_id(str(view["callback_id"]))
+        # Modal (view_submission) round-trips the question_id through
+        # ``private_metadata`` (B3 request-changes modal). It carries
+        # ``"<verb>:<question_id>"`` (or a bare id); recover the id suffix. This
+        # is checked BEFORE the view ``callback_id`` because the B3 modal's
+        # callback_id is a modal identifier (``plan_decision:...``), NOT a
+        # ``shq:<question_id>`` carrier.
+        private_metadata = view.get("private_metadata")
+        if private_metadata:
+            return _question_id_from_value(str(private_metadata))
+        # A view whose callback_id IS an ``shq:`` carrier (a non-B3 modal that
+        # embedded the question id there directly) still resolves.
+        view_callback_id = view.get("callback_id")
+        if view_callback_id and str(view_callback_id).startswith(
+            f"{CALLBACK_ID_PREFIX}:"
+        ):
+            return parse_callback_id(str(view_callback_id))
     message = raw.get("message")
     if isinstance(message, Mapping):
         metadata = message.get("metadata")
@@ -353,6 +568,25 @@ def _extract_question_id(raw: Mapping[str, Any]) -> str:
             if isinstance(payload, Mapping) and payload.get("question_id"):
                 return str(payload["question_id"])
 
+    # Block-action button value: the B3 plan-gate buttons encode
+    # ``"<verb>:<question_id>"`` so the id is recoverable even with no
+    # callback_id / message metadata on the payload.
+    actions = raw.get("actions")
+    if (
+        isinstance(actions, Sequence)
+        and not isinstance(actions, (str, bytes))
+        and actions
+    ):
+        for action in actions:
+            if not isinstance(action, Mapping):
+                continue
+            value = action.get("value")
+            if not value:
+                continue
+            qid = _question_id_from_plan_decision_value(str(value))
+            if qid:
+                return qid
+
     question_id = raw.get("question_id")
     if question_id:
         return str(question_id)
@@ -360,17 +594,88 @@ def _extract_question_id(raw: Mapping[str, Any]) -> str:
     raise ValueError("Slack payload carries no recoverable question_id")
 
 
+_PLAN_DECISION_VALUE_VERBS = frozenset({"approve", "request_changes", "abandon"})
+
+
+def _question_id_from_value(value: str) -> str:
+    """Recover the question_id from a ``private_metadata`` string.
+
+    The B3 request-changes modal stores ``"<verb>:<question_id>"`` (or a bare
+    ``question_id``) in ``private_metadata``. A leading known decision verb is a
+    prefix to strip; otherwise the whole value IS the id. Raises
+    :class:`ValueError` on an empty value.
+    """
+    qid = _question_id_from_plan_decision_value(value)
+    if qid:
+        return qid
+    if not value:
+        raise ValueError("empty private_metadata; no recoverable question_id")
+    return value
+
+
+def _question_id_from_plan_decision_value(value: str) -> str | None:
+    """Return the ``question_id`` suffix of a ``"<verb>:<question_id>"`` value.
+
+    Only matches when the prefix is a known plan-decision verb so an ordinary
+    button value (e.g. a clarifier's free-text answer) is never mis-parsed.
+    Returns ``None`` if the value is not a ``"<verb>:<question_id>"`` encoding.
+    """
+    verb, sep, rest = value.partition(":")
+    if sep and verb in _PLAN_DECISION_VALUE_VERBS and rest:
+        return rest
+    return None
+
+
+def _modal_input_text(view: Mapping[str, Any]) -> Any:
+    """Recover the submitted text from a ``view_submission`` view (B3 modal).
+
+    Walks ``view.state.values`` (``{block_id: {action_id: {value: ...}}}``) and
+    returns the first non-empty ``plain_text_input`` value. The B3 request-
+    changes modal has a single input, so the first value is the notes text.
+    Returns ``None`` if no input value is present.
+    """
+    state = view.get("state")
+    if not isinstance(state, Mapping):
+        return None
+    values = state.get("values")
+    if not isinstance(values, Mapping):
+        return None
+    for block in values.values():
+        if not isinstance(block, Mapping):
+            continue
+        for action in block.values():
+            if isinstance(action, Mapping) and action.get("value") is not None:
+                return action["value"]
+    return None
+
+
 def _extract_answer(raw: Mapping[str, Any]) -> Any:
     """Recover the answer value from any supported inbound payload."""
+    # view_submission (modal): the answer is the submitted input text (B3).
+    view = raw.get("view")
+    if isinstance(view, Mapping):
+        text = _modal_input_text(view)
+        if text is not None:
+            return text
+
     actions = raw.get("actions")
     if (
         isinstance(actions, Sequence)
         and not isinstance(actions, (str, bytes))
         and actions
     ):
-        return _join_answer_actions(
+        mappings = [action for action in actions if isinstance(action, Mapping)]
-            [action for action in actions if isinstance(action, Mapping)]
+        # A single plan-decision button encodes ``"<verb>:<question_id>"``; the
-        )
+        # question_id is recovered separately (callback_id / metadata / value),
+        # so the ANSWER is the bare verb. Stripping the suffix here lets the
+        # kind-aware normalizer match it against the approve/abandon allowlists.
+        if len(mappings) == 1:
+            value = mappings[0].get("value")
+            if value is not None:
+                verb, sep, rest = str(value).partition(":")
+                if sep and verb in _PLAN_DECISION_VALUE_VERBS and rest:
+                    return verb
+        return _join_answer_actions(mappings)
 
     if "answer" in raw:
         return raw["answer"]

agent-team/agent_team/transport/slack_listener.py

@@ -74,15 +74,26 @@ from __future__ import annotations
 
 import logging
 import re
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from pathlib import Path
 from typing import Any
 
 from collections.abc import Callable
 
-from agent_team.db.schema import connect, find_open_question_by_channel_ref
+from agent_team.db.schema import (
+    connect,
+    find_open_question_kind_by_channel_ref,
+)
 from agent_team.responder import AnswerOutcome, EnqueueResume, submit_answer
-from agent_team.transport.slack_adapter import SlackPoster, SlackTransport
+from agent_team.transport.slack_adapter import (
+    PLAN_DECISION_KIND,
+    PLAN_DECISION_REQUEST_CHANGES_ACTION,
+    SlackPoster,
+    SlackTransport,
+    build_request_changes_modal,
+    map_plan_decision,
+)
+from agent_team.ledger import get_question
 
 __all__ = [
     "NewTaskCallback",
@@ -521,9 +532,11 @@ class SlackListener:
 
         1. **Explicit id** — if :meth:`SlackTransport.parse_answer` can already
            recover a ``question_id`` (callback_id / nested view-or-message
-           metadata / bare ``question_id``), pass the original payload straight
+           metadata / button value / modal ``private_metadata`` / bare
-           through unchanged. This covers block_actions, view_submission, slash
+           ``question_id``), pass the original payload through unchanged UNLESS
-           commands, and any synthetic-but-explicit reply.
+           the row's ``kind == 'plan_decision'`` (see KIND-AWARE below). This
+           covers block_actions, view_submission, slash commands, and any
+           synthetic-but-explicit reply.
         2. **Thread-reply fallback** — only when (1) raises ``ValueError`` (no
            recoverable id): a real free-text Events API reply. Resolve the
            question by the inner event's ``thread_ts`` against the OPEN ledger
@@ -533,17 +546,40 @@ class SlackListener:
            ``status='open'``-constrained lookup is anti-replay: a thread_ts for
            a closed / answered / expired row resolves to ``None`` → ignored.
 
+        KIND-AWARE DECISION NORMALIZATION (B3, the load-bearing safety map). For
+        a row whose ``kind == 'plan_decision'`` — in BOTH the explicit-id and
+        thread-reply paths — the raw answer is mapped to a structured decision
+        ``{"decision","notes"}`` via :func:`map_plan_decision` BEFORE it reaches
+        the graph. Without this, the graph's decision parser maps any
+        unrecognized verb to ``abandon`` → terminal FAILED, so arbitrary
+        change-request prose ("use pytest fixtures instead") would silently FAIL
+        the task instead of requesting changes. ``approve``/``abandon`` map to
+        the named verb; EVERYTHING else maps to ``request_changes`` carrying the
+        full original reply as notes (never an accidental abandon). A
+        ``clarify`` row passes through UNCHANGED (its free text IS the answer).
+
         Returns the payload to submit, or ``None`` when no question can be
         resolved (the caller ignores the event without crashing). Re-raises
         nothing: a genuinely unrecoverable explicit payload surfaces as the
         ``ValueError`` from the final ``submit_answer`` call in ``handle_event``.
         """
         try:
-            self._transport.parse_answer(raw_payload)
+            question_id, answer, _via = self._transport.parse_answer(raw_payload)
         except ValueError:
             pass
         else:
-            # Explicit id recovered; submit the original payload unchanged.
+            # Explicit id recovered (button / view_submission / explicit reply).
+            # Kind-aware normalization (B3): a ``plan_decision`` row's raw answer
+            # MUST be mapped to a structured decision BEFORE it reaches the graph
+            # (whose parser FAILs any unrecognized verb). Look up the row's kind
+            # and, for a plan decision, synthesize an explicit payload carrying
+            # the {"decision","notes"} dict; otherwise pass through unchanged.
+            kind = self._question_kind(conn, question_id)
+            if kind == PLAN_DECISION_KIND:
+                return {
+                    "question_id": question_id,
+                    "answer": map_plan_decision(answer),
+                }
             return raw_payload
 
         # No explicit id: try the events-API thread-reply mapping.
@@ -556,8 +592,8 @@ class SlackListener:
             # and ignores.
             return raw_payload
 
-        question_id = find_open_question_by_channel_ref(conn, str(thread_ts))
+        resolved = find_open_question_kind_by_channel_ref(conn, str(thread_ts))
-        if question_id is None:
+        if resolved is None:
             # thread_ts matched no OPEN row (stale / replayed / answered): ignore.
             _LOG.debug(
                 "ignoring Slack thread reply: thread_ts %r matched no open "
@@ -565,14 +601,84 @@ class SlackListener:
                 thread_ts,
             )
             return None
+        question_id, kind = resolved
 
         text = event.get("text")
         answer = text.strip() if isinstance(text, str) else text
+        # Kind-aware normalization (B3): a free-text reply to a plan-decision row
+        # — arbitrary change-request prose — is mapped to a structured decision
+        # (request_changes carrying the full reply as notes; NEVER an accidental
+        # abandon). A clarify row's free text passes through unchanged: the text
+        # IS the answer.
+        if kind == PLAN_DECISION_KIND:
+            answer = map_plan_decision(answer)
         # Synthesize an explicit payload the transport already understands
         # (bare question_id + answer). The answer is opaque DATA — stored
         # verbatim as JSON by the responder, never interpreted.
         return {"question_id": question_id, "answer": answer}
 
+    def _question_kind(self, conn: Any, question_id: str) -> str | None:
+        """Return the ``kind`` of an open ledger row, or ``None`` if absent.
+
+        Used by :meth:`_resolve_payload` to decide whether an explicit-id answer
+        (button / modal / explicit reply) needs kind-aware decision
+        normalization. A missing row (forged / stale id) returns ``None`` so the
+        answer passes through unchanged and the responder's compare-and-set
+        rejects it (anti-replay). Tolerant: any read error degrades to ``None``
+        (pass-through), never crashing the listen loop.
+        """
+        try:
+            row = get_question(conn, question_id)
+        except Exception:  # noqa: BLE001 - a ledger read error must not crash handling
+            _LOG.debug(
+                "kind lookup failed for question_id=%s; treating as non-plan",
+                question_id,
+                exc_info=True,
+            )
+            return None
+        return row.kind if row is not None else None
+
+    def _open_request_changes_modal(self, body: Mapping[str, Any], client: Any) -> None:
+        """Open the "Request changes" notes modal for an authorized owner (B3).
+
+        Wired to the dedicated ``@app.action(plan_decision:request_changes)``
+        listener. Preserves AUTHZ-01 ordering: authorize the sender BEFORE any
+        side effect (the ``views.open`` is the side effect here). On rejection,
+        does nothing (the catch-all already skips this action, so no answer is
+        submitted). The modal's eventual ``view_submission`` carries the notes
+        and flows through :meth:`handle_event` like any other answer.
+
+        Fully guarded: recovering the trigger_id / question_id or the
+        ``views_open`` call failing is swallowed (best-effort) — the owner can
+        always fall back to a free-text thread reply. Never raises into Bolt.
+        """
+        if not self._is_authorized(body):
+            return
+        trigger_id = body.get("trigger_id")
+        if not trigger_id:
+            _LOG.debug("request-changes action carried no trigger_id; ignoring")
+            return
+        try:
+            question_id = self._transport.parse_answer(body)[0]
+        except Exception:  # noqa: BLE001 - no recoverable id: cannot open a modal
+            _LOG.debug(
+                "request-changes action carried no recoverable question_id; "
+                "ignoring (owner can still reply free-text)",
+            )
+            return
+        try:
+            client.views_open(
+                trigger_id=str(trigger_id),
+                view=build_request_changes_modal(question_id),
+            )
+        except Exception:  # noqa: BLE001 - a modal-open failure must not crash the loop
+            _LOG.warning(
+                "failed to open request-changes modal for question_id=%s; "
+                "owner can still reply free-text in-thread",
+                question_id,
+                exc_info=True,
+            )
+
     def serve(self) -> None:  # pragma: no cover - live socket, not unit-tested
         """Open the Socket Mode connection and forward events to ``handle_event``.
 
@@ -626,14 +732,30 @@ class SlackListener:
         def _forward(body: Mapping[str, Any]) -> None:
             self.handle_event(body)
 
+        # DEDICATED handler for the plan-gate "Request changes" button — MUST be
+        # registered BEFORE the generic catch-all so it wins. Unlike Approve /
+        # Abandon (which submit immediately), this button has no free-form notes,
+        # so instead of forwarding to handle_event it AUTHORIZES the sender
+        # (AUTHZ-01, before any side effect) and opens a notes modal. The modal's
+        # eventual ``view_submission`` carries the notes and flows through
+        # handle_event like any other answer.
+        @app.action(PLAN_DECISION_REQUEST_CHANGES_ACTION)
+        def _on_request_changes(ack: Any, body: Mapping[str, Any], client: Any) -> None:
+            ack()
+            self._open_request_changes_modal(body, client)
+
         # Match ANY block_actions interaction. slack_bolt rejects an empty-dict
         # constraint (``BoltError: action ({}) must be any of str, Pattern, and
         # dict``); a catch-all ``action_id`` regex is the supported way to
         # register a single handler for every block action. ``handle_event`` does
-        # the real filtering + auth, so over-matching here is safe.
+        # the real filtering + auth, so over-matching here is safe. The
+        # request-changes button is handled by its dedicated listener above; skip
+        # it here so it does not ALSO submit an empty (notes-less) decision.
         @app.action(re.compile(r".*"))  # any block_actions interaction
         def _on_action(ack: Any, body: Mapping[str, Any]) -> None:
             ack()
+            if _is_request_changes_action(body):
+                return
             _forward(body)
 
         @app.event("message")
@@ -690,6 +812,23 @@ class SlackListener:
             _LOG.debug("SlackListener.close: handler teardown raised; ignoring")
 
 
+def _is_request_changes_action(body: Mapping[str, Any]) -> bool:
+    """Return ``True`` iff ``body`` is a plan-gate "Request changes" button click.
+
+    The generic catch-all action listener uses this to SKIP the request-changes
+    button (handled by its dedicated modal-opening listener), so it does not
+    ALSO submit an empty (notes-less) decision through ``handle_event``.
+    """
+    actions = body.get("actions")
+    if not isinstance(actions, Sequence) or isinstance(actions, (str, bytes)):
+        return False
+    return any(
+        isinstance(action, Mapping)
+        and action.get("action_id") == PLAN_DECISION_REQUEST_CHANGES_ACTION
+        for action in actions
+    )
+
+
 def _inner_event(raw_payload: Mapping[str, Any]) -> Mapping[str, Any]:
     """Return the Events API inner event, or an empty mapping if there is none.
 

agent-team/tests/test_schema.py

@@ -16,6 +16,7 @@ from agent_team.db.schema import (
     answer_question,
     connect,
     expire_question,
+    find_open_question_kind_by_channel_ref,
     init_db,
     issue_already_ingested,
     migrate,
@@ -58,6 +59,49 @@ def test_question_states_match_ddl_check() -> None:
         assert f"'{state}'" in PENDING_QUESTIONS_DDL
 
 
+def test_find_open_question_kind_by_channel_ref_returns_qid_and_kind(
+    tmp_path: Path,
+) -> None:
+    db = tmp_path / "db.sqlite"
+    init_db(db)
+    conn = connect(db)
+    try:
+        conn.execute(
+            "INSERT INTO pending_questions "
+            "(question_id, thread_id, turn, status, transport, channel_ref, kind) "
+            "VALUES ('q1', 't1', 0, 'open', 'slack', 'TS.1', 'plan_decision')"
+        )
+        conn.commit()
+        assert find_open_question_kind_by_channel_ref(conn, "TS.1") == (
+            "q1",
+            "plan_decision",
+        )
+        # Empty ref / no match -> None.
+        assert find_open_question_kind_by_channel_ref(conn, "") is None
+        assert find_open_question_kind_by_channel_ref(conn, "NOPE") is None
+    finally:
+        conn.close()
+
+
+def test_find_open_question_kind_by_channel_ref_constrained_to_open(
+    tmp_path: Path,
+) -> None:
+    """Anti-replay: a non-open row's channel_ref resolves to None."""
+    db = tmp_path / "db.sqlite"
+    init_db(db)
+    conn = connect(db)
+    try:
+        conn.execute(
+            "INSERT INTO pending_questions "
+            "(question_id, thread_id, turn, status, transport, channel_ref, kind) "
+            "VALUES ('q1', 't1', 0, 'answered', 'slack', 'TS.1', 'plan_decision')"
+        )
+        conn.commit()
+        assert find_open_question_kind_by_channel_ref(conn, "TS.1") is None
+    finally:
+        conn.close()
+
+
 def test_connect_sets_pragmas(tmp_path: Path) -> None:
     conn = connect(tmp_path / "db.sqlite")
     try:

agent-team/tests/test_slack_adapter.py

@@ -16,11 +16,18 @@ import pytest
 from agent_team.transport.base import QuestionSet, Transport
 from agent_team.transport.slack_adapter import (
     CALLBACK_ID_PREFIX,
+    PLAN_DECISION_ABANDON_ACTION,
+    PLAN_DECISION_APPROVE_ACTION,
+    PLAN_DECISION_KIND,
+    PLAN_DECISION_REQUEST_CHANGES_ACTION,
     VIA_SLACK,
     SlackPostError,
     SlackTransport,
     build_callback_id,
+    build_plan_decision_blocks,
     build_question_blocks,
+    build_request_changes_modal,
+    map_plan_decision,
     parse_callback_id,
 )
 
@@ -396,3 +403,270 @@ def test_post_then_parse_round_trips_question_id() -> None:
     assert qid == "q-round"
     assert answer == "approved"
     assert via == VIA_SLACK
+
+
+# --------------------------------------------------------------------------- #
+# B3 — kind-aware plan-decision normalization (map_plan_decision)             #
+# --------------------------------------------------------------------------- #
+
+
+@pytest.mark.parametrize(
+    "raw",
+    ["approve", "approved", "yes", "ok", "lgtm", "ship", "APPROVE", "  Yes  "],
+)
+def test_map_plan_decision_approve_verbs(raw: str) -> None:
+    """All approve-allowlist verbs map to approve (case + whitespace insensitive)."""
+    result = map_plan_decision(raw)
+    assert result == {"decision": "approve", "notes": ""}
+
+
+@pytest.mark.parametrize(
+    "raw",
+    ["abandon", "reject", "cancel", "stop", "kill", "ABANDON", "  Cancel  "],
+)
+def test_map_plan_decision_abandon_verbs(raw: str) -> None:
+    """All abandon-allowlist verbs map to abandon (case + whitespace insensitive)."""
+    result = map_plan_decision(raw)
+    assert result == {"decision": "abandon", "notes": ""}
+
+
+def test_map_plan_decision_arbitrary_prose_is_request_changes_not_abandon() -> None:
+    """THE ANTI-FAIL TEST: arbitrary change prose -> request_changes, NOT abandon.
+
+    The graph maps any unrecognized verb to abandon -> terminal FAILED. This is
+    the load-bearing guard that real change-request notes never silently FAIL a
+    task: free prose becomes request_changes carrying the FULL original reply as
+    notes, and is explicitly asserted to NOT be abandon (or approve).
+    """
+    raw = "use pytest fixtures instead of setUp methods"
+    result = map_plan_decision(raw)
+    assert result["decision"] == "request_changes"
+    assert result["decision"] != "abandon"
+    assert result["decision"] != "approve"
+    # The full original reply is preserved as the notes.
+    assert result["notes"] == raw
+
+
+def test_map_plan_decision_preserves_original_casing_in_notes() -> None:
+    """request_changes notes keep the human's exact wording (not lowercased)."""
+    raw = "Please Add Type Hints To The New Helper"
+    result = map_plan_decision(raw)
+    assert result == {"decision": "request_changes", "notes": raw}
+
+
+@pytest.mark.parametrize("raw", ["", "   ", "\n\t "])
+def test_map_plan_decision_empty_is_request_changes_safe_default(raw: str) -> None:
+    """Empty / whitespace-only input maps to request_changes (never abandon)."""
+    result = map_plan_decision(raw)
+    assert result["decision"] == "request_changes"
+    assert result["decision"] != "abandon"
+    assert result["notes"] == ""
+
+
+def test_map_plan_decision_none_is_request_changes() -> None:
+    """A None answer degrades to request_changes with empty notes (never abandon)."""
+    result = map_plan_decision(None)
+    assert result == {"decision": "request_changes", "notes": ""}
+
+
+def test_map_plan_decision_verb_with_trailing_text_is_request_changes() -> None:
+    """A verb embedded in a sentence is NOT a bare verb -> request_changes.
+
+    "approve but tweak X" is a change request, not an approval — only an exact
+    bare verb match approves.
+    """
+    raw = "approve but please tweak the error handling first"
+    result = map_plan_decision(raw)
+    assert result["decision"] == "request_changes"
+    assert result["notes"] == raw
+
+
+# --------------------------------------------------------------------------- #
+# B3 — gate decision blocks (the three buttons)                               #
+# --------------------------------------------------------------------------- #
+
+
+def test_build_plan_decision_blocks_has_three_buttons_with_action_ids() -> None:
+    blocks = build_plan_decision_blocks("q-xyz", "Plan summary + findings")
+    # A section (the body) + an actions block with three buttons.
+    assert blocks[0]["type"] == "section"
+    assert "Plan summary" in blocks[0]["text"]["text"]
+    actions = blocks[1]
+    assert actions["type"] == "actions"
+    action_ids = [e["action_id"] for e in actions["elements"]]
+    assert action_ids == [
+        PLAN_DECISION_APPROVE_ACTION,
+        PLAN_DECISION_REQUEST_CHANGES_ACTION,
+        PLAN_DECISION_ABANDON_ACTION,
+    ]
+
+
+def test_build_plan_decision_blocks_buttons_embed_recoverable_question_id() -> None:
+    """Each button value encodes ``"<verb>:<question_id>"`` for recovery."""
+    blocks = build_plan_decision_blocks("q-recover", "body")
+    values = [e["value"] for e in blocks[1]["elements"]]
+    assert values == [
+        "approve:q-recover",
+        "request_changes:q-recover",
+        "abandon:q-recover",
+    ]
+    # The question_id is recoverable from every button value.
+    for v in values:
+        assert v.endswith(":q-recover")
+
+
+def test_build_plan_decision_blocks_abandon_has_danger_confirm() -> None:
+    blocks = build_plan_decision_blocks("q1", "body")
+    abandon = blocks[1]["elements"][2]
+    assert abandon["style"] == "danger"
+    assert "confirm" in abandon
+    approve = blocks[1]["elements"][0]
+    assert approve["style"] == "primary"
+
+
+# --------------------------------------------------------------------------- #
+# B3 — block_actions button parse_answer recovery                            #
+# --------------------------------------------------------------------------- #
+
+
+def test_parse_answer_recovers_qid_and_verb_from_button_value() -> None:
+    """An approve button with no callback_id recovers qid + bare verb from value."""
+    t = SlackTransport(channel="C1", poster=_RecordingPoster())
+    payload = {
+        "type": "block_actions",
+        "actions": [
+            {"action_id": PLAN_DECISION_APPROVE_ACTION, "value": "approve:q-btn"}
+        ],
+    }
+    qid, answer, via = t.parse_answer(payload)
+    assert qid == "q-btn"
+    # The answer is the bare verb (suffix stripped) so map_plan_decision matches.
+    assert answer == "approve"
+    assert via == VIA_SLACK
+
+
+def test_parse_answer_recovers_abandon_button_value() -> None:
+    t = SlackTransport(channel="C1", poster=_RecordingPoster())
+    payload = {
+        "type": "block_actions",
+        "actions": [{"action_id": PLAN_DECISION_ABANDON_ACTION, "value": "abandon:q9"}],
+    }
+    qid, answer, _via = t.parse_answer(payload)
+    assert qid == "q9"
+    assert answer == "abandon"
+
+
+def test_parse_answer_button_metadata_takes_precedence_for_qid() -> None:
+    """When message metadata carries the qid, it is used (callback_id dropped live)."""
+    t = SlackTransport(channel="C1", poster=_RecordingPoster())
+    payload = {
+        "type": "block_actions",
+        "message": {
+            "metadata": {"event_payload": {"question_id": "q-meta"}},
+        },
+        "actions": [
+            {"action_id": PLAN_DECISION_APPROVE_ACTION, "value": "approve:q-meta"}
+        ],
+    }
+    qid, answer, _via = t.parse_answer(payload)
+    assert qid == "q-meta"
+    assert answer == "approve"
+
+
+# --------------------------------------------------------------------------- #
+# B3 — request-changes modal + view_submission                               #
+# --------------------------------------------------------------------------- #
+
+
+def test_build_request_changes_modal_round_trips_question_id() -> None:
+    modal = build_request_changes_modal("q-modal")
+    assert modal["type"] == "modal"
+    assert modal["private_metadata"] == "request_changes:q-modal"
+    # One required multiline input.
+    block = modal["blocks"][0]
+    assert block["type"] == "input"
+    assert block["element"]["type"] == "plain_text_input"
+    assert block["element"]["multiline"] is True
+
+
+def _view_submission(question_id: str, notes: str) -> dict[str, Any]:
+    """A realistic ``view_submission`` payload for the request-changes modal."""
+    return {
+        "type": "view_submission",
+        "user": {"id": "U_OWNER"},
+        "view": {
+            "callback_id": "plan_decision:request_changes_modal",
+            "private_metadata": f"request_changes:{question_id}",
+            "state": {
+                "values": {
+                    "plan_decision_notes_block": {
+                        "plan_decision_notes_input": {
+                            "type": "plain_text_input",
+                            "value": notes,
+                        }
+                    }
+                }
+            },
+        },
+    }
+
+
+def test_parse_answer_recovers_qid_and_notes_from_view_submission() -> None:
+    """The modal submit recovers qid from private_metadata + notes from input."""
+    t = SlackTransport(channel="C1", poster=_RecordingPoster())
+    payload = _view_submission("q-sub", "switch to dependency injection")
+    qid, answer, via = t.parse_answer(payload)
+    assert qid == "q-sub"
+    assert answer == "switch to dependency injection"
+    assert via == VIA_SLACK
+
+
+def test_view_submission_notes_map_to_request_changes() -> None:
+    """The modal's free-text notes normalize to request_changes (via the mapper)."""
+    t = SlackTransport(channel="C1", poster=_RecordingPoster())
+    _qid, answer, _via = t.parse_answer(
+        _view_submission("q1", "use a factory function")
+    )
+    decision = map_plan_decision(answer)
+    assert decision == {
+        "decision": "request_changes",
+        "notes": "use a factory function",
+    }
+
+
+# --------------------------------------------------------------------------- #
+# B3 — post_question renders decision buttons for a plan_decision question    #
+# --------------------------------------------------------------------------- #
+
+
+def test_post_question_renders_plan_decision_buttons() -> None:
+    poster = _RecordingPoster()
+    t = SlackTransport(channel="C1", poster=poster)
+    qs = QuestionSet(
+        thread_id="t1",
+        question_id="q-gate",
+        turn=2,
+        questions=["Approve, request changes, or abandon this plan?"],
+        context={"kind": PLAN_DECISION_KIND, "presentation": "Plan body here"},
+    )
+    t.post_question(
+        thread_id="t1",
+        question_id="q-gate",
+        turn=2,
+        question_set=qs,
+        deadline="2026-06-18T00:00:00Z",
+    )
+    blocks = poster.calls[0]["blocks"]
+    action_ids = [
+        e["action_id"]
+        for b in blocks
+        if b.get("type") == "actions"
+        for e in b["elements"]
+    ]
+    assert PLAN_DECISION_APPROVE_ACTION in action_ids
+    assert PLAN_DECISION_REQUEST_CHANGES_ACTION in action_ids
+    assert PLAN_DECISION_ABANDON_ACTION in action_ids
+    # The presentation body is the message text fallback.
+    assert poster.calls[0]["text"] == "Plan body here"
+    # The question_id is double-anchored in message metadata.
+    assert poster.calls[0]["metadata"]["event_payload"]["question_id"] == "q-gate"

agent-team/tests/test_slack_listener.py

@@ -878,3 +878,224 @@ def test_new_task_root_post_failure_degrades_to_empty_root_ts(db_path: Path) ->
     assert listener.handle_event(payload) is None
     # Task still started, with an empty root_ts (top-level questions).
     assert seen == [("do the thing", "slack", "")]
+
+
+# ---------------------------------------------------------------------------
+# B3 — kind-aware plan-decision normalization in _resolve_payload.
+# ---------------------------------------------------------------------------
+
+
+def _seed_plan_decision_question(
+    db_path: Path,
+    *,
+    thread_id: str = "t1",
+    question_id: str = "q1",
+    turn: int = 0,
+    channel_ref: str = SEED_CHANNEL_REF,
+) -> None:
+    """Insert a real ``open`` ``kind='plan_decision'`` ledger row + channel_ref.
+
+    Mirrors a B2b plan-gate row: open, plan_decision kind, with the gate
+    message ts recorded as channel_ref so an inbound thread reply maps back.
+    """
+    from agent_team import ledger as ledger_mod
+
+    conn = connect(db_path)
+    try:
+        ledger_mod.post_question(
+            conn,
+            question_id=question_id,
+            thread_id=thread_id,
+            turn=turn,
+            transport="SlackTransport",
+            deadline_at="2026-06-18T00:00:00+00:00",
+            kind="plan_decision",
+        )
+        ledger_mod.set_channel_ref(
+            conn, question_id=question_id, channel_ref=channel_ref
+        )
+    finally:
+        conn.close()
+
+
+def test_plan_decision_free_text_prose_maps_to_request_changes_not_fail(
+    db_path: Path,
+) -> None:
+    """THE ANTI-FAIL TEST at the listener seam (end-to-end-ish).
+
+    A free-text reply of arbitrary change-request prose to a plan_decision row
+    must reach submit_answer as a structured {"decision":"request_changes",...}
+    carrying the full reply as notes — NEVER an abandon (which the graph would
+    map to terminal FAILED). This is the whole point of B3.
+    """
+    _seed_plan_decision_question(
+        db_path, question_id="q1", channel_ref=SEED_CHANNEL_REF
+    )
+    queue = RecordingQueue()
+    listener = _listener(db_path, queue, owner_ids={OWNER_ID})
+
+    outcome = listener.handle_event(
+        _events_api_reply(text="please add type hints", thread_ts=SEED_CHANNEL_REF)
+    )
+
+    assert outcome is not None and outcome.accepted is True
+    assert len(queue.jobs) == 1
+    answer = queue.jobs[0].answer
+    assert isinstance(answer, dict)
+    assert answer["decision"] == "request_changes"
+    assert answer["decision"] != "abandon"
+    assert answer["notes"] == "please add type hints"
+
+
+def test_plan_decision_free_text_approve_maps_to_approve(db_path: Path) -> None:
+    _seed_plan_decision_question(
+        db_path, question_id="q1", channel_ref=SEED_CHANNEL_REF
+    )
+    queue = RecordingQueue()
+    listener = _listener(db_path, queue, owner_ids={OWNER_ID})
+
+    listener.handle_event(_events_api_reply(text="approve", thread_ts=SEED_CHANNEL_REF))
+
+    answer = queue.jobs[0].answer
+    assert answer == {"decision": "approve", "notes": ""}
+
+
+def test_plan_decision_free_text_abandon_maps_to_abandon(db_path: Path) -> None:
+    _seed_plan_decision_question(
+        db_path, question_id="q1", channel_ref=SEED_CHANNEL_REF
+    )
+    queue = RecordingQueue()
+    listener = _listener(db_path, queue, owner_ids={OWNER_ID})
+
+    listener.handle_event(_events_api_reply(text="cancel", thread_ts=SEED_CHANNEL_REF))
+
+    answer = queue.jobs[0].answer
+    assert answer == {"decision": "abandon", "notes": ""}
+
+
+def test_clarify_free_text_passes_through_unchanged_regression(db_path: Path) -> None:
+    """A clarify row's free text is the answer verbatim (NOT decision-mapped)."""
+    # _seed_open_question seeds a default kind='clarify' row.
+    _seed_open_question(db_path, question_id="q1", channel_ref=SEED_CHANNEL_REF)
+    queue = RecordingQueue()
+    listener = _listener(db_path, queue, owner_ids={OWNER_ID})
+
+    listener.handle_event(
+        _events_api_reply(text="use the release branch", thread_ts=SEED_CHANNEL_REF)
+    )
+
+    # The answer is the raw string, NOT a {"decision",...} dict.
+    assert queue.jobs[0].answer == "use the release branch"
+
+
+def test_plan_decision_approve_button_resolves_decision(db_path: Path) -> None:
+    """An approve BUTTON (block_actions) yields {"decision":"approve"}."""
+    _seed_plan_decision_question(
+        db_path, question_id="q1", channel_ref=SEED_CHANNEL_REF
+    )
+    queue = RecordingQueue()
+    listener = _listener(db_path, queue, owner_ids={OWNER_ID})
+
+    payload = {
+        "type": "block_actions",
+        "user": {"id": OWNER_ID},
+        "message": {"metadata": {"event_payload": {"question_id": "q1"}}},
+        "actions": [{"action_id": "plan_decision:approve", "value": "approve:q1"}],
+    }
+    outcome = listener.handle_event(payload)
+
+    assert outcome is not None and outcome.accepted is True
+    assert queue.jobs[0].answer == {"decision": "approve", "notes": ""}
+
+
+def test_plan_decision_abandon_button_resolves_decision(db_path: Path) -> None:
+    _seed_plan_decision_question(
+        db_path, question_id="q1", channel_ref=SEED_CHANNEL_REF
+    )
+    queue = RecordingQueue()
+    listener = _listener(db_path, queue, owner_ids={OWNER_ID})
+
+    payload = {
+        "type": "block_actions",
+        "user": {"id": OWNER_ID},
+        "message": {"metadata": {"event_payload": {"question_id": "q1"}}},
+        "actions": [{"action_id": "plan_decision:abandon", "value": "abandon:q1"}],
+    }
+    listener.handle_event(payload)
+    assert queue.jobs[0].answer == {"decision": "abandon", "notes": ""}
+
+
+def test_view_submission_modal_yields_request_changes_with_notes(db_path: Path) -> None:
+    """A request-changes modal submit yields {"decision":"request_changes",notes}."""
+    _seed_plan_decision_question(
+        db_path, question_id="q1", channel_ref=SEED_CHANNEL_REF
+    )
+    queue = RecordingQueue()
+    listener = _listener(db_path, queue, owner_ids={OWNER_ID})
+
+    payload = {
+        "type": "view_submission",
+        "user": {"id": OWNER_ID},
+        "view": {
+            "callback_id": "plan_decision:request_changes_modal",
+            "private_metadata": "request_changes:q1",
+            "state": {
+                "values": {
+                    "plan_decision_notes_block": {
+                        "plan_decision_notes_input": {
+                            "type": "plain_text_input",
+                            "value": "tighten the error handling",
+                        }
+                    }
+                }
+            },
+        },
+    }
+    outcome = listener.handle_event(payload)
+
+    assert outcome is not None and outcome.accepted is True
+    assert queue.jobs[0].answer == {
+        "decision": "request_changes",
+        "notes": "tighten the error handling",
+    }
+
+
+def test_open_request_changes_modal_authorizes_before_opening(db_path: Path) -> None:
+    """AUTHZ-01: a non-owner request-changes click never opens a modal."""
+
+    class _Client:
+        def __init__(self) -> None:
+            self.opened: list[dict[str, Any]] = []
+
+        def views_open(self, *, trigger_id: str, view: dict[str, Any]) -> None:
+            self.opened.append({"trigger_id": trigger_id, "view": view})
+
+    _seed_plan_decision_question(
+        db_path, question_id="q1", channel_ref=SEED_CHANNEL_REF
+    )
+    listener = _listener(db_path, RecordingQueue(), owner_ids={OWNER_ID})
+    client = _Client()
+
+    # Non-owner: must NOT open a modal.
+    intruder = {
+        "type": "block_actions",
+        "user": {"id": "U_INTRUDER"},
+        "trigger_id": "TRIG.1",
+        "message": {"metadata": {"event_payload": {"question_id": "q1"}}},
+        "actions": [
+            {
+                "action_id": "plan_decision:request_changes",
+                "value": "request_changes:q1",
+            }
+        ],
+    }
+    listener._open_request_changes_modal(intruder, client)
+    assert client.opened == []
+
+    # Owner: opens a modal whose private_metadata round-trips the question_id.
+    owner = dict(intruder)
+    owner["user"] = {"id": OWNER_ID}
+    listener._open_request_changes_modal(owner, client)
+    assert len(client.opened) == 1
+    assert client.opened[0]["trigger_id"] == "TRIG.1"
+    assert client.opened[0]["view"]["private_metadata"] == "request_changes:q1"