STATE-WP-0069 T04/T06: body metering and workplan-first state internals

Share LegacyWorkstreamIdBodyMixin across create schemas; meter POST /tasks/
and /decisions/ workstream_id bodies. State summary uses workplan flow;
NextStep dual-writes workplan_* fields alongside legacy workstream_*.

api/routers/decisions.py

@@ -15,7 +15,7 @@ from api.events import EventEnvelope, publish_event
 from api.models.decision import Decision, DecisionStatus, DecisionType
 from api.models.progress_event import ProgressEvent
 from api.schemas.decision import DecisionCreate, DecisionRead, DecisionResolve, DecisionUpdate
-from api.services.legacy_compat import meter_legacy_query_param
+from api.services.legacy_compat import meter_legacy_body_from_model, meter_legacy_query_param
 
 router = APIRouter(prefix="/decisions", tags=["decisions"])
 
@@ -75,9 +75,20 @@ async def list_decisions(
 
 @router.post("/", response_model=DecisionRead, status_code=status.HTTP_201_CREATED)
 async def create_decision(
+    request: Request,
+    response: Response,
     body: DecisionCreate,
     session: AsyncSession = Depends(get_session),
 ) -> Decision:
+    await meter_legacy_body_from_model(
+        body,
+        session=session,
+        request=request,
+        response=response,
+        method="POST",
+        route="/decisions/",
+        replacement_ref="POST /decisions/ with workplan_id",
+    )
     data = body.model_dump()
     note = _needs_escalation(body)
     if note:

api/routers/state.py

@@ -53,6 +53,22 @@ from api.services.summary_cache import (
     get_summary_cache,
     register_summary_cache_invalidation,
 )
+
+
+def _dual_workplan_refs(
+    workplan_id,
+    title: str | None,
+    slug: str | None,
+) -> dict:
+    """Dual-write workplan-first and legacy workstream fields for NextStep."""
+    return {
+        "workplan_id": workplan_id,
+        "workplan_title": title,
+        "workplan_slug": slug,
+        "workstream_id": workplan_id,
+        "workstream_title": title,
+        "workstream_slug": slug,
+    }
 from api.task_status import TERMINAL_TASK_STATUSES, status_value
 from api.workplan_status import (
     CLOSED_WORKPLAN_STATUSES,
@@ -183,7 +199,7 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
     )
     open_ws = list(open_ws_rows.scalars().all())
 
-    # Task counts per workstream (used to enrich open_workstreams)
+    # Task counts per workplan (used to enrich open_workplans / open_workstreams)
     task_per_ws: dict = {}
     task_statuses_per_ws: dict = {}
     for ws_id, tstat, cnt in await session.execute(
@@ -260,7 +276,7 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
                 description=d.description,
             ))
 
-    workstream_flow = load_flow("workstream")
+    workplan_flow = load_flow("workplan")
     flow_engine = FlowEngine()
     effective_status: dict = {}
     blocked_reasons: dict = {}
@@ -275,7 +291,7 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
                 if d.from_workplan_id == w.id and d.to_workplan_id and d.to_workplan_id in ws_lookup
             ],
         }
-        flow_result = flow_engine.evaluate(flow_obj, workstream_flow)
+        flow_result = flow_engine.evaluate(flow_obj, workplan_flow)
         effective_status[w.id] = "blocked" if flow_result.exit_blocked else w.status
         blocked_reasons[w.id] = [
             assertion_result_to_dict(item) for item in flow_result.blocking_assertions
@@ -529,7 +545,7 @@ async def _build_dashboard_overview(session: AsyncSession) -> DashboardOverview:
         dep_rows = list(dep_result.scalars().all())
 
     ws_lookup = {w.id: w for w in workstreams_all}
-    workstream_flow = load_flow("workstream")
+    workplan_flow = load_flow("workplan")
     flow_engine = FlowEngine()
     effective_status: dict = {}
     for w in open_ws:
@@ -543,7 +559,7 @@ async def _build_dashboard_overview(session: AsyncSession) -> DashboardOverview:
                 if d.from_workplan_id == w.id and d.to_workplan_id and d.to_workplan_id in ws_lookup
             ],
         }
-        flow_result = flow_engine.evaluate(flow_obj, workstream_flow)
+        flow_result = flow_engine.evaluate(flow_obj, workplan_flow)
         effective_status[w.id] = "blocked" if flow_result.exit_blocked else normalize_workplan_status(w.status)
 
     topic_counts = {r[0]: r[1] for r in await session.execute(
@@ -923,13 +939,15 @@ 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("*")])
-        domain_slug = await _get_domain_slug_for_workstream(ws, session)
+        domain_slug = await _get_domain_slug_for_workplan(ws, session)
         steps.append(NextStep(
             type="resolved_decision",
             domain=domain_slug,
-            workstream_id=ws.id if ws else None,
-            workstream_title=ws.title if ws else None,
-            workstream_slug=ws.slug if ws else None,
+            **_dual_workplan_refs(
+                ws.id if ws else None,
+                ws.title if ws else None,
+                ws.slug if ws else None,
+            ),
             task_id=task.id,
             task_title=task.title,
             message=(
@@ -1015,9 +1033,7 @@ async def _derive_next_steps(session: AsyncSession) -> tuple[list[NextStep], lis
         steps.append(NextStep(
             type="dependency_cleared",
             domain=domain_slug,
-            workstream_id=from_ws_id,
-            workstream_title=from_ws["title"],
-            workstream_slug=from_ws["slug"],
+            **_dual_workplan_refs(from_ws_id, from_ws["title"], from_ws["slug"]),
             task_id=task.id,
             task_title=task.title,
             message=(
@@ -1040,9 +1056,7 @@ async def _derive_next_steps(session: AsyncSession) -> tuple[list[NextStep], lis
         steps.append(NextStep(
             type="open_suggestion",
             domain=suggestion.domain_slug,
-            workstream_id=suggestion.workplan_id,
-            workstream_title=None,
-            workstream_slug=suggestion.origin_ref,
+            **_dual_workplan_refs(suggestion.workplan_id, None, suggestion.origin_ref),
             task_id=None,
             task_title=None,
             message=(
@@ -1054,8 +1068,8 @@ async def _derive_next_steps(session: AsyncSession) -> tuple[list[NextStep], lis
     return steps, open_suggestions
 
 
-async def _get_domain_slug_for_workstream(ws: Workplan | None, session: AsyncSession) -> str | None:
-    """Get the domain slug for a workstream via its topic."""
+async def _get_domain_slug_for_workplan(ws: Workplan | None, session: AsyncSession) -> str | None:
+    """Get the domain slug for a workplan via its topic."""
     if ws is None or ws.topic_id is None:
         return None
     return await _get_domain_slug_for_topic(ws.topic_id, session)

api/routers/tasks.py

@@ -18,7 +18,7 @@ from api.schemas.task import (
     TaskStatusBulkSyncRead,
     TaskUpdate,
 )
-from api.services.legacy_compat import meter_legacy_query_param
+from api.services.legacy_compat import meter_legacy_body_from_model, meter_legacy_query_param
 from api.services.lifecycle import status_value, transition_task_status
 from api.task_status import normalize_task_status
 
@@ -105,9 +105,20 @@ async def count_tasks(
 
 @router.post("/", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
 async def create_task(
+    request: Request,
+    response: Response,
     body: TaskCreate,
     session: AsyncSession = Depends(get_session),
 ) -> Task:
+    await meter_legacy_body_from_model(
+        body,
+        session=session,
+        request=request,
+        response=response,
+        method="POST",
+        route="/tasks/",
+        replacement_ref="POST /tasks/ with workplan_id",
+    )
     task = Task(**body.model_dump())
     session.add(task)
     if status_value(task.status) == "progress":

api/schemas/compat.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import uuid
+from typing import Any
 
 from pydantic import AliasChoices, Field, computed_field, model_validator
 
@@ -13,6 +14,20 @@ def workplan_id_field(*, default: uuid.UUID | None = None) -> uuid.UUID | None:
     )
 
 
+class LegacyWorkstreamIdBodyMixin:
+    """Detect POST/PUT bodies that used legacy ``workstream_id`` without ``workplan_id``."""
+
+    used_legacy_workstream_id: bool = Field(default=False, exclude=True)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _detect_legacy_workstream_id(cls, data: Any) -> Any:
+        if isinstance(data, dict):
+            if data.get("workstream_id") is not None and data.get("workplan_id") is None:
+                return {**data, "used_legacy_workstream_id": True}
+        return data
+
+
 class WorkplanIdCompatMixin:
     """Accept ``workplan_id`` or legacy ``workstream_id`` on input; emit both on output."""
 
@@ -24,7 +39,7 @@ class WorkplanIdCompatMixin:
         return self.workplan_id
 
 
-class WorkplanIdCreateMixin:
+class WorkplanIdCreateMixin(LegacyWorkstreamIdBodyMixin):
     workplan_id: uuid.UUID | None = workplan_id_field(default=None)
 
     @model_validator(mode="after")

api/schemas/decision.py

@@ -4,11 +4,11 @@ from datetime import datetime
 from pydantic import BaseModel, ConfigDict, model_validator
 
 from api.models.decision import DecisionStatus, DecisionType
-from api.schemas.compat import OptionalWorkplanIdCompatMixin
+from api.schemas.compat import LegacyWorkstreamIdBodyMixin, OptionalWorkplanIdCompatMixin
 from pydantic import AliasChoices, Field
 
 
-class DecisionCreate(BaseModel):
+class DecisionCreate(LegacyWorkstreamIdBodyMixin, BaseModel):
     topic_id: uuid.UUID | None = None
     workplan_id: uuid.UUID | None = Field(
         default=None,

api/schemas/progress_event.py

@@ -2,18 +2,17 @@ import uuid
 from datetime import datetime
 from typing import Any
 
-from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
+from pydantic import AliasChoices, BaseModel, ConfigDict, Field
 
-from api.schemas.compat import OptionalWorkplanIdCompatMixin
+from api.schemas.compat import LegacyWorkstreamIdBodyMixin, OptionalWorkplanIdCompatMixin
 
 
-class ProgressEventCreate(BaseModel):
+class ProgressEventCreate(LegacyWorkstreamIdBodyMixin, BaseModel):
     topic_id: uuid.UUID | None = None
     workplan_id: uuid.UUID | None = Field(
         default=None,
         validation_alias=AliasChoices("workplan_id", "workstream_id"),
     )
-    used_legacy_workstream_id: bool = Field(default=False, exclude=True)
     task_id: uuid.UUID | None = None
     decision_id: uuid.UUID | None = None
     event_type: str
@@ -22,14 +21,6 @@ class ProgressEventCreate(BaseModel):
     author: str | None = None
     session_id: str | None = None
 
-    @model_validator(mode="before")
-    @classmethod
-    def _detect_legacy_workstream_id(cls, data: Any) -> Any:
-        if isinstance(data, dict):
-            if data.get("workstream_id") is not None and data.get("workplan_id") is None:
-                return {**data, "used_legacy_workstream_id": True}
-        return data
-
 
 class ProgressEventRead(OptionalWorkplanIdCompatMixin, BaseModel):
     model_config = ConfigDict(from_attributes=True)

api/schemas/state.py

@@ -64,6 +64,9 @@ class NextStep(BaseModel):
     """
     type: str          # unblocked_task | resolved_decision | dependency_cleared
     domain: str | None = None
+    workplan_id: uuid.UUID | None = None
+    workplan_title: str | None = None
+    workplan_slug: str | None = None
     workstream_id: uuid.UUID | None = None
     workstream_title: str | None = None
     workstream_slug: str | None = None

api/services/legacy_compat.py

@@ -84,4 +84,28 @@ async def meter_legacy_body_field(
         )
     except Exception:
         await session.rollback()
-        logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)
+        logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)
+
+
+async def meter_legacy_body_from_model(
+    body: object,
+    *,
+    session: AsyncSession,
+    request: Request | None,
+    response: Response | None,
+    method: str,
+    route: str,
+    replacement_ref: str,
+    field: str = "workstream_id",
+) -> None:
+    if not getattr(body, "used_legacy_workstream_id", False):
+        return
+    await meter_legacy_body_field(
+        session=session,
+        request=request,
+        response=response,
+        method=method,
+        route=route,
+        replacement_ref=replacement_ref,
+        field=field,
+    )