fix(alembic/009): make enum reconcile dynamic — handles ALL postgres enums, not just 3

Original 009 only reconciled agentrole/team/taskstatus. Production hit LookupError: 'CELL' is not among defined channeltype values during seed bootstrap — channeltype (and ~16 other enums declared in 001) had uppercase members from the original create_all bootstrap that 009 never touched.

Rewrite queries pg_enum at migration time to find every enum with uppercase members, then for each: read current members, lowercase them, add desired_additions for the 3 enums with new ORM members, find every (table, column) referencing the enum via pg_attribute, RENAME-old/CREATE-new/ALTER-USING-lower/DROP-old.

alembic/versions/009_enum_reconcile.py

@@ -1,19 +1,23 @@
-"""Reconcile enum values with the ORM and add missing members.
+"""Reconcile every postgres enum with the ORM (lowercase) + add new members.
 
 Two adjustments to bring postgres enum types in line with the StrEnum
 classes the ORM serializes:
 
-1. Add missing values that were introduced after migration 001:
+1. Add missing members that were introduced after migration 001:
-   - agentrole.system  (used internally for orchestrator-owned operations)
+   - agentrole.system        (orchestrator-owned operations)
-   - team.fullstack    (cross-team work)
+   - team.fullstack          (cross-team work)
-   - taskstatus.quarantined (safe-park state for problematic tasks)
+   - taskstatus.quarantined  (safe-park state for problematic tasks)
 
-2. If the database was previously bootstrapped via Base.metadata.create_all
+2. If any enum was previously bootstrapped via Base.metadata.create_all
-   (which uses the StrEnum member NAME — uppercase), reconcile the enum
+   (which uses the StrEnum member NAME — uppercase), reconcile every
-   values to lowercase so they match alembic 001's declared values and
+   affected enum to lowercase so values match alembic 001's declared
-   the new ORM (Enum(..., values_callable=...)) serialization. This is a
+   members and the new ORM serialization (Enum(..., values_callable=...)).
-   conditional rebuild: if the enum already has lowercase members the
+
-   block is a no-op.
+The reconcile path is dynamic: any enum found with uppercase members is
+rebuilt by renaming the old type, creating a fresh lowercase type with
+the same members (lower-cased) plus any desired additions, ALTER-ing
+every column that references the type with `USING lower(col::text)::T`,
+then dropping the old type. Repeats per affected enum.
 
 Revision ID: 009_enum_reconcile
 Revises: 008_align_skills
@@ -22,139 +26,127 @@ Create Date: 2026-05-02
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
 from alembic import op
 from sqlalchemy import text
 
-if TYPE_CHECKING:
-    from sqlalchemy.sql.elements import TextClause
-
 revision = "009_enum_reconcile"
 down_revision = "008_align_skills"
 branch_labels = None
 depends_on = None
 
 
-# Per-enum desired member set (matches the StrEnum `.value` lists in
+# Members the ORM defines that may not be in the alembic-declared enum.
-# roboco/models/base.py + roboco/models/a2a.py + work_session.py).
+# Every value here is a literal already used by the ORM today; adding
-_DESIRED: dict[str, tuple[str, ...]] = {
+# them to the postgres enum is what unblocks the next-write path.
-    "agentrole": (
+_DESIRED_ADDITIONS: dict[str, tuple[str, ...]] = {
-        "system",
+    "agentrole": ("system",),
-        "ceo",
+    "team": ("fullstack",),
-        "product_owner",
+    "taskstatus": ("quarantined",),
-        "head_marketing",
-        "auditor",
-        "main_pm",
-        "cell_pm",
-        "developer",
-        "qa",
-        "documenter",
-    ),
-    "team": (
-        "backend",
-        "frontend",
-        "ux_ui",
-        "fullstack",
-        "main_pm",
-        "board",
-        "marketing",
-    ),
-    "taskstatus": (
-        "backlog",
-        "pending",
-        "claimed",
-        "in_progress",
-        "blocked",
-        "paused",
-        "verifying",
-        "needs_revision",
-        "awaiting_qa",
-        "awaiting_documentation",
-        "awaiting_pm_review",
-        "awaiting_ceo_approval",
-        "completed",
-        "cancelled",
-        "quarantined",
-    ),
 }
 
 
-# (enum_name, table_name, column_name) — used by the conditional rebuild.
-_USAGES: tuple[tuple[str, str, str], ...] = (
-    ("agentrole", "agents", "role"),
-    ("team", "agents", "team"),
-    ("team", "tasks", "team"),
-    ("team", "projects", "assigned_cell"),
-    ("taskstatus", "tasks", "status"),
-)
-
-
 def upgrade() -> None:
-    """Add missing values; rebuild if uppercase drift is detected."""
+    """Add missing values; rebuild any enum found with uppercase members."""
     bind = op.get_bind()
 
-    # Step 1: detect drift. If any enum has uppercase members, the DB was
+    # Step 1: find every enum that has at least one uppercase member.
-    # bootstrapped via create_all and needs a full rebuild for ALL the
+    drifted = [
-    # enums we use. If everything is lowercase already, just ADD missing
+        row[0]
-    # values one by one (the cheap path).
+        for row in bind.execute(
-    drifted = bool(
+            text(
-        bind.execute(
+                """
-            _drift_query(),
+                SELECT DISTINCT t.typname
-        ).scalar()
+                FROM pg_enum e
-    )
+                JOIN pg_type t ON e.enumtypid = t.oid
+                WHERE e.enumlabel ~ '[A-Z]'
+                ORDER BY t.typname
+                """
+            )
+        )
+    ]
 
     if not drifted:
-        # Cheap path: ADD missing values per enum.
+        # Cheap path: add the missing values for known additions.
-        for enum_name, members in _DESIRED.items():
+        for enum_name, additions in _DESIRED_ADDITIONS.items():
-            for value in members:
+            for value in additions:
                 op.execute(f"ALTER TYPE {enum_name} ADD VALUE IF NOT EXISTS '{value}'")
         return
 
-    # Drift path: rebuild the affected enums. Each rebuild is a
+    # Step 2: rebuild each drifted enum.
-    # rename-old / create-new / alter-column / drop-old sequence. We
+    # For each enum:
-    # USING lower(col::text)::new_enum to convert uppercase data.
+    #   - read current members
-    for enum_name, members in _DESIRED.items():
+    #   - construct new member list as lowercase(current) plus desired_additions
+    #   - rename old, create new, ALTER every (table, column) using it,
+    #     drop old.
+    for enum_name in drifted:
+        current = [
+            row[0]
+            for row in bind.execute(
+                text(
+                    """
+                    SELECT e.enumlabel
+                    FROM pg_enum e
+                    JOIN pg_type t ON e.enumtypid = t.oid
+                    WHERE t.typname = :name
+                    ORDER BY e.enumsortorder
+                    """
+                ),
+                {"name": enum_name},
+            )
+        ]
+        new_members: list[str] = []
+        seen: set[str] = set()
+        for member in current:
+            lowered = member.lower()
+            if lowered not in seen:
+                new_members.append(lowered)
+                seen.add(lowered)
+        for addition in _DESIRED_ADDITIONS.get(enum_name, ()):
+            if addition not in seen:
+                new_members.append(addition)
+                seen.add(addition)
+
+        usages = [
+            (row[0], row[1])
+            for row in bind.execute(
+                text(
+                    """
+                    SELECT n.nspname || '.' || c.relname AS table_qualified, a.attname
+                    FROM pg_attribute a
+                    JOIN pg_class c ON a.attrelid = c.oid
+                    JOIN pg_namespace n ON c.relnamespace = n.oid
+                    JOIN pg_type t ON a.atttypid = t.oid
+                    WHERE t.typname = :name
+                      AND a.attnum > 0
+                      AND NOT a.attisdropped
+                      AND c.relkind = 'r'
+                    """
+                ),
+                {"name": enum_name},
+            )
+        ]
+
         op.execute(f"ALTER TYPE {enum_name} RENAME TO {enum_name}_old")
-        members_sql = ", ".join(f"'{v}'" for v in members)
+        members_sql = ", ".join(f"'{v}'" for v in new_members)
         op.execute(f"CREATE TYPE {enum_name} AS ENUM ({members_sql})")
 
-    for enum_name, table, column in _USAGES:
+        for table_qualified, column in usages:
-        op.execute(
+            op.execute(
-            f"ALTER TABLE {table} "
+                f"ALTER TABLE {table_qualified} "
-            f"ALTER COLUMN {column} TYPE {enum_name} "
+                f"ALTER COLUMN {column} TYPE {enum_name} "
-            f"USING lower({column}::text)::{enum_name}"
+                f"USING lower({column}::text)::{enum_name}"
-        )
+            )
 
-    for enum_name in _DESIRED:
         op.execute(f"DROP TYPE {enum_name}_old")
 
 
 def downgrade() -> None:
-    """Remove the values added by upgrade.
+    """Intentional no-op.
 
-    Postgres has no DROP VALUE primitive; the only way to remove an enum
+    Postgres has no DROP VALUE primitive — removing an enum member would
-    member is the rebuild dance from upgrade. For the no-drift path we
+    require the same rebuild dance from upgrade. We can't recover the
-    simply leave the added values in place — they cause no harm and
+    prior uppercase shape after the data was already lowercased without
-    backing them out would be a destructive rebuild on healthy data. For
+    losing referential integrity, so we leave the reconciled state in
-    the drift path we cannot recover the prior uppercase values without
+    place.
-    losing referential integrity, so we likewise leave the rebuild in
-    place. This downgrade is therefore intentionally a no-op.
     """
     return None
-
-
-def _drift_query() -> TextClause:
-    """SELECT true iff any tracked enum has uppercase members."""
-    enum_list = ", ".join(f"'{name}'" for name in _DESIRED)
-    return text(
-        f"""
-        SELECT EXISTS (
-            SELECT 1
-            FROM pg_enum e
-            JOIN pg_type t ON e.enumtypid = t.oid
-            WHERE t.typname IN ({enum_list})
-              AND e.enumlabel ~ '[A-Z]'
-        )
-        """
-    )