feat(agent-team): add pending_questions.kind discriminator + migration (Phase B1)

The plan-review gate (coming next) needs to tell its decision questions apart
from clarifier questions in the durable ledger. Add a `kind` column to
pending_questions (values 'clarify' | 'plan_decision').

- Fresh DBs: `kind TEXT NOT NULL DEFAULT 'clarify'` (+ CHECK) in the DDL.
- Live ledger: idempotent additive migration (SCHEMA_VERSION 3→4) — a guarded
  ALTER (PRAGMA table_info) run from both migrate() and init_db; legacy rows
  take the 'clarify' default, never null. (SQLite can't add a CHECK via ALTER,
  so the migrated column is NOT NULL DEFAULT only; value constraint is enforced
  on fresh DBs by the CHECK and on all writes by the typed helper.)
- ledger.post_question gains a keyword-only `kind="clarify"` (backward
  compatible — existing callers unchanged); PendingQuestion.from_row reads it.

Tests: fresh-DB column+default, idempotent init_db, legacy-DB backfill to
'clarify', plan_decision round-trip. 1396 passed.

agent-team/agent_team/db/schema.py

@@ -52,7 +52,7 @@ __all__ = [
 ]
 
 # Bump when the DDL below changes; migrate() steps a connection forward.
-SCHEMA_VERSION: int = 3
+SCHEMA_VERSION: int = 4
 
 # Default SQLite busy timeout (ms) so concurrent writers wait for the write
 # lock rather than failing immediately.
@@ -76,7 +76,9 @@ CREATE TABLE IF NOT EXISTS pending_questions (
     deadline_at   TEXT,
     answer_json   TEXT,
     answered_at   TEXT,
-    answered_via  TEXT
+    answered_via  TEXT,
+    kind          TEXT NOT NULL DEFAULT 'clarify'
+                  CHECK (kind IN ('clarify', 'plan_decision'))
 )
 """.strip()
 
@@ -216,6 +218,35 @@ def connect(db_path: Path) -> sqlite3.Connection:
     return conn
 
 
+def _pending_questions_has_kind(conn: sqlite3.Connection) -> bool:
+    """Return True if ``pending_questions`` already has the ``kind`` column.
+
+    Inspects ``PRAGMA table_info(pending_questions)`` so the additive ``kind``
+    migration can be applied only when absent — making it idempotent on an
+    already-migrated (or freshly created) DB.
+    """
+    rows = conn.execute("PRAGMA table_info(pending_questions)").fetchall()
+    return any(row["name"] == "kind" for row in rows)
+
+
+def _ensure_pending_questions_kind(conn: sqlite3.Connection) -> None:
+    """Idempotently add the ``kind`` discriminator column to ``pending_questions``.
+
+    Fresh DBs get ``kind`` from :data:`PENDING_QUESTIONS_DDL`; an existing (live
+    R720) ledger whose table predates the column gets it via an in-place additive
+    ``ALTER TABLE``, guarded by :func:`_pending_questions_has_kind` so a second
+    run is a no-op. Existing rows take the ``'clarify'`` default. SQLite cannot
+    add a CHECK constraint via ALTER, so the added column carries only the
+    NOT NULL DEFAULT; the CHECK is enforced on fresh DBs via the CREATE DDL and
+    on writes via the typed insert helper.
+    """
+    if not _pending_questions_has_kind(conn):
+        conn.execute(
+            "ALTER TABLE pending_questions "
+            "ADD COLUMN kind TEXT NOT NULL DEFAULT 'clarify'"
+        )
+
+
 def init_db(db_path: Path) -> None:
     """Create the agent-team tables in ``db_path`` if absent.
 
@@ -229,6 +260,10 @@ def init_db(db_path: Path) -> None:
     try:
         conn.execute(SCHEMA_META_DDL)
         conn.execute(PENDING_QUESTIONS_DDL)
+        # Additive in-place migration for an existing ledger whose
+        # pending_questions predates the ``kind`` column (CREATE IF NOT EXISTS
+        # above never alters an existing table). No-op on fresh/already-migrated.
+        _ensure_pending_questions_kind(conn)
         for stmt in _split_statements(PENDING_QUESTIONS_INDEXES_DDL):
             conn.execute(stmt)
         conn.execute(BUDGET_LEDGER_DDL)
@@ -286,7 +321,17 @@ def migrate(conn: sqlite3.Connection) -> None:
             conn.execute(stmt)
         current = 3
 
-    # Future steps go here: `if current < 4: ...; current = 4`.
+    if current < 4:
+        # v4: add the ``kind`` discriminator to pending_questions so the
+        # responder/resume layer can tell a clarifier question apart from a
+        # plan-review decision. Additive in-place ALTER (guarded), placed in its
+        # own version block ABOVE the unconditional tail per the ORDERING
+        # CONSTRAINT below — the tail's CREATE ... IF NOT EXISTS would NOT apply
+        # this alter. Existing rows take the 'clarify' default.
+        _ensure_pending_questions_kind(conn)
+        current = 4
+
+    # Future steps go here: `if current < 5: ...; current = 5`.
 
     # Applied UNCONDITIONALLY (idempotent IF NOT EXISTS) so an already-stamped DB
     # — which skips the version blocks above — still gains these tables without a

agent-team/agent_team/db/schema.sql

@@ -25,7 +25,13 @@ CREATE TABLE IF NOT EXISTS pending_questions (
     deadline_at   TEXT,
     answer_json   TEXT,
     answered_at   TEXT,
-    answered_via  TEXT
+    answered_via  TEXT,
+    -- kind: discriminates the human gate this question belongs to —
+    -- 'clarify' (the clarifier) or 'plan_decision' (the plan-review gate).
+    -- Defaults to 'clarify' so an in-place ALTER on a legacy ledger and any
+    -- existing rows take the clarifier value.
+    kind          TEXT NOT NULL DEFAULT 'clarify'
+                  CHECK (kind IN ('clarify', 'plan_decision'))
 );
 
 CREATE INDEX IF NOT EXISTS idx_pending_questions_thread

agent-team/agent_team/ledger.py

@@ -90,10 +90,12 @@ class PendingQuestion:
     answer_json: str | None = None
     answered_at: str | None = None
     answered_via: str | None = None
+    kind: str = "clarify"
 
     @classmethod
     def from_row(cls, row: sqlite3.Row) -> PendingQuestion:
         """Build a :class:`PendingQuestion` from a ``sqlite3.Row``."""
+        keys = row.keys()
         return cls(
             question_id=row["question_id"],
             thread_id=row["thread_id"],
@@ -106,6 +108,9 @@ class PendingQuestion:
             answer_json=row["answer_json"],
             answered_at=row["answered_at"],
             answered_via=row["answered_via"],
+            # Tolerate a row read before the kind column exists (legacy/partial
+            # SELECT): fall back to the 'clarify' default rather than KeyError.
+            kind=row["kind"] if "kind" in keys else "clarify",
         )
 
 
@@ -123,6 +128,7 @@ def post_question(
     transport: str,
     deadline_at: str | None = None,
     posted_at: str | None = None,
+    kind: str = "clarify",
 ) -> None:
     """Insert a new ``open`` question row (delivery step 1 of §3.3.1).
 
@@ -132,14 +138,20 @@ def post_question(
     retries delivery idempotently. The caller records the ref via
     :func:`set_channel_ref` once the post succeeds.
 
+    ``kind`` discriminates the human gate this question belongs to —
+    ``'clarify'`` (the clarifier, the default so existing callers are unchanged)
+    or ``'plan_decision'`` (the plan-review gate). The keyword-only default keeps
+    every existing call site writing clarifier rows with no signature change.
+
     Raises :class:`sqlite3.IntegrityError` if ``question_id`` already exists
     (PK) — re-posting the same question is the caller's reconcile concern, not a
     silent overwrite. ``posted_at`` defaults to now (UTC ISO-8601).
     """
     conn.execute(
         "INSERT INTO pending_questions "
-        "(question_id, thread_id, turn, status, transport, posted_at, deadline_at) "
+        "(question_id, thread_id, turn, status, transport, posted_at, "
-        "VALUES (?, ?, ?, 'open', ?, ?, ?)",
+        "deadline_at, kind) "
+        "VALUES (?, ?, ?, 'open', ?, ?, ?, ?)",
         (
             question_id,
             thread_id,
@@ -147,6 +159,7 @@ def post_question(
             transport,
             posted_at or _utc_now_iso(),
             deadline_at,
+            kind,
         ),
     )
 

agent-team/tests/test_ledger.py

@@ -91,6 +91,29 @@ def test_post_question_duplicate_id_raises(conn: sqlite3.Connection) -> None:
         post_question(conn, question_id="dup", thread_id="t", turn=1, transport="slack")
 
 
+def test_post_question_defaults_kind_clarify(conn: sqlite3.Connection) -> None:
+    """Existing call sites (no kind arg) keep writing clarifier rows."""
+    post_question(conn, question_id="q1", thread_id="t", turn=0, transport="slack")
+    q = get_question(conn, "q1")
+    assert q is not None
+    assert q.kind == "clarify"
+
+
+def test_post_question_accepts_plan_decision_kind(conn: sqlite3.Connection) -> None:
+    """A plan-review row round-trips with kind='plan_decision' on the read path."""
+    post_question(
+        conn,
+        question_id="q-pd",
+        thread_id="t",
+        turn=0,
+        transport="slack",
+        kind="plan_decision",
+    )
+    q = get_question(conn, "q-pd")
+    assert q is not None
+    assert q.kind == "plan_decision"
+
+
 # --------------------------------------------------------------------------
 # set_channel_ref — delivery step 2, guarded on status='open'.
 # --------------------------------------------------------------------------

agent-team/tests/test_schema.py

@@ -536,3 +536,157 @@ def test_migrate_adds_ingested_issues_to_a_legacy_v1_db(tmp_path: Path) -> None:
         assert ver == SCHEMA_VERSION
     finally:
         conn.close()
+
+
+# --------------------------------------------------------------------------- #
+# pending_questions.kind discriminator (schema v4)
+# --------------------------------------------------------------------------- #
+
+
+# Legacy (pre-kind) pending_questions DDL, used to construct a DB whose table
+# predates the additive migration.
+_LEGACY_PENDING_QUESTIONS_DDL = """
+CREATE TABLE IF NOT EXISTS pending_questions (
+    question_id   TEXT PRIMARY KEY,
+    thread_id     TEXT NOT NULL,
+    turn          INTEGER NOT NULL,
+    status        TEXT NOT NULL
+                  CHECK (status IN ('open', 'answered', 'expired', 'superseded')),
+    transport     TEXT NOT NULL,
+    channel_ref   TEXT,
+    posted_at     TEXT,
+    deadline_at   TEXT,
+    answer_json   TEXT,
+    answered_at   TEXT,
+    answered_via  TEXT
+)
+""".strip()
+
+
+def _pq_columns(conn: sqlite3.Connection) -> list[str]:
+    return [r["name"] for r in conn.execute("PRAGMA table_info(pending_questions)")]
+
+
+def test_schema_version_is_at_least_4() -> None:
+    assert SCHEMA_VERSION >= 4
+
+
+def test_init_db_pending_questions_has_kind_defaulting_clarify(
+    tmp_path: Path,
+) -> None:
+    """A fresh init_db gives pending_questions a kind column defaulting clarify."""
+    db = tmp_path / "db.sqlite"
+    init_db(db)
+    conn = connect(db)
+    try:
+        assert "kind" in _pq_columns(conn)
+        _insert_open_question(conn, "q-default")
+        kind = conn.execute(
+            "SELECT kind FROM pending_questions WHERE question_id='q-default'"
+        ).fetchone()["kind"]
+    finally:
+        conn.close()
+    assert kind == "clarify"
+
+
+def test_init_db_kind_is_idempotent(tmp_path: Path) -> None:
+    """Running init_db twice does not error and kind exists exactly once."""
+    db = tmp_path / "db.sqlite"
+    init_db(db)
+    init_db(db)  # must not raise (no duplicate-column error)
+    conn = connect(db)
+    try:
+        cols = _pq_columns(conn)
+    finally:
+        conn.close()
+    assert cols.count("kind") == 1
+
+
+def test_migrate_adds_kind_to_legacy_db_rows_read_clarify(tmp_path: Path) -> None:
+    """A legacy pending_questions (no kind) gains the column; old rows read clarify."""
+    db = tmp_path / "legacy.sqlite"
+    conn = connect(db)
+    try:
+        # Build the OLD table by hand and seed a row, with NO kind column.
+        conn.execute(_LEGACY_PENDING_QUESTIONS_DDL)
+        conn.execute(
+            "INSERT INTO pending_questions "
+            "(question_id, thread_id, turn, status, transport) "
+            "VALUES ('legacy', 't', 0, 'open', 'slack')"
+        )
+        assert "kind" not in _pq_columns(conn)
+
+        init_db(db)
+
+        assert "kind" in _pq_columns(conn)
+        # The pre-existing row reads back as 'clarify' (NOT null).
+        kind = conn.execute(
+            "SELECT kind FROM pending_questions WHERE question_id='legacy'"
+        ).fetchone()["kind"]
+    finally:
+        conn.close()
+    assert kind == "clarify"
+
+
+def test_migrate_helper_adds_kind_to_legacy_db(tmp_path: Path) -> None:
+    """migrate() (not just init_db) installs the v4 kind column on a legacy DB."""
+    db = tmp_path / "legacy2.sqlite"
+    conn = connect(db)
+    try:
+        conn.execute(_LEGACY_PENDING_QUESTIONS_DDL)
+        conn.execute(
+            "CREATE TABLE IF NOT EXISTS schema_meta "
+            "(id INTEGER PRIMARY KEY CHECK (id = 1), schema_version INTEGER NOT NULL)"
+        )
+        conn.execute("INSERT INTO schema_meta (id, schema_version) VALUES (1, 3)")
+        assert "kind" not in _pq_columns(conn)
+
+        migrate(conn)
+
+        assert "kind" in _pq_columns(conn)
+        ver = conn.execute(
+            "SELECT schema_version FROM schema_meta WHERE id = 1"
+        ).fetchone()[0]
+    finally:
+        conn.close()
+    assert ver == SCHEMA_VERSION
+
+
+def test_kind_plan_decision_round_trips(tmp_path: Path) -> None:
+    """A row written with kind='plan_decision' round-trips; default is 'clarify'."""
+    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, kind) "
+            "VALUES ('pd', 't', 0, 'open', 'slack', 'plan_decision')"
+        )
+        _insert_open_question(conn, "cl")  # no kind -> default
+        pd_kind = conn.execute(
+            "SELECT kind FROM pending_questions WHERE question_id='pd'"
+        ).fetchone()["kind"]
+        cl_kind = conn.execute(
+            "SELECT kind FROM pending_questions WHERE question_id='cl'"
+        ).fetchone()["kind"]
+    finally:
+        conn.close()
+    assert pd_kind == "plan_decision"
+    assert cl_kind == "clarify"
+
+
+def test_kind_check_rejects_unknown_value(tmp_path: Path) -> None:
+    """The CHECK constraint on a fresh DB rejects an out-of-range kind."""
+    db = tmp_path / "db.sqlite"
+    init_db(db)
+    conn = connect(db)
+    try:
+        with pytest.raises(sqlite3.IntegrityError):
+            conn.execute(
+                "INSERT INTO pending_questions "
+                "(question_id, thread_id, turn, status, transport, kind) "
+                "VALUES ('bad', 't', 0, 'open', 'slack', 'bogus')"
+            )
+    finally:
+        conn.close()