From 388b3308090cf6d9da2400dc68875ee9691355c1 Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 8 Jul 2026 23:13:28 +0200 Subject: [PATCH] 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 | 13 ++++- api/routers/state.py | 48 ++++++++++++------- api/routers/tasks.py | 13 ++++- api/schemas/compat.py | 17 ++++++- api/schemas/decision.py | 4 +- api/schemas/progress_event.py | 15 ++---- api/schemas/state.py | 3 ++ api/services/legacy_compat.py | 26 +++++++++- ...n-terminology-legacy-retirement-backlog.md | 2 + tests/test_legacy_meter.py | 43 +++++++++++++++++ ...-workplan-terminology-legacy-retirement.md | 9 +++- 11 files changed, 157 insertions(+), 36 deletions(-) diff --git a/api/routers/decisions.py b/api/routers/decisions.py index 01d2811..9e777b7 100644 --- a/api/routers/decisions.py +++ b/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: diff --git a/api/routers/state.py b/api/routers/state.py index d18f0e5..9171ef0 100644 --- a/api/routers/state.py +++ b/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) diff --git a/api/routers/tasks.py b/api/routers/tasks.py index 1a69b5f..b13eb06 100644 --- a/api/routers/tasks.py +++ b/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": diff --git a/api/schemas/compat.py b/api/schemas/compat.py index 4338124..37de0b8 100644 --- a/api/schemas/compat.py +++ b/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") diff --git a/api/schemas/decision.py b/api/schemas/decision.py index a02cae6..09e5260 100644 --- a/api/schemas/decision.py +++ b/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, diff --git a/api/schemas/progress_event.py b/api/schemas/progress_event.py index 8086c5d..2f7c1e3 100644 --- a/api/schemas/progress_event.py +++ b/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) diff --git a/api/schemas/state.py b/api/schemas/state.py index 015b3f2..a3e5dfa 100644 --- a/api/schemas/state.py +++ b/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 diff --git a/api/services/legacy_compat.py b/api/services/legacy_compat.py index 66436c1..b47edd6 100644 --- a/api/services/legacy_compat.py +++ b/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) \ No newline at end of file + 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, + ) \ No newline at end of file diff --git a/docs/workplan-terminology-legacy-retirement-backlog.md b/docs/workplan-terminology-legacy-retirement-backlog.md index 13074d7..1c8cd48 100644 --- a/docs/workplan-terminology-legacy-retirement-backlog.md +++ b/docs/workplan-terminology-legacy-retirement-backlog.md @@ -76,6 +76,8 @@ These accept `workstream_id` alongside `workplan_id` on preferred routes: | `rest_api:GET /execution/launch-requests?workstream_id` | `GET /execution/launch-requests` | `GET /execution/launch-requests?workplan_id=` | | `rest_api:GET /progress/?workstream_id` | `GET /progress/` | `GET /progress/?workplan_id=` | | `rest_api:POST /progress/ workstream_id` | `POST /progress/` | `POST /progress/` with `workplan_id` body field | +| `rest_api:POST /tasks/ workstream_id` | `POST /tasks/` | `POST /tasks/` with `workplan_id` body field | +| `rest_api:POST /decisions/ workstream_id` | `POST /decisions/` | `POST /decisions/` with `workplan_id` body field | Retire param aliases in T04 after zero-usage windows; callers surface via `X-StateHub-Component` in weekly review. diff --git a/tests/test_legacy_meter.py b/tests/test_legacy_meter.py index 2108c3f..b80f448 100644 --- a/tests/test_legacy_meter.py +++ b/tests/test_legacy_meter.py @@ -234,6 +234,49 @@ class TestWorkplanAliasesAndLegacyMeter: summary = (await client.get("/legacy-meter/summary")).json() assert _summary_by_key(summary).get("rest_api:POST /progress/ workstream_id") is None + async def test_legacy_workstream_id_body_on_tasks_post_is_metered(self, client): + await _create_domain(client) + topic = await _create_topic(client) + wp = await _create_workplan(client, topic["id"]) + + r = await client.post( + "/tasks/", + json={ + "workstream_id": str(wp["id"]), + "title": "legacy task body", + "status": "todo", + }, + headers={"X-StateHub-Component": "old-task-writer"}, + ) + assert r.status_code == 201 + assert r.headers["Deprecation"] == "true" + + summary = (await client.get("/legacy-meter/summary")).json() + item = _summary_by_key(summary)["rest_api:POST /tasks/ workstream_id"] + assert item["window"]["calls"] == 1 + assert item["window"]["components"] == {"old-task-writer": 1} + + async def test_legacy_workstream_id_body_on_decisions_post_is_metered(self, client): + await _create_domain(client) + topic = await _create_topic(client) + wp = await _create_workplan(client, topic["id"]) + + r = await client.post( + "/decisions/", + json={ + "workstream_id": str(wp["id"]), + "title": "legacy decision body", + "decision_type": "pending", + }, + headers={"X-StateHub-Component": "old-decision-writer"}, + ) + assert r.status_code == 201 + assert r.headers["Deprecation"] == "true" + + summary = (await client.get("/legacy-meter/summary")).json() + item = _summary_by_key(summary)["rest_api:POST /decisions/ workstream_id"] + assert item["window"]["calls"] == 1 + async def test_workplan_id_query_param_on_tasks_is_not_metered(self, client): await _create_domain(client) topic = await _create_topic(client) diff --git a/workplans/STATE-WP-0069-workplan-terminology-legacy-retirement.md b/workplans/STATE-WP-0069-workplan-terminology-legacy-retirement.md index da5e3d9..b933b92 100644 --- a/workplans/STATE-WP-0069-workplan-terminology-legacy-retirement.md +++ b/workplans/STATE-WP-0069-workplan-terminology-legacy-retirement.md @@ -10,7 +10,7 @@ topic_slug: custodian planning_priority: medium planning_order: 69 created: "2026-07-08" -updated: "2026-07-13" +updated: "2026-07-14" state_hub_workstream_id: "923bb94a-d16c-422c-b81e-16328bd7b60c" --- @@ -189,6 +189,9 @@ Progress 2026-07-13 (T04): legacy responses now include `Sunset` (Jun 2027 plann horizon). POST `/progress/` meters `workstream_id` request bodies; hub-core progress router accepts optional body-meter hook. +Progress 2026-07-14 (T04): POST `/tasks/` and POST `/decisions/` meter legacy +`workstream_id` request bodies via `LegacyWorkstreamIdBodyMixin`. + ## Task: Legacy completion event — stop dual-publish ```task @@ -236,6 +239,10 @@ Progress 2026-07-09 (T06): added `flows/workplan.yaml` (`custodian.workplan.v1`) first with `open_workstreams` fallback. Legacy `flows/workstream.yaml` and summary dual-key remain until legacy-meter clears callers. +Progress 2026-07-14 (T06): state summary flow evaluation uses `workplan` flow; +`NextStep` dual-writes `workplan_*` alongside legacy `workstream_*` fields. +`LegacyWorkstreamIdBodyMixin` shared across create schemas. + ## Task: Closeout — registry cleanup and verification ```task