feat(mcp): add legacy workstream alias deprecation warnings (STATE-WP-0069 T03)

Legacy MCP tools and the workstreams resource now return _deprecation payloads
pointing to workplan-named replacements. get_domain_summary uses workplan-first
prose and exposes a workplans key alongside the legacy workstreams field.

mcp_server/server.py

@@ -26,7 +26,7 @@ API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")
 mcp = FastMCP(
     name=MCP_SERVER_NAME,
     instructions=(
-        "Custodian State Hub: tracks topics, workstreams, tasks, decisions, and progress events. "
+        "Custodian State Hub: tracks topics, workplans, tasks, decisions, and progress events. "
         "Start every session with get_state_summary() for orientation. "
         "When working inside a single registered domain repo, prefer get_domain_summary(domain_slug) "
         "— it returns the same actionable data scoped to that domain at ~10% of the token cost. "
@@ -172,6 +172,68 @@ def _json_result(result: Any) -> str:
     return json.dumps(result, indent=2)
 
 
+_LEGACY_MCP_TOOL_REPLACEMENTS: dict[str, str] = {
+    "create_workstream": "create_workplan",
+    "list_workstreams": "list_workplans",
+    "update_workstream": "update_workplan",
+    "update_workstream_status": "update_workplan_status",
+}
+
+_LEGACY_MCP_RESOURCE_REPLACEMENTS: dict[str, str] = {
+    "state://workstreams/{topic_slug}": "state://workplans/{topic_slug}",
+}
+
+
+def _legacy_mcp_deprecation(*, tool: str | None = None, resource: str | None = None) -> dict[str, str]:
+    if tool:
+        replacement = _LEGACY_MCP_TOOL_REPLACEMENTS[tool]
+        return {
+            "deprecated": "true",
+            "tool": tool,
+            "replacement": replacement,
+            "message": (
+                f"{tool} is a legacy MCP alias; use {replacement} instead. "
+                "Retirement tracked by legacy-meter (STATE-WP-0069)."
+            ),
+        }
+    if resource:
+        replacement = _LEGACY_MCP_RESOURCE_REPLACEMENTS[resource]
+        return {
+            "deprecated": "true",
+            "resource": resource,
+            "replacement": replacement,
+            "message": (
+                f"{resource} is a legacy MCP resource; use {replacement} instead. "
+                "Retirement tracked by legacy-meter (STATE-WP-0069)."
+            ),
+        }
+    raise ValueError("tool or resource required")
+
+
+def _attach_legacy_deprecation(tool_name: str, result_json: str) -> str:
+    if tool_name not in _LEGACY_MCP_TOOL_REPLACEMENTS:
+        return result_json
+    try:
+        payload = json.loads(result_json)
+    except json.JSONDecodeError:
+        return result_json
+    if isinstance(payload, dict) and payload.get("error"):
+        return result_json
+    if isinstance(payload, dict):
+        payload["_deprecation"] = _legacy_mcp_deprecation(tool=tool_name)
+        return json.dumps(payload, indent=2)
+    if isinstance(payload, list):
+        return json.dumps(
+            {
+                "_deprecation": _legacy_mcp_deprecation(tool=tool_name),
+                "workplans": payload,
+                "workstreams": payload,
+            },
+            indent=2,
+        )
+    return result_json
+
+
 # ---------------------------------------------------------------------------
 # Resources
 # ---------------------------------------------------------------------------
@@ -188,14 +250,32 @@ def resource_topics() -> str:
     return json.dumps(_get("/topics", {"status": "active"}), indent=2)
 
 
-@mcp.resource("state://workstreams/{topic_slug}")
+@mcp.resource("state://workplans/{topic_slug}")
-def resource_workstreams(topic_slug: str) -> str:
+def resource_workplans(topic_slug: str) -> str:
-    """Workstreams for a topic (by slug)."""
+    """Workplans for a topic (by slug)."""
     topics = _get("/topics", {"status": "active"})
     match = next((t for t in topics if t["slug"] == topic_slug), None)
     if not match:
         return json.dumps({"error": f"Topic '{topic_slug}' not found"})
-    return json.dumps(_get("/workstreams", {"topic_id": match["id"]}), indent=2)
+    return json.dumps(_get("/workplans", {"topic_id": match["id"]}), indent=2)
+
+
+@mcp.resource("state://workstreams/{topic_slug}")
+def resource_workstreams(topic_slug: str) -> str:
+    """Legacy resource alias — prefer state://workplans/{topic_slug}."""
+    topics = _get("/topics", {"status": "active"})
+    match = next((t for t in topics if t["slug"] == topic_slug), None)
+    if not match:
+        return json.dumps({"error": f"Topic '{topic_slug}' not found"})
+    rows = _get("/workstreams", {"topic_id": match["id"]})
+    return json.dumps(
+        {
+            "_deprecation": _legacy_mcp_deprecation(resource="state://workstreams/{topic_slug}"),
+            "workplans": rows,
+            "workstreams": rows,
+        },
+        indent=2,
+    )
 
 
 @mcp.resource("state://decisions/blocking")
@@ -221,8 +301,8 @@ def resource_blocked_tasks() -> str:
 def get_state_summary() -> str:
     """Primary orientation tool. Call at the start of every session.
 
-    Returns a full snapshot: topic/workstream/task/decision totals, blocking
+    Returns a full snapshot: topic/workplan/task/decision totals, blocking
-    decisions, waiting tasks, open workstreams, and the 20 most recent events.
+    decisions, waiting tasks, open workplans, and the 20 most recent events.
 
     NOTE: This response is large (~10k tokens). When working inside a single
     registered domain repo, use get_domain_summary(domain_slug) instead —
@@ -242,7 +322,7 @@ def get_domain_summary(domain_slug: str) -> str:
     Args:
         domain_slug: the domain slug, e.g. "railiance", "markitect"
 
-    Returns: topic, active workstreams, open blocking decisions for this
+    Returns: topic, active workplans, open blocking decisions for this
     topic, 5 most recent progress events, repo SBOM status, and goal guidance
     (needs_workplan signals + alignment warnings).
     """
@@ -276,8 +356,8 @@ def get_domain_summary(domain_slug: str) -> str:
         if ws.get("repo_id"):
             ws_by_repo.setdefault(ws["repo_id"], []).append(ws)
 
-    needs_workplan: list[dict] = []   # active goal with no linked workstream
+    needs_workplan: list[dict] = []   # active goal with no linked workplan
-    alignment_warnings: list[dict] = []  # workstreams not linked to active goal
+    alignment_warnings: list[dict] = []  # workplans not linked to active goal
 
     for repo in repos:
         repo_slug = repo["slug"]
@@ -297,30 +377,31 @@ def get_domain_summary(domain_slug: str) -> str:
                     "goal_description": goal["description"],
                     "priority": goal["priority"],
                     "action": (
-                        f"No workstream is linked to repo goal '{goal['title']}'. "
+                        f"No workplan is linked to repo goal '{goal['title']}'. "
-                        "Create a workplan file in workplans/ and register a workstream "
+                        "Create a workplan file in workplans/ and run fix-consistency "
                         f"with repo_goal_id='{goal['id']}' to start delivering this goal."
                     ),
                 })
 
-        # Check if repo has active workstreams not tied to any active goal
+        # Check if repo has active workplans not tied to any active goal
         repo_ws = ws_by_repo.get(repo_id, [])
         unlinked_ws = [
             ws for ws in repo_ws
             if ws.get("repo_goal_id") not in active_goal_ids
         ]
         if unlinked_ws:
-            # Most recently updated workstream = the one to suggest continuing
             recent_ws = max(unlinked_ws, key=lambda w: w.get("updated_at", ""))
             alignment_warnings.append({
                 "repo_slug": repo_slug,
+                "recent_workplan_id": recent_ws["id"],
+                "recent_workplan_title": recent_ws["title"],
                 "recent_workstream_id": recent_ws["id"],
                 "recent_workstream_title": recent_ws["title"],
                 "active_goal_titles": [g["title"] for g in active_goals],
                 "message": (
-                    f"Workstream '{recent_ws['title']}' is not linked to the current "
+                    f"Workplan '{recent_ws['title']}' is not linked to the current "
                     f"repo goal(s) for {repo_slug}. "
-                    "Continue this workstream if the work is still relevant, but verify "
+                    "Continue this workplan if the work is still relevant, but verify "
                     "alignment with the active goal before committing to new tasks."
                 ),
             })
@@ -336,6 +417,7 @@ def get_domain_summary(domain_slug: str) -> str:
         "domain": domain_slug,
         "topic_id": topic_id,
         "topic_title": topic["title"],
+        "workplans": workstreams,
         "workstreams": workstreams,
         "blocking_decisions": blocking,
         "recent_progress": recent,
@@ -583,7 +665,7 @@ def _create_workplan_impl(
     })
     if progress_error:
         return _json_result(progress_error)
-    return _json_result(wp)
+    return _attach_legacy_deprecation(tool_name, _json_result(wp))
 
 
 def _update_workplan_status_impl(workplan_id: str, status: str, *, tool_name: str) -> str:
@@ -601,7 +683,7 @@ def _update_workplan_status_impl(workplan_id: str, status: str, *, tool_name: st
     })
     if progress_error:
         return _json_result(progress_error)
-    return _json_result(wp)
+    return _attach_legacy_deprecation(tool_name, _json_result(wp))
 
 
 def _update_workplan_impl(
@@ -630,6 +712,31 @@ def _update_workplan_impl(
     return _json_result(_patch(f"/workplans/{workplan_id}", payload))
 
 
+def _update_workplan_legacy_impl(
+    workplan_id: str,
+    *,
+    tool_name: str,
+    title: str | None = None,
+    description: str | None = None,
+    owner: str | None = None,
+    due_date: str | None = None,
+    repo_goal_id: str | None = None,
+    status: str | None = None,
+) -> str:
+    return _attach_legacy_deprecation(
+        tool_name,
+        _update_workplan_impl(
+            workplan_id,
+            title=title,
+            description=description,
+            owner=owner,
+            due_date=due_date,
+            repo_goal_id=repo_goal_id,
+            status=status,
+        ),
+    )
+
+
 # ---------------------------------------------------------------------------
 # Mutate tools
 # ---------------------------------------------------------------------------
@@ -685,7 +792,7 @@ def create_workstream(
     planning_priority: str | None = None,
     planning_order: int | None = None,
 ) -> str:
-    """Legacy alias for create_workplan — prefer create_workplan(repo_id=...)."""
+    """DEPRECATED legacy alias — use create_workplan(repo_id=...) instead."""
     if not repo_id:
         return _json_result(_mcp_error("create_workstream", "repo_id is required"))
     return _create_workplan_impl(
@@ -1102,13 +1209,16 @@ def list_workstreams(
     owner: str | None = None,
     slug: str | None = None,
 ) -> str:
-    """Legacy alias for list_workplans."""
+    """DEPRECATED legacy alias — use list_workplans instead."""
-    return list_workplans(
+    return _attach_legacy_deprecation(
-        repo_id=repo_id,
+        "list_workstreams",
-        topic_id=topic_id,
+        list_workplans(
-        status=status,
+            repo_id=repo_id,
-        owner=owner,
+            topic_id=topic_id,
-        slug=slug,
+            status=status,
+            owner=owner,
+            slug=slug,
+        ),
     )
 
 
@@ -1129,7 +1239,7 @@ def update_workstream_status(
     workplan_id: str | None = None,
     workstream_id: str | None = None,
 ) -> str:
-    """Legacy alias for update_workplan_status."""
+    """DEPRECATED legacy alias — use update_workplan_status instead."""
     parent_id = workplan_id or workstream_id
     if not parent_id:
         return _json_result(_mcp_error("update_workstream_status", "workplan_id is required"))
@@ -1169,12 +1279,13 @@ def update_workstream(
     repo_goal_id: str | None = None,
     status: str | None = None,
 ) -> str:
-    """Legacy alias for update_workplan."""
+    """DEPRECATED legacy alias — use update_workplan instead."""
     parent_id = workplan_id or workstream_id
     if not parent_id:
         return _json_result(_mcp_error("update_workstream", "workplan_id is required"))
-    return _update_workplan_impl(
+    return _update_workplan_legacy_impl(
         parent_id,
+        tool_name="update_workstream",
         title=title,
         description=description,
         owner=owner,

tests/test_mcp_legacy_deprecation.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import json
+
+import mcp_server.server as server
+
+
+def test_legacy_mcp_deprecation_payload_names_replacement() -> None:
+    payload = server._legacy_mcp_deprecation(tool="create_workstream")
+    assert payload["replacement"] == "create_workplan"
+    assert payload["deprecated"] == "true"
+
+
+def test_attach_legacy_deprecation_preserves_success_payload() -> None:
+    raw = server._attach_legacy_deprecation(
+        "update_workstream_status",
+        json.dumps({"id": "wp-1", "status": "active"}),
+    )
+    body = json.loads(raw)
+    assert body["id"] == "wp-1"
+    assert body["_deprecation"]["tool"] == "update_workstream_status"
+
+
+def _fake_get(path: str, params: dict | None = None):
+    if path == "/topics":
+        return [{"id": "topic-1", "domain_slug": "infotech", "title": "Infotech"}]
+    if path == "/state/summary":
+        return {
+            "open_workstreams": [
+                {
+                    "id": "wp-1",
+                    "topic_id": "topic-1",
+                    "title": "Goalless",
+                    "repo_id": "repo-1",
+                }
+            ]
+        }
+    if path == "/decisions":
+        return []
+    if path == "/progress":
+        return []
+    if path == "/repos":
+        return [{"id": "repo-1", "slug": "demo-repo", "domain_slug": "infotech"}]
+    if path == "/repo-goals":
+        return [{"id": "goal-1", "title": "Ship it", "description": "d", "priority": "high"}]
+    if path == "/capability-catalog/":
+        return []
+    return []
+
+
+def test_get_domain_summary_goal_guidance_is_workplan_first(monkeypatch) -> None:
+    monkeypatch.setattr(server, "_get", lambda path, params=None: _fake_get(path, params))
+    payload = json.loads(server.get_domain_summary("infotech"))
+    assert "workplans" in payload
+    assert payload["workstreams"] == payload["workplans"]
+    action = payload["goal_guidance"]["needs_workplan"][0]["action"]
+    assert "workplan" in action.lower()
+    assert "workstream is linked" not in action.lower()

tests/test_mcp_write_tools.py

@@ -75,8 +75,24 @@ class TestMCPWriteTools:
         )
 
         assert body["id"] == "wp-1"
+        assert body["_deprecation"]["replacement"] == "create_workplan"
         assert [path for path, _ in calls] == ["/workplans", "/progress"]
 
+    async def test_list_workstreams_legacy_alias_wraps_deprecation(self, monkeypatch):
+        monkeypatch.setattr(
+            server,
+            "_get",
+            lambda path, params=None: [{"id": "wp-1", "title": "Example"}]
+            if path == "/workplans"
+            else [],
+        )
+
+        body = await _call_tool("list_workstreams", {})
+
+        assert body["_deprecation"]["replacement"] == "list_workplans"
+        assert body["workplans"] == [{"id": "wp-1", "title": "Example"}]
+        assert body["workstreams"] == body["workplans"]
+
     async def test_create_task_returns_rest_shape_and_emits_progress(self, monkeypatch):
         calls: list[tuple[str, dict[str, Any]]] = []
 

workplans/STATE-WP-0069-workplan-terminology-legacy-retirement.md

@@ -119,7 +119,7 @@ Done when dashboard `npm test` passes and scan shows zero `prose:workstream` in
 
 ```task
 id: STATE-WP-0069-T03
-status: todo
+status: progress
 priority: medium
 state_hub_task_id: "f701aa56-aa06-4d0e-a632-758677e7ac98"
 ```