Index workplan flavor and omit residuals from default views.

STATE-WP-0092: persist flavor from files, treat depends_on as the C-20
canonical edge, exclude flavor=residual from summary/next_steps/deps
unless include_residuals is set. Live primary still needs the alembic
revision applied.

Assistant: grok
Assistant-Session: 01a09dc1-b21e-77e1-919e-fcad2f82b267

api/models/task.py

@@ -42,6 +42,7 @@ class Task(Base, TimestampMixin):
     # title, so renaming a heading looks like one task vanishing and another
     # appearing.
     record_id: Mapped[str | None] = mapped_column(String(120), nullable=True, index=True)
+    flavor: Mapped[str | None] = mapped_column(String(32), nullable=True, index=True)
     title: Mapped[str] = mapped_column(String(255), nullable=False)
     description: Mapped[str | None] = mapped_column(Text, nullable=True)
     status: Mapped[TaskStatus] = mapped_column(

api/models/workplan.py

@@ -39,6 +39,9 @@ class Workplan(Base, TimestampMixin):
     queue_rank: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True)
     execution_group: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True)
     scheduled_for: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+    flavor: Mapped[str | None] = mapped_column(String(32), nullable=True, index=True)
+    flavor_promotion_reason: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    flavor_promoted_from: Mapped[str | None] = mapped_column(String(32), nullable=True)
 
     repo_id: Mapped[uuid.UUID] = mapped_column(
         UUID(as_uuid=True),

api/routers/state.py

@@ -1,7 +1,7 @@
 import time
 from datetime import datetime, timedelta, timezone
 
-from fastapi import APIRouter, Depends, Request, Response
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
 from fastapi.responses import JSONResponse
 from sqlalchemy import func, select, text
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -75,6 +75,12 @@ def _dual_workplan_refs(
         "workstream_slug": slug,
     }
 from api.task_status import TERMINAL_TASK_STATUSES, status_value
+from api.work_record_flavor import (
+    RESIDUAL_FLAVOR,
+    WORK_RECORD_FLAVORS,
+    is_residual_flavor,
+    normalize_flavor,
+)
 from api.workplan_status import (
     CLOSED_WORKPLAN_STATUSES,
     OPEN_WORKPLAN_STATUSES,
@@ -100,12 +106,44 @@ def _summary_cache_headers(
     response.headers["Cache-Control"] = "max-age=15, stale-while-revalidate=120"
 
 
+def _apply_summary_flavor_view(
+    summary: StateSummary,
+    *,
+    include_residuals: bool,
+    flavor: str | None,
+) -> StateSummary:
+    """Default views omit residual workplans; totals.residual_open stays honest."""
+    wanted = normalize_flavor(flavor)
+    if wanted is not None and wanted not in WORK_RECORD_FLAVORS:
+        raise HTTPException(
+            status_code=422,
+            detail=(
+                f"Unknown work-record flavor {wanted!r}; "
+                f"expected one of {', '.join(WORK_RECORD_FLAVORS)}"
+            ),
+        )
+    show_residuals = include_residuals or wanted == RESIDUAL_FLAVOR
+    residual_rows = list(summary.residual_open_workplans)
+    open_rows = list(summary.open_workplans)
+    if wanted is not None:
+        open_rows = [row for row in open_rows if row.flavor == wanted]
+        residual_rows = [row for row in residual_rows if row.flavor == wanted]
+    return summary.model_copy(
+        update={
+            "open_workplans": open_rows,
+            "residual_open_workplans": residual_rows if show_residuals else [],
+        }
+    )
+
+
 @router.get("/summary", response_model=StateSummary)
 async def get_summary(
     request: Request,
     response: Response,
     session: AsyncSession = Depends(get_session),
     refresh: bool = False,
+    include_residuals: bool = Query(False),
+    flavor: str | None = Query(None),
 ) -> StateSummary:
     revision = await fetch_summary_revision(session)
     revision_token = revision.combined_fingerprint()
@@ -116,22 +154,38 @@ async def get_summary(
 
     if cache_status == "hit-revision" and cached is not None:
         _summary_cache_headers(response, cache_status="hit-revision", revision=revision_token)
-        return cached.model_copy(update={"ops_runs": await get_ops_run_projection()})
+        return _apply_summary_flavor_view(
+            cached.model_copy(update={"ops_runs": await get_ops_run_projection()}),
+            include_residuals=include_residuals,
+            flavor=flavor,
+        )
 
     if cache_status == "progress-section" and cached is not None:
         result = await apply_progress_section(session, cached, revision)
         _summary_cache_headers(response, cache_status="hit-revision", revision=revision_token)
-        return result.model_copy(update={"ops_runs": await get_ops_run_projection()})
+        return _apply_summary_flavor_view(
+            result.model_copy(update={"ops_runs": await get_ops_run_projection()}),
+            include_residuals=include_residuals,
+            flavor=flavor,
+        )
 
     if cache_status == "stale" and cached is not None:
         cache.schedule_refresh(revision)
         _summary_cache_headers(response, cache_status="stale", revision=revision_token)
-        return cached.model_copy(update={"ops_runs": await get_ops_run_projection()})
+        return _apply_summary_flavor_view(
+            cached.model_copy(update={"ops_runs": await get_ops_run_projection()}),
+            include_residuals=include_residuals,
+            flavor=flavor,
+        )
 
     result = await build_state_summary(session)
     cache.store(result, revision)
     _summary_cache_headers(response, cache_status="miss", revision=revision_token)
-    return result.model_copy(update={"ops_runs": await get_ops_run_projection(refresh=force_refresh)})
+    return _apply_summary_flavor_view(
+        result.model_copy(update={"ops_runs": await get_ops_run_projection(refresh=force_refresh)}),
+        include_residuals=include_residuals,
+        flavor=flavor,
+    )
 
 
 async def build_state_summary(session: AsyncSession) -> StateSummary:
@@ -336,6 +390,7 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
             ),
             archived=ws_counts.get("archived", 0),
             total=sum(ws_counts.values()),
+            residual_open=0,
         ),
         tasks=TaskTotals(
             wait=task_counts.get(TaskStatus.wait, 0),
@@ -397,6 +452,15 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
         )
         for w in open_ws
     ]
+    relevant_rows = [
+        row for row, w in zip(open_workplan_rows, open_ws)
+        if not is_residual_flavor(w.flavor)
+    ]
+    residual_rows = [
+        row for row, w in zip(open_workplan_rows, open_ws)
+        if is_residual_flavor(w.flavor)
+    ]
+    totals.workstreams.residual_open = len(residual_rows)
 
     result = StateSummary(
         generated_at=datetime.now(tz=timezone.utc),
@@ -418,7 +482,8 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
         licence_risk_count=licence_risk_count,
         open_capability_requests=open_cap_req_count,
         ranked_suggestions=ranked_suggestions,
-        open_workplans=open_workplan_rows,
+        open_workplans=relevant_rows,
+        residual_open_workplans=residual_rows,
     )
     return result
 
@@ -521,6 +586,7 @@ async def _build_dashboard_overview(session: AsyncSession) -> DashboardOverview:
         w for w in workstreams_all
         if normalize_workplan_status(w.status) in OPEN_WORKPLAN_STATUSES
     ]
+    residual_open = sum(1 for w in open_ws if is_residual_flavor(w.flavor))
     open_ws_ids = [w.id for w in open_ws]
     dep_rows = []
     if open_ws_ids:
@@ -580,6 +646,7 @@ async def _build_dashboard_overview(session: AsyncSession) -> DashboardOverview:
             ),
             archived=ws_counts.get("archived", 0),
             total=sum(ws_counts.values()),
+            residual_open=residual_open,
         ),
         tasks=TaskTotals(
             wait=task_totals_by_status.get("wait", 0),
@@ -668,6 +735,7 @@ async def _build_dashboard_overview(session: AsyncSession) -> DashboardOverview:
             id=w.id,
             title=w.title,
             status=normalize_workplan_status(w.status),
+            flavor=w.flavor,
             domain=repo["domain_slug"] if repo else (topic.domain_slug if topic else "unknown"),
             repo_label=repo["slug"] if repo else workplan.get("repo_slug", "unassigned"),
             workplan_filename=workplan.get("filename"),
@@ -795,7 +863,10 @@ async def _build_domain_summaries(session: AsyncSession) -> list[DomainSummary]:
 
 
 @router.get("/deps", response_model=list[WorkstreamWithDeps])
-async def get_deps(session: AsyncSession = Depends(get_session)) -> list[WorkstreamWithDeps]:
+async def get_deps(
+    session: AsyncSession = Depends(get_session),
+    include_residuals: bool = Query(False),
+) -> list[WorkstreamWithDeps]:
     """Lightweight dep-graph endpoint: open workstreams with their dependency edges only.
 
     Returns the same structure as open_workplans in /state/summary but skips
@@ -809,6 +880,8 @@ async def get_deps(session: AsyncSession = Depends(get_session)) -> list[Workstr
         .order_by(Workplan.due_date.asc().nullslast(), Workplan.created_at)
     )
     open_ws = list(open_ws_rows.scalars().all())
+    if not include_residuals:
+        open_ws = [w for w in open_ws if not is_residual_flavor(w.flavor)]
 
     open_ws_ids = [w.id for w in open_ws]
     dep_rows = []
@@ -944,6 +1017,10 @@ async def _derive_next_steps(session: AsyncSession) -> tuple[list[NextStep], lis
         if task.id in seen_task_ids:
             continue
         ws = await session.get(Workplan, decision.workplan_id, options=[noload("*")])
+        if ws is not None and is_residual_flavor(ws.flavor):
+            continue
+        if is_residual_flavor(task.flavor):
+            continue
         domain_slug = await _get_domain_slug_for_workplan(ws, session)
         steps.append(NextStep(
             type="resolved_decision",
@@ -988,6 +1065,7 @@ async def _derive_next_steps(session: AsyncSession) -> tuple[list[NextStep], lis
                 Workplan.title,
                 Workplan.slug,
                 Workplan.topic_id,
+                Workplan.flavor,
             ).where(Workplan.id.in_(dep_ws_ids))
         )
         ws_info = {
@@ -996,8 +1074,9 @@ async def _derive_next_steps(session: AsyncSession) -> tuple[list[NextStep], lis
                 "title": title,
                 "slug": slug,
                 "topic_id": topic_id,
+                "flavor": flavor,
             }
-            for ws_id, status, title, slug, topic_id in ws_rows
+            for ws_id, status, title, slug, topic_id, flavor in ws_rows
         }
 
     ready_from_ws_ids = [
@@ -1023,12 +1102,16 @@ async def _derive_next_steps(session: AsyncSession) -> tuple[list[NextStep], lis
 
     for from_ws_id in ready_from_ws_ids:
         from_ws = ws_info.get(from_ws_id, {})
+        if is_residual_flavor(from_ws.get("flavor")):
+            continue
         todo_tasks = todo_by_ws.get(from_ws_id, [])
         if not todo_tasks:
             continue
         task = min(todo_tasks, key=lambda t: (_PRIORITY_RANK.get(t.priority, 99), t.created_at))
         if task.id in seen_task_ids:
             continue
+        if is_residual_flavor(task.flavor):
+            continue
         domain_slug = await _get_domain_slug_for_topic(from_ws.get("topic_id"), session)
         _blocker_slugs = []
         for tid in dep_map[from_ws_id]:

api/routers/workstreams.py

@@ -9,7 +9,7 @@ from typing import Any
 
 import yaml
 from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
-from sqlalchemy import select
+from sqlalchemy import or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.database import get_session
@@ -24,6 +24,11 @@ from api.schemas.workplan import (
 )
 from api.services.lifecycle import transition_workplan_status
 from api.services.legacy_compat import retire_legacy_route
+from api.work_record_flavor import (
+    RESIDUAL_FLAVOR,
+    WORK_RECORD_FLAVORS,
+    normalize_flavor,
+)
 from api.workplan_status import (
     is_supported_workplan_status,
     normalize_workplan_status,
@@ -88,6 +93,8 @@ async def _list_workplans(
     status_filter: str | None,
     owner: str | None,
     slug: str | None,
+    flavor: str | None = None,
+    include_residuals: bool = True,
     session: AsyncSession,
 ) -> list[Workplan]:
     q = select(Workplan)
@@ -106,6 +113,19 @@ async def _list_workplans(
         q = q.where(Workplan.owner == owner)
     if slug:
         q = q.where(Workplan.slug == slug)
+    wanted = normalize_flavor(flavor)
+    if wanted is not None:
+        if wanted not in WORK_RECORD_FLAVORS:
+            raise HTTPException(
+                status_code=422,
+                detail=(
+                    f"Unknown work-record flavor {wanted!r}; "
+                    f"expected one of {', '.join(WORK_RECORD_FLAVORS)}"
+                ),
+            )
+        q = q.where(Workplan.flavor == wanted)
+    elif not include_residuals:
+        q = q.where(or_(Workplan.flavor.is_(None), Workplan.flavor != RESIDUAL_FLAVOR))
     q = q.order_by(
         Workplan.planning_priority.asc().nullslast(),
         Workplan.planning_order.asc().nullslast(),
@@ -365,6 +385,8 @@ async def list_workplans(
     status: str | None = None,
     owner: str | None = None,
     slug: str | None = None,
+    flavor: str | None = Query(None),
+    include_residuals: bool = Query(True),
     session: AsyncSession = Depends(get_session),
 ) -> list[Workplan]:
     return await _list_workplans(
@@ -374,6 +396,8 @@ async def list_workplans(
         status_filter=status,
         owner=owner,
         slug=slug,
+        flavor=flavor,
+        include_residuals=include_residuals,
         session=session,
     )
 

api/schemas/state.py

@@ -30,6 +30,7 @@ class WorkstreamTotals(BaseModel):
     finished: int = 0
     archived: int = 0
     total: int = 0
+    residual_open: int = 0
 
 
 class TaskTotals(BaseModel):
@@ -85,6 +86,7 @@ class StateSummary(BaseModel):
     blocked_tasks: list[TaskRead] = []
     recent_progress: list[ProgressEventRead]
     open_workplans: list[WorkstreamWithDeps] = []
+    residual_open_workplans: list[WorkstreamWithDeps] = []
     next_steps: list[NextStep] = []
     domains: list[DomainSummary] = []
     contribution_counts: dict[str, int] = {}
@@ -98,6 +100,7 @@ class DashboardWorkplanRow(BaseModel):
     id: uuid.UUID
     title: str
     status: str
+    flavor: str | None = None
     domain: str = "unknown"
     repo_label: str = "unassigned"
     workplan_filename: str | None = None

api/schemas/task.py

@@ -7,6 +7,7 @@ from pydantic import BaseModel, ConfigDict, field_validator, model_validator
 from api.models.task import TaskPriority, TaskStatus
 from api.schemas.compat import WorkplanIdCompatMixin, WorkplanIdCreateMixin
 from api.task_status import normalize_task_status
+from api.work_record_flavor import WORK_RECORD_FLAVORS, normalize_flavor
 
 
 class TaskStatusMixin(BaseModel):
@@ -18,9 +19,23 @@ class TaskStatusMixin(BaseModel):
         return normalize_task_status(value)
 
 
-class TaskCreate(TaskStatusMixin, WorkplanIdCreateMixin):
+class TaskFlavorMixin(BaseModel):
+    @field_validator("flavor", mode="before", check_fields=False)
+    @classmethod
+    def _normalize_flavor(cls, value):
+        flavor = normalize_flavor(value)
+        if flavor is not None and flavor not in WORK_RECORD_FLAVORS:
+            raise ValueError(
+                f"Unknown work-record flavor {flavor!r}; "
+                f"expected one of {', '.join(WORK_RECORD_FLAVORS)}"
+            )
+        return flavor
+
+
+class TaskCreate(TaskStatusMixin, TaskFlavorMixin, WorkplanIdCreateMixin):
     id: uuid.UUID | None = None
     record_id: str | None = None
+    flavor: str | None = None
     title: str
     description: str | None = None
     status: TaskStatus = TaskStatus.todo
@@ -39,8 +54,9 @@ class TaskCreate(TaskStatusMixin, WorkplanIdCreateMixin):
         return self
 
 
-class TaskUpdate(TaskStatusMixin):
+class TaskUpdate(TaskStatusMixin, TaskFlavorMixin):
     title: str | None = None
+    flavor: str | None = None
     description: str | None = None
     status: TaskStatus | None = None
     priority: TaskPriority | None = None
@@ -102,6 +118,7 @@ class TaskRead(TaskStatusMixin, WorkplanIdCompatMixin):
     model_config = ConfigDict(from_attributes=True)
     id: uuid.UUID
     record_id: str | None = None
+    flavor: str | None = None
     title: str
     description: str | None = None
     status: TaskStatus

api/schemas/workplan.py

@@ -5,6 +5,12 @@ from typing import Literal
 from pydantic import BaseModel, ConfigDict, field_validator
 
 from api.schemas.workplan_dependency import WorkplanDepStub
+from api.work_record_flavor import (
+    FLAVOR_PROMOTION_REASONS,
+    WORK_RECORD_FLAVORS,
+    normalize_flavor,
+    normalize_promotion_reason,
+)
 from api.workplan_status import normalize_workplan_status
 
 WorkplanStatus = Literal[
@@ -28,7 +34,42 @@ class WorkplanStatusMixin(BaseModel):
         return normalize_workplan_status(value)
 
 
-class WorkplanCreate(WorkplanStatusMixin):
+class WorkplanFlavorMixin(BaseModel):
+    @field_validator("flavor", mode="before", check_fields=False)
+    @classmethod
+    def _normalise_flavor(cls, value):
+        flavor = normalize_flavor(value)
+        if flavor is not None and flavor not in WORK_RECORD_FLAVORS:
+            raise ValueError(
+                f"Unknown work-record flavor {flavor!r}; "
+                f"expected one of {', '.join(WORK_RECORD_FLAVORS)}"
+            )
+        return flavor
+
+    @field_validator("flavor_promotion_reason", mode="before", check_fields=False)
+    @classmethod
+    def _normalise_promotion_reason(cls, value):
+        reason = normalize_promotion_reason(value)
+        if reason is not None and reason not in FLAVOR_PROMOTION_REASONS:
+            raise ValueError(
+                f"Unknown flavor_promotion_reason {reason!r}; "
+                f"expected one of {', '.join(FLAVOR_PROMOTION_REASONS)}"
+            )
+        return reason
+
+    @field_validator("flavor_promoted_from", mode="before", check_fields=False)
+    @classmethod
+    def _normalise_promoted_from(cls, value):
+        flavor = normalize_flavor(value)
+        if flavor is not None and flavor not in WORK_RECORD_FLAVORS:
+            raise ValueError(
+                f"Unknown flavor_promoted_from {flavor!r}; "
+                f"expected one of {', '.join(WORK_RECORD_FLAVORS)}"
+            )
+        return flavor
+
+
+class WorkplanCreate(WorkplanStatusMixin, WorkplanFlavorMixin):
     id: uuid.UUID | None = None
     repo_id: uuid.UUID
     topic_id: uuid.UUID | None = None
@@ -40,6 +81,9 @@ class WorkplanCreate(WorkplanStatusMixin):
     due_date: date | None = None
     planning_priority: str | None = None
     planning_order: int | None = None
+    flavor: str | None = None
+    flavor_promotion_reason: str | None = None
+    flavor_promoted_from: str | None = None
     execution_state: ExecutionState = "manual"
     launch_mode: LaunchMode = "manual"
     concurrency_mode: ConcurrencyMode = "sequential"
@@ -49,7 +93,7 @@ class WorkplanCreate(WorkplanStatusMixin):
     repo_goal_id: uuid.UUID | None = None
 
 
-class WorkplanUpdate(WorkplanStatusMixin):
+class WorkplanUpdate(WorkplanStatusMixin, WorkplanFlavorMixin):
     title: str | None = None
     description: str | None = None
     status: WorkplanStatus | None = None
@@ -57,6 +101,9 @@ class WorkplanUpdate(WorkplanStatusMixin):
     due_date: date | None = None
     planning_priority: str | None = None
     planning_order: int | None = None
+    flavor: str | None = None
+    flavor_promotion_reason: str | None = None
+    flavor_promoted_from: str | None = None
     execution_state: ExecutionState | None = None
     launch_mode: LaunchMode | None = None
     concurrency_mode: ConcurrencyMode | None = None
@@ -95,6 +142,9 @@ class WorkplanRead(WorkplanStatusMixin):
     due_date: date | None = None
     planning_priority: str | None = None
     planning_order: int | None = None
+    flavor: str | None = None
+    flavor_promotion_reason: str | None = None
+    flavor_promoted_from: str | None = None
     execution_state: ExecutionState = "manual"
     launch_mode: LaunchMode = "manual"
     concurrency_mode: ConcurrencyMode = "sequential"

api/work_record_flavor.py

@@ -0,0 +1,48 @@
+"""Work-record flavor (STATE-WP-0092).
+
+Flavor is a closed bucket on workplans and tasks, orthogonal to kind and
+status. Unset flavor is not residual.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+WORK_RECORD_FLAVORS: tuple[str, ...] = (
+    "planning",
+    "implementation",
+    "refactoring",
+    "extension",
+    "residual",
+)
+RESIDUAL_FLAVOR = "residual"
+FLAVOR_PROMOTION_REASONS: tuple[str, ...] = ("demand", "risk")
+
+_EMPTY = {"", "~", "null", "none", "nil"}
+
+
+def normalize_flavor(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip().lower()
+    if text in _EMPTY:
+        return None
+    return text
+
+
+def is_known_flavor(value: Any) -> bool:
+    flavor = normalize_flavor(value)
+    return flavor is None or flavor in WORK_RECORD_FLAVORS
+
+
+def is_residual_flavor(value: Any) -> bool:
+    return normalize_flavor(value) == RESIDUAL_FLAVOR
+
+
+def normalize_promotion_reason(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip().lower()
+    if text in _EMPTY:
+        return None
+    return text