STATE-WP-0069 T04: meter workstream_id query-param aliases

Add legacy_compat helper and Deprecation headers when callers filter
/tasks, /decisions, /token-events, or /execution/launch-requests with
workstream_id. Preferred workplan_id filters are unchanged. Route removal
remains gated on legacy-meter zero-usage windows.

api/routers/decisions.py

@@ -4,7 +4,7 @@ import uuid
 from datetime import datetime, timezone
 from pathlib import Path
 
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
 
 logger = logging.getLogger(__name__)
 from sqlalchemy import select
@@ -15,6 +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
 
 router = APIRouter(prefix="/decisions", tags=["decisions"])
 
@@ -39,6 +40,8 @@ def _needs_escalation(body: DecisionCreate) -> str | None:
 
 @router.get("/", response_model=list[DecisionRead])
 async def list_decisions(
+    request: Request,
+    response: Response,
     topic_id: uuid.UUID | None = None,
     workplan_id: uuid.UUID | None = None,
     workstream_id: uuid.UUID | None = None,
@@ -46,6 +49,15 @@ async def list_decisions(
     decision_type: DecisionType | None = None,
     session: AsyncSession = Depends(get_session),
 ) -> list[Decision]:
+    if workstream_id is not None and workplan_id is None:
+        await meter_legacy_query_param(
+            session=session,
+            request=request,
+            response=response,
+            method="GET",
+            route="/decisions/",
+            replacement_ref="/decisions/?workplan_id=<workplan_id>",
+        )
     q = select(Decision)
     if topic_id:
         q = q.where(Decision.topic_id == topic_id)

api/routers/execution.py

@@ -28,6 +28,7 @@ from api.services.execution_queue import (
     workplan_blockers,
 )
 from api.routers.workstreams import _legacy_key, _meter_legacy_route
+from api.services.legacy_compat import meter_legacy_query_param
 from api.workplan_status import CLOSED_WORKPLAN_STATUSES, normalize_workplan_status
 
 router = APIRouter(prefix="/execution", tags=["execution"])
@@ -193,13 +194,26 @@ async def create_launch_request(
 
 @router.get("/launch-requests", response_model=list[LaunchRequestRead])
 async def list_launch_requests(
+    request: Request,
+    response: Response,
+    workplan_id: uuid.UUID | None = None,
     workstream_id: uuid.UUID | None = None,
     request_status: str | None = None,
     session: AsyncSession = Depends(get_session),
 ) -> list[WorkplanLaunchRequest]:
+    if workstream_id is not None and workplan_id is None:
+        await meter_legacy_query_param(
+            session=session,
+            request=request,
+            response=response,
+            method="GET",
+            route="/execution/launch-requests",
+            replacement_ref="/execution/launch-requests?workplan_id=<workplan_id>",
+        )
     q = select(WorkplanLaunchRequest).order_by(WorkplanLaunchRequest.created_at.desc())
-    if workstream_id:
-        q = q.where(WorkplanLaunchRequest.workplan_id == workstream_id)
+    scope_id = workplan_id or workstream_id
+    if scope_id:
+        q = q.where(WorkplanLaunchRequest.workplan_id == scope_id)
     if request_status:
         q = q.where(WorkplanLaunchRequest.status == request_status)
     result = await session.execute(q)

api/routers/tasks.py

@@ -1,7 +1,7 @@
 import uuid
 from datetime import date
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -18,6 +18,7 @@ from api.schemas.task import (
     TaskStatusBulkSyncRead,
     TaskUpdate,
 )
+from api.services.legacy_compat import meter_legacy_query_param
 from api.services.lifecycle import status_value, transition_task_status
 from api.task_status import normalize_task_status
 
@@ -26,6 +27,8 @@ router = APIRouter(prefix="/tasks", tags=["tasks"])
 
 @router.get("/", response_model=list[TaskRead])
 async def list_tasks(
+    request: Request,
+    response: Response,
     workplan_id: uuid.UUID | None = None,
     workstream_id: uuid.UUID | None = None,
     status: str | None = None,
@@ -37,6 +40,15 @@ async def list_tasks(
     offset: int = Query(0, ge=0),
     session: AsyncSession = Depends(get_session),
 ) -> list[Task]:
+    if workstream_id is not None and workplan_id is None:
+        await meter_legacy_query_param(
+            session=session,
+            request=request,
+            response=response,
+            method="GET",
+            route="/tasks/",
+            replacement_ref="/tasks/?workplan_id=<workplan_id>",
+        )
     q = select(Task)
     scope_id = workplan_id or workstream_id
     if scope_id:
@@ -62,11 +74,22 @@ async def list_tasks(
 
 @router.get("/counts", response_model=list[TaskCountRead])
 async def count_tasks(
+    request: Request,
+    response: Response,
     workplan_id: uuid.UUID | None = None,
     workstream_id: uuid.UUID | None = None,
     status: str | None = None,
     session: AsyncSession = Depends(get_session),
 ) -> list[TaskCountRead]:
+    if workstream_id is not None and workplan_id is None:
+        await meter_legacy_query_param(
+            session=session,
+            request=request,
+            response=response,
+            method="GET",
+            route="/tasks/counts",
+            replacement_ref="/tasks/counts?workplan_id=<workplan_id>",
+        )
     q = select(Task.workplan_id, Task.status, func.count()).group_by(Task.workplan_id, Task.status)
     scope_id = workplan_id or workstream_id
     if scope_id:

api/routers/token_events.py

@@ -3,7 +3,7 @@ from collections import defaultdict
 from datetime import datetime
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -12,6 +12,7 @@ from api.models.managed_repo import ManagedRepo
 from api.models.task import Task
 from api.models.token_event import TokenEvent
 from api.models.workplan import Workplan
+from api.services.legacy_compat import meter_legacy_query_param
 from api.schemas.token_event import (
     RepoTokenSummary,
     TokenAggregateRow,
@@ -152,6 +153,7 @@ def _filter_query(
     q,
     *,
     task_id: uuid.UUID | None = None,
+    workplan_id: uuid.UUID | None = None,
     workstream_id: uuid.UUID | None = None,
     repo_id: uuid.UUID | None = None,
     ref_type: str | None = None,
@@ -168,8 +170,9 @@ def _filter_query(
 ):
     if task_id:
         q = q.where(TokenEvent.task_id == task_id)
-    if workstream_id:
-        q = q.where(TokenEvent.workplan_id == workstream_id)
+    scope_id = workplan_id or workstream_id
+    if scope_id:
+        q = q.where(TokenEvent.workplan_id == scope_id)
     if repo_id:
         q = q.where(TokenEvent.repo_id == repo_id)
     if ref_type:
@@ -600,7 +603,10 @@ async def get_token_event(
 
 @router.get("/", response_model=list[TokenEventRead])
 async def list_token_events(
+    request: Request,
+    response: Response,
     task_id: uuid.UUID | None = None,
+    workplan_id: uuid.UUID | None = None,
     workstream_id: uuid.UUID | None = None,
     repo_id: uuid.UUID | None = None,
     ref_type: str | None = None,
@@ -618,9 +624,19 @@ async def list_token_events(
     limit: int = Query(100, le=1000),
     session: AsyncSession = Depends(get_session),
 ) -> list[TokenEvent]:
+    if workstream_id is not None and workplan_id is None:
+        await meter_legacy_query_param(
+            session=session,
+            request=request,
+            response=response,
+            method="GET",
+            route="/token-events/",
+            replacement_ref="/token-events/?workplan_id=<workplan_id>",
+        )
     q = _filter_query(
         select(TokenEvent),
         task_id=task_id,
+        workplan_id=workplan_id,
         workstream_id=workstream_id,
         repo_id=repo_id,
         ref_type=ref_type,

api/services/legacy_compat.py

@@ -0,0 +1,52 @@
+"""Shared legacy REST compat metering (routes and query params)."""
+from __future__ import annotations
+
+import logging
+
+from fastapi import Request, Response
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.services.legacy_meter import identity_from_request, record_legacy_usage
+
+logger = logging.getLogger(__name__)
+
+_LEGACY_OWNER = "state-hub.api"
+
+
+def mark_legacy_response(response: Response | None, replacement_ref: str) -> None:
+    if response is None:
+        return
+    response.headers["Deprecation"] = "true"
+    response.headers["X-StateHub-Replacement"] = replacement_ref
+    response.headers.append("Link", f'<{replacement_ref}>; rel="successor-version"')
+
+
+def legacy_query_param_key(method: str, route: str, param: str = "workstream_id") -> str:
+    return f"rest_api:{method} {route}?{param}"
+
+
+async def meter_legacy_query_param(
+    *,
+    session: AsyncSession,
+    request: Request | None,
+    response: Response | None,
+    method: str,
+    route: str,
+    replacement_ref: str,
+    param: str = "workstream_id",
+) -> None:
+    interface_key = legacy_query_param_key(method, route, param)
+    mark_legacy_response(response, replacement_ref)
+    try:
+        await record_legacy_usage(
+            session,
+            interface_key=interface_key,
+            interface_kind="rest_api",
+            replacement_ref=replacement_ref,
+            owner_component=_LEGACY_OWNER,
+            replacement_verified=True,
+            identity=identity_from_request(request),
+        )
+    except Exception:
+        await session.rollback()
+        logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)

docs/workplan-terminology-legacy-retirement-backlog.md

@@ -63,16 +63,20 @@ Risk order: REST > MCP > events > dashboard prose > internal identifiers.
 | 6 | `open_workstreams` summary cache key | `open_workplans` | Internal | T06 |
 | 6 | `flows/workstream.yaml` entity id | `flows/workplan.yaml` (`custodian.workplan.v1`) | Internal | T06 (workplan flow shipped; workstream yaml retained) |
 
-### Query-param aliases (not separately metered today)
+### Query-param aliases (metered from 2026-07-09)
 
 These accept `workstream_id` alongside `workplan_id` on preferred routes:
 
-- `GET /tasks/` — `api/routers/tasks.py`
-- `GET /decisions/` — `api/routers/decisions.py`
-- `GET /token-events/` — `api/routers/token_events.py`
+| Legacy-meter key | Route | Replacement |
+| --- | --- | --- |
+| `rest_api:GET /tasks/?workstream_id` | `GET /tasks/` | `GET /tasks/?workplan_id=` |
+| `rest_api:GET /tasks/counts?workstream_id` | `GET /tasks/counts` | `GET /tasks/counts?workplan_id=` |
+| `rest_api:GET /decisions/?workstream_id` | `GET /decisions/` | `GET /decisions/?workplan_id=` |
+| `rest_api:GET /token-events/?workstream_id` | `GET /token-events/` | `GET /token-events/?workplan_id=` |
+| `rest_api:GET /execution/launch-requests?workstream_id` | `GET /execution/launch-requests` | `GET /execution/launch-requests?workplan_id=` |
 
-Retire param aliases in T04 after route retirement; document callers via
-weekly review component headers (`X-StateHub-Component`).
+Retire param aliases in T04 after zero-usage windows; callers surface via
+`X-StateHub-Component` in weekly review.
 
 ## Grep budget by phase
 

tests/test_legacy_meter.py

@@ -143,6 +143,46 @@ class TestWorkplanAliasesAndLegacyMeter:
         assert event["interface"]["replacement_ref"] == "org.statehub.workplan.completed"
         assert event["window"]["components"] == {"state-hub.events": 1}
 
+    async def test_legacy_workstream_id_query_param_on_tasks_is_metered(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        wp = await _create_workplan(client, topic["id"])
+        task_r = await client.post("/tasks/", json={
+            "workplan_id": wp["id"],
+            "title": "Meter me",
+            "status": "todo",
+        })
+        assert task_r.status_code == 201, task_r.text
+
+        r = await client.get(
+            f"/tasks/?workstream_id={wp['id']}",
+            headers={
+                "X-StateHub-Tenant": "tenant-a",
+                "X-StateHub-User": "alice",
+                "X-StateHub-Component": "old-task-client",
+            },
+        )
+        assert r.status_code == 200
+        assert r.headers["Deprecation"] == "true"
+        assert r.headers["X-StateHub-Replacement"] == "/tasks/?workplan_id=<workplan_id>"
+
+        summary = (await client.get("/legacy-meter/summary")).json()
+        item = _summary_by_key(summary)["rest_api:GET /tasks/?workstream_id"]
+        assert item["window"]["calls"] == 1
+        assert item["window"]["components"] == {"old-task-client": 1}
+
+    async def test_workplan_id_query_param_on_tasks_is_not_metered(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        wp = await _create_workplan(client, topic["id"])
+
+        r = await client.get(f"/tasks/?workplan_id={wp['id']}")
+        assert r.status_code == 200
+        assert r.headers.get("Deprecation") != "true"
+
+        summary = (await client.get("/legacy-meter/summary")).json()
+        assert _summary_by_key(summary).get("rest_api:GET /tasks/?workstream_id") is None
+
     async def test_workplan_dependency_and_execution_aliases(self, client):
         await _create_domain(client)
         topic = await _create_topic(client)

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-09"
+updated: "2026-07-10"
 state_hub_workstream_id: "923bb94a-d16c-422c-b81e-16328bd7b60c"
 ---
 
@@ -152,7 +152,7 @@ progress events linked to this workplan.
 
 ```task
 id: STATE-WP-0069-T04
-status: todo
+status: progress
 priority: high
 state_hub_task_id: "6fba665d-ebfa-42ab-8cf9-4460f77c5375"
 ```
@@ -171,6 +171,11 @@ and `tests/test_legacy_meter.py` retirement cases.
 Done when OpenAPI documents `/workplans` as the public CRUD surface and
 `/workstreams` returns 410 Gone or is unmounted, with regression tests green.
 
+Progress 2026-07-10 (T04 prep): `api/services/legacy_compat.py` meters
+`workstream_id` query-param usage on `/tasks/`, `/tasks/counts`, `/decisions/`,
+`/token-events/`, and `/execution/launch-requests` with Deprecation headers and
+legacy-meter keys. Route-level `/workstreams` removal remains gated on zero usage.
+
 ## Task: Legacy completion event — stop dual-publish
 
 ```task