feat(api): retire phase-1/2 /workstreams REST aliases with 410 stranglers

Add retire_legacy_route() helper and convert CRUD, dependency, and
execution intent legacy routes to 410 Gone while preserving final-call
legacy-meter attribution. Update tests, retirement backlog, and close
STATE-WP-0075 with handoff to STATE-WP-0070 T02 and STATE-WP-0073 T04.

api/routers/execution.py

@@ -27,7 +27,8 @@ from api.services.execution_queue import (
     queue_sort_key,
     workplan_blockers,
 )
-from api.routers.workstreams import _legacy_key, _meter_legacy_route
+from api.routers.workstreams import _legacy_key
+from api.services.legacy_compat import retire_legacy_route
 from api.services.legacy_compat import meter_legacy_query_param
 from api.workplan_status import CLOSED_WORKPLAN_STATUSES, normalize_workplan_status
 
@@ -62,22 +63,23 @@ async def _update_execution_intent(
     return _intent_read(ws)
 
 
-@router.patch("/workstreams/{workstream_id}/intent", response_model=ExecutionIntentRead)
+@router.patch("/workstreams/{workstream_id}/intent", status_code=status.HTTP_410_GONE)
 async def update_execution_intent(
     request: Request,
     response: Response,
     workstream_id: uuid.UUID,
     body: ExecutionIntentUpdate,
     session: AsyncSession = Depends(get_session),
-) -> ExecutionIntentRead:
-    await _meter_legacy_route(
+) -> None:
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("PATCH", "/execution/workstreams/{workstream_id}/intent"),
         replacement_ref="/execution/workplans/{workplan_id}/intent",
+        detail="Legacy PATCH /execution/workstreams/{workstream_id}/intent retired; "
+        "use PATCH /execution/workplans/{workplan_id}/intent",
     )
-    return await _update_execution_intent(workstream_id=workstream_id, body=body, session=session)
 
 
 @router.patch("/workplans/{workplan_id}/intent", response_model=ExecutionIntentRead)

api/routers/workstream_dependencies.py

@@ -9,7 +9,8 @@ from api.models.task import Task
 from api.models.workplan import Workplan
 from api.models.workplan_dependency import WorkplanDependency
 from api.schemas.workplan_dependency import WorkplanDependencyCreate, WorkplanDependencyRead
-from api.routers.workstreams import _legacy_key, _meter_legacy_route
+from api.routers.workstreams import _legacy_key
+from api.services.legacy_compat import retire_legacy_route
 
 router = APIRouter(prefix="/workstreams", tags=["dependencies"])
 workplan_router = APIRouter(prefix="/workplans", tags=["dependencies"])
@@ -82,8 +83,7 @@ async def _delete_dependency(
 
 @router.post(
     "/{workstream_id}/dependencies/",
-    response_model=WorkplanDependencyRead,
-    status_code=status.HTTP_201_CREATED,
+    status_code=status.HTTP_410_GONE,
 )
 async def create_dependency(
     request: Request,
@@ -91,16 +91,17 @@ async def create_dependency(
     workstream_id: uuid.UUID,
     body: WorkplanDependencyCreate,
     session: AsyncSession = Depends(get_session),
-) -> WorkplanDependency:
-    """Record that workstream_id depends on another workplan or a task."""
-    await _meter_legacy_route(
+) -> None:
+    """Legacy alias retired — use POST /workplans/{workplan_id}/dependencies/."""
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("POST", "/workstreams/{workstream_id}/dependencies/"),
         replacement_ref="/workplans/{workplan_id}/dependencies/",
+        detail="Legacy POST /workstreams/{workstream_id}/dependencies/ retired; "
+        "use POST /workplans/{workplan_id}/dependencies/",
     )
-    return await _create_dependency(workplan_id=workstream_id, body=body, session=session)
 
 
 @workplan_router.post(
@@ -118,23 +119,24 @@ async def create_workplan_dependency(
 
 @router.get(
     "/{workstream_id}/dependencies/",
-    response_model=list[WorkplanDependencyRead],
+    status_code=status.HTTP_410_GONE,
 )
 async def list_dependencies(
     request: Request,
     response: Response,
     workstream_id: uuid.UUID,
     session: AsyncSession = Depends(get_session),
-) -> list[WorkplanDependency]:
-    """Return all dependency edges touching this workplan (both directions)."""
-    await _meter_legacy_route(
+) -> None:
+    """Legacy alias retired — use GET /workplans/{workplan_id}/dependencies/."""
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("GET", "/workstreams/{workstream_id}/dependencies/"),
         replacement_ref="/workplans/{workplan_id}/dependencies/",
+        detail="Legacy GET /workstreams/{workstream_id}/dependencies/ retired; "
+        "use GET /workplans/{workplan_id}/dependencies/",
     )
-    return await _list_dependencies(workplan_id=workstream_id, session=session)
 
 
 @workplan_router.get(
@@ -150,7 +152,7 @@ async def list_workplan_dependencies(
 
 @router.delete(
     "/{workstream_id}/dependencies/{dep_id}",
-    status_code=status.HTTP_204_NO_CONTENT,
+    status_code=status.HTTP_410_GONE,
 )
 async def delete_dependency(
     request: Request,
@@ -159,15 +161,16 @@ async def delete_dependency(
     dep_id: uuid.UUID,
     session: AsyncSession = Depends(get_session),
 ) -> None:
-    """Hard-delete a dependency edge. Removing a constraint is safe — no information is lost."""
-    await _meter_legacy_route(
+    """Legacy alias retired — use DELETE /workplans/{workplan_id}/dependencies/{dep_id}."""
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("DELETE", "/workstreams/{workstream_id}/dependencies/{dep_id}"),
         replacement_ref="/workplans/{workplan_id}/dependencies/{dep_id}",
+        detail="Legacy DELETE /workstreams/{workstream_id}/dependencies/{dep_id} retired; "
+        "use DELETE /workplans/{workplan_id}/dependencies/{dep_id}",
     )
-    await _delete_dependency(workplan_id=workstream_id, dep_id=dep_id, session=session)
 
 
 @workplan_router.delete(

api/routers/workstreams.py

@@ -23,8 +23,7 @@ from api.schemas.workplan import (
     WorkplanUpdate,
 )
 from api.services.lifecycle import transition_workplan_status
-from api.services.legacy_compat import legacy_response_headers, mark_legacy_response
-from api.services.legacy_meter import identity_from_request, record_legacy_usage
+from api.services.legacy_compat import retire_legacy_route
 from api.workplan_status import (
     is_supported_workplan_status,
     normalize_workplan_status,
@@ -42,7 +41,6 @@ _INDEX_TTL = 30.0
 _INDEX_REFRESH_TASK: asyncio.Task | None = None
 _INDEX_LAST_ERROR: str | None = None
 
-_LEGACY_OWNER = "state-hub.api"
 _COMPLETED_WORKPLAN_EVENT = "org.statehub.workplan.completed"
 
 
@@ -82,30 +80,6 @@ def _legacy_key(method: str, route: str) -> str:
     return f"rest_api:{method} {route}"
 
 
-async def _meter_legacy_route(
-    *,
-    session: AsyncSession,
-    request: Request | None,
-    response: Response | None,
-    interface_key: str,
-    replacement_ref: str,
-) -> None:
-    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)
-
-
 async def _list_workplans(
     *,
     topic_id: uuid.UUID | None,
@@ -361,7 +335,7 @@ async def _publish_completion_events(wp: Workplan, session: AsyncSession) -> Non
     asyncio.create_task(publish_event(_COMPLETED_WORKPLAN_EVENT, workplan_envelope))
 
 
-@router.get("/", response_model=list[WorkplanRead])
+@router.get("/", status_code=status.HTTP_410_GONE)
 async def list_workstreams(
     request: Request,
     response: Response,
@@ -372,22 +346,14 @@ async def list_workstreams(
     owner: str | None = None,
     slug: str | None = None,
     session: AsyncSession = Depends(get_session),
-) -> list[Workplan]:
-    await _meter_legacy_route(
+) -> None:
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("GET", "/workstreams/"),
         replacement_ref="/workplans/",
-    )
-    return await _list_workplans(
-        topic_id=topic_id,
-        repo_id=repo_id,
-        repo_goal_id=repo_goal_id,
-        status_filter=status,
-        owner=owner,
-        slug=slug,
-        session=session,
+        detail="Legacy GET /workstreams/ retired; use GET /workplans/",
     )
 
 
@@ -419,18 +385,13 @@ async def workplan_index(
     refresh: bool = Query(False, description="Force cache invalidation"),
     session: AsyncSession = Depends(get_session),
 ) -> None:
-    replacement_ref = "/workplans/index"
-    await _meter_legacy_route(
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("GET", "/workstreams/workplan-index"),
-        replacement_ref=replacement_ref,
-    )
-    raise HTTPException(
-        status_code=status.HTTP_410_GONE,
+        replacement_ref="/workplans/index",
         detail="Legacy GET /workstreams/workplan-index retired; use GET /workplans/index",
-        headers=legacy_response_headers(replacement_ref),
     )
 
 
@@ -464,21 +425,21 @@ async def sync_workplan_bindings(
     return {"updated": updated, "received": len(body.bindings)}
 
 
-@router.post("/", response_model=WorkplanRead, status_code=status.HTTP_201_CREATED)
+@router.post("/", status_code=status.HTTP_410_GONE)
 async def create_workstream(
     request: Request,
     response: Response,
     body: WorkplanCreate,
     session: AsyncSession = Depends(get_session),
-) -> Workplan:
-    await _meter_legacy_route(
+) -> None:
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("POST", "/workstreams/"),
         replacement_ref="/workplans/",
+        detail="Legacy POST /workstreams/ retired; use POST /workplans/",
     )
-    return await _create_workplan(body=body, session=session)
 
 
 @workplan_router.post("/", response_model=WorkplanRead, status_code=status.HTTP_201_CREATED)
@@ -489,21 +450,21 @@ async def create_workplan(
     return await _create_workplan(body=body, session=session)
 
 
-@router.get("/{workstream_id}", response_model=WorkplanRead)
+@router.get("/{workstream_id}", status_code=status.HTTP_410_GONE)
 async def get_workstream(
     request: Request,
     response: Response,
     workstream_id: uuid.UUID,
     session: AsyncSession = Depends(get_session),
-) -> Workplan:
-    await _meter_legacy_route(
+) -> None:
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("GET", "/workstreams/{workstream_id}"),
         replacement_ref="/workplans/{workplan_id}",
+        detail="Legacy GET /workstreams/{workstream_id} retired; use GET /workplans/{workplan_id}",
     )
-    return await _get_workplan(workplan_id=workstream_id, session=session)
 
 
 @workplan_router.get("/{workplan_id}", response_model=WorkplanRead)
@@ -514,22 +475,22 @@ async def get_workplan(
     return await _get_workplan(workplan_id=workplan_id, session=session)
 
 
-@router.patch("/{workstream_id}", response_model=WorkplanRead)
+@router.patch("/{workstream_id}", status_code=status.HTTP_410_GONE)
 async def update_workstream(
     request: Request,
     response: Response,
     workstream_id: uuid.UUID,
     body: WorkplanUpdate,
     session: AsyncSession = Depends(get_session),
-) -> Workplan:
-    await _meter_legacy_route(
+) -> None:
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("PATCH", "/workstreams/{workstream_id}"),
         replacement_ref="/workplans/{workplan_id}",
+        detail="Legacy PATCH /workstreams/{workstream_id} retired; use PATCH /workplans/{workplan_id}",
     )
-    return await _update_workplan(workplan_id=workstream_id, body=body, session=session)
 
 
 @workplan_router.patch("/{workplan_id}", response_model=WorkplanRead)
@@ -548,18 +509,13 @@ async def archive_workstream(
     workstream_id: uuid.UUID,
     session: AsyncSession = Depends(get_session),
 ) -> None:
-    replacement_ref = "DELETE /workplans/{workplan_id}"
-    await _meter_legacy_route(
+    await retire_legacy_route(
         session=session,
         request=request,
         response=response,
         interface_key=_legacy_key("DELETE", "/workstreams/{workstream_id}"),
-        replacement_ref=replacement_ref,
-    )
-    raise HTTPException(
-        status_code=status.HTTP_410_GONE,
+        replacement_ref="DELETE /workplans/{workplan_id}",
         detail="Legacy DELETE /workstreams/{workstream_id} retired; use DELETE /workplans/{workplan_id}",
-        headers=legacy_response_headers(replacement_ref),
     )
 
 

api/services/legacy_compat.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 import logging
 
-from fastapi import Request, Response
+from fastapi import HTTPException, Request, Response, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.services.legacy_meter import identity_from_request, record_legacy_usage
@@ -97,6 +97,37 @@ async def meter_legacy_body_field(
         logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)
 
 
+async def retire_legacy_route(
+    *,
+    session: AsyncSession,
+    request: Request | None,
+    response: Response | None,
+    interface_key: str,
+    replacement_ref: str,
+    detail: str,
+) -> None:
+    """Record final legacy-meter usage for this route, then return 410 Gone."""
+    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)
+    raise HTTPException(
+        status_code=status.HTTP_410_GONE,
+        detail=detail,
+        headers=legacy_response_headers(replacement_ref),
+    )
+
+
 async def meter_legacy_body_from_model(
     body: object,
     *,