coulomb/state-hub
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

STATE-WP-0070 T03: remove legacy workstream MCP aliases

Browse source 
Meter-gated removal (7 consecutive zero-usage windows reached). Deletes:
- MCP tools: create_workstream, list_workstreams, update_workstream,
  update_workstream_status
- MCP resource: state://workstreams/{topic_slug}
- Dead helpers: _LEGACY_MCP_* maps, _legacy_mcp_deprecation,
  _meter_legacy_mcp, _attach_legacy_deprecation, _update_workplan_legacy_impl

TOOLS.md keeps a retired→preferred migration map (per backlog). Removed the
now-dead STATEHUB_MCP_LEGACY_METER guard in conftest. Retargeted the
create_workplan error-skip test to the preferred tool; dropped alias-only tests.

Staged on branch state-wp-0070-legacy-retirement — do not merge until the 7th
documented zero-usage window is captured.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-07-13 09:29:12 +02:00
parent 1acad4c02c
commit ef62cb3872
5 changed files with 28 additions and 328 deletions
Download patch file Download diff file Expand all files Collapse all files

41
mcp_server/TOOLS.md
View file

@ -25,18 +25,25 @@ Do not use them as a substitute for formal work definition inside the domain rep
**Preferred terms:** workplan, `workplan_id`, `/workplans/…`
**Legacy compatibility:** `workstream`, `workstream_id`, `/workstreams/…`, and
`create_workstream` / `update_workstream` MCP tools remain available as aliases.
They call the same implementation as the workplan-named tools and endpoints.
**Legacy compatibility:** the `workstream_id` query/body field is still accepted
as an alias on preferred routes. The `/workstreams/…` REST routes and the
`create_workstream` / `update_workstream` / `update_workstream_status` /
`list_workstreams` MCP tools and the `state://workstreams/{topic_slug}` resource
were **retired** by STATE-WP-0070 (phase-2 legacy retirement) after the
legacy-meter showed seven consecutive zero-usage windows. Use the workplan-named
tools below.
| Preferred (workplan) | Legacy alias (workstream) |
Migration map (retired → preferred):
| Retired legacy alias | Preferred (workplan) |
|---|---|
| `create_workplan(repo_id, …)` | `create_workstream(repo_id, …)` |
| `update_workplan` / `update_workplan_status` | `update_workstream` / `update_workstream_status` |
| `list_workplans` | `list_workstreams` |
| `create_workplan_dependency` | `create_dependency` |
| `POST /workplans/` | `POST /workstreams/` (deprecated headers) |
| `workplan_id` query/body field | `workstream_id` (accepted alias) |
| `create_workstream(repo_id, …)` | `create_workplan(repo_id, …)` |
| `update_workstream` / `update_workstream_status` | `update_workplan` / `update_workplan_status` |
| `list_workstreams` | `list_workplans` |
| `create_dependency` | `create_workplan_dependency` |
| `POST /workstreams/` (410 Gone) | `POST /workplans/` |
| `state://workstreams/{topic_slug}` (removed) | `state://workplans/{topic_slug}` |
| `workstream_id` (still-accepted field alias) | `workplan_id` query/body field |
Repo classification filters: `list_repos_by_classification(category?, domain?,
capability_tag?, business_stake?)` and extended `list_domain_repos(...)` query
@ -56,7 +63,6 @@ endpoint they wrap:
| `list_workplans(...)` | `GET /workplans/` |
| `update_workplan_status(...)` | `PATCH /workplans/{workplan_id}` |
| `list_repos_by_classification(...)` | `GET /repos/?category=…` |
| `create_workstream(...)` | `POST /workstreams/` |
| `create_task(...)` | `POST /tasks/` |
| `update_task_status(...)` | `PATCH /tasks/{task_id}` |
| `bulk_update_task_statuses(...)` | `POST /tasks/bulk-status-sync` |
@ -126,12 +132,12 @@ and replay with statehub outbox status and statehub outbox replay.
| Tool | Key Args | Notes |
|------|----------|-------|
| `create_workplan(topic_id, title, ...)` | `slug?`; `owner?`; `description?`; `due_date?` | **Preferred.** Creates workplan under a topic. Legacy alias: `create_workstream`. |
| `create_workplan(topic_id, title, ...)` | `slug?`; `owner?`; `description?`; `due_date?` | **Preferred.** Creates workplan under a topic. |
| `create_task(workplan_id, title, ...)` | `priority`: low/medium/high/critical; `assignee?`; `due_date?` | Creates task under a workplan (`workstream_id` alias accepted). |
| `update_task_status(task_id, status, ...)` | `status`: wait/todo/progress/done/cancel; `blocking_reason?` describes wait conditions | Legacy aliases `blocked`, `in_progress`, `cancelled`, and `canceled` are accepted during migration. |
| `bulk_update_task_statuses(updates, author?, session_id?)` | `updates`: list of `{task_id, status, blocking_reason?}` | Updates many task statuses in one REST call and emits one `task_status_changed` progress event per task. Prefer this at session checkpoints instead of many single-task calls. |
| `update_workstream_status(workstream_id, status)` | `status`: proposed/ready/active/blocked/backlog/finished/archived | Thin shortcut — use `update_workstream` for full field control. |
| `update_workstream(workstream_id, ...)` | `title?`; `description?`; `owner?`; `due_date?`; `repo_goal_id?`; `status?` | Patch any subset of workstream fields. Pass empty string for `repo_goal_id` to clear the link. |
| `update_workplan_status(workplan_id, status)` | `status`: proposed/ready/active/blocked/backlog/finished/archived | Thin shortcut — use `update_workplan` for full field control. |
| `update_workplan(workplan_id, ...)` | `title?`; `description?`; `owner?`; `due_date?`; `repo_goal_id?`; `status?` | Patch any subset of workplan fields. Pass empty string for `repo_goal_id` to clear the link. |
---
@ -186,8 +192,7 @@ Agents should call `record_token_event` (or pass `tokens_in`/`tokens_out` via
|-----|---------|
| `state://summary` | Full StateSummary JSON |
| `state://topics` | Active topics list |
| `state://workplans/{topic_slug}` | Workplans for a topic (by slug) — preferred |
| `state://workstreams/{topic_slug}` | Legacy resource alias (deprecated; same payload + `_deprecation`) |
| `state://workplans/{topic_slug}` | Workplans for a topic (by slug) |
| `state://decisions/blocking` | All pending decisions |
| `state://tasks/blocked` | Legacy resource name; returns all `wait` tasks |
@ -371,6 +376,6 @@ add_progress_event(
)
# First Session Protocol only — bootstrap a new project:
create_workstream(topic_id="<uuid>", title="My Workstream", owner="me")
create_task(workstream_id="<uuid>", title="Do the thing", priority="high")
create_workplan(topic_id="<uuid>", title="My Workplan", owner="me")
create_task(workplan_id="<uuid>", title="Do the thing", priority="high")
```

238
mcp_server/server.py
View file

@ -181,102 +181,6 @@ 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 _meter_legacy_mcp(*, tool: str | None = None, resource: str | None = None) -> None:
"""Record legacy MCP alias usage on the legacy meter (best-effort).
Keys match docs/workplan-terminology-legacy-retirement-backlog.md phase 3
(`mcp:<tool_name>` / resource URI). Failures never surface to the caller —
the meter gates retirement, it must not break the aliased call itself.
"""
if os.environ.get("STATEHUB_MCP_LEGACY_METER", "on") == "off":
return
if tool:
key = f"mcp:{tool}"
replacement = _LEGACY_MCP_TOOL_REPLACEMENTS[tool]
elif resource:
key = resource
replacement = _LEGACY_MCP_RESOURCE_REPLACEMENTS[resource]
else:
raise ValueError("tool or resource required")
try:
with _client() as c:
c.post(
"/legacy-meter/usage",
json={
"interface_key": key,
"interface_kind": "mcp_tool",
"replacement_ref": replacement,
"owner_component": "state-hub.mcp",
"replacement_verified": True,
"component_key": "state-hub.mcp",
},
)
except Exception:
pass
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
# ---------------------------------------------------------------------------
@ -303,25 +207,6 @@ def resource_workplans(topic_slug: str) -> str:
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}."""
_meter_legacy_mcp(resource="state://workstreams/{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("/workplans", {"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")
def resource_blocking_decisions() -> str:
"""All pending/escalated decisions."""
@ -709,7 +594,7 @@ def _create_workplan_impl(
})
if progress_error:
return _json_result(progress_error)
return _attach_legacy_deprecation(tool_name, _json_result(wp))
return _json_result(wp)
def _update_workplan_status_impl(workplan_id: str, status: str, *, tool_name: str) -> str:
@ -727,7 +612,7 @@ def _update_workplan_status_impl(workplan_id: str, status: str, *, tool_name: st
})
if progress_error:
return _json_result(progress_error)
return _attach_legacy_deprecation(tool_name, _json_result(wp))
return _json_result(wp)
def _update_workplan_impl(
@ -756,31 +641,6 @@ 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
# ---------------------------------------------------------------------------
@ -824,36 +684,6 @@ def create_workplan(
)
@mcp.tool()
def create_workstream(
title: str,
repo_id: str | None = None,
topic_id: str | None = None,
slug: str | None = None,
description: str | None = None,
owner: str | None = None,
due_date: str | None = None,
planning_priority: str | None = None,
planning_order: int | None = None,
) -> str:
"""DEPRECATED legacy alias — use create_workplan(repo_id=...) instead."""
_meter_legacy_mcp(tool="create_workstream")
if not repo_id:
return _json_result(_mcp_error("create_workstream", "repo_id is required"))
return _create_workplan_impl(
repo_id=repo_id,
title=title,
topic_id=topic_id,
slug=slug,
description=description,
owner=owner,
due_date=due_date,
planning_priority=planning_priority,
planning_order=planning_order,
tool_name="create_workstream",
)
@mcp.tool()
def create_task(
workplan_id: str | None = None,
@ -1246,28 +1076,6 @@ def list_workplans(
)
@mcp.tool()
def list_workstreams(
topic_id: str | None = None,
repo_id: str | None = None,
status: str | None = None,
owner: str | None = None,
slug: str | None = None,
) -> str:
"""DEPRECATED legacy alias — use list_workplans instead."""
_meter_legacy_mcp(tool="list_workstreams")
return _attach_legacy_deprecation(
"list_workstreams",
list_workplans(
repo_id=repo_id,
topic_id=topic_id,
status=status,
owner=owner,
slug=slug,
),
)
@mcp.tool()
def update_workplan_status(workplan_id: str, status: str) -> str:
"""Update a workplan's status.
@ -1279,20 +1087,6 @@ def update_workplan_status(workplan_id: str, status: str) -> str:
return _update_workplan_status_impl(workplan_id, status, tool_name="update_workplan_status")
@mcp.tool()
def update_workstream_status(
status: str,
workplan_id: str | None = None,
workstream_id: str | None = None,
) -> str:
"""DEPRECATED legacy alias — use update_workplan_status instead."""
_meter_legacy_mcp(tool="update_workstream_status")
parent_id = workplan_id or workstream_id
if not parent_id:
return _json_result(_mcp_error("update_workstream_status", "workplan_id is required"))
return _update_workplan_status_impl(parent_id, status, tool_name="update_workstream_status")
@mcp.tool()
def update_workplan(
workplan_id: str,
@ -1315,34 +1109,6 @@ def update_workplan(
)
@mcp.tool()
def update_workstream(
workplan_id: str | None = None,
workstream_id: str | None = None,
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:
"""DEPRECATED legacy alias — use update_workplan instead."""
_meter_legacy_mcp(tool="update_workstream")
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_legacy_impl(
parent_id,
tool_name="update_workstream",
title=title,
description=description,
owner=owner,
due_date=due_date,
repo_goal_id=repo_goal_id,
status=status,
)
# ---------------------------------------------------------------------------
# Next-steps suggestion tool (S2.3) — sanctioned write use case #2
# ---------------------------------------------------------------------------

4
tests/conftest.py
View file

@ -23,10 +23,6 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
# Make api/ importable when running pytest from state-hub/
sys.path.insert(0, str(Path(__file__).parent.parent))
# Tests exercising legacy MCP aliases must not record real usage on the live
# legacy meter — that would reset zero-window retirement streaks every run.
os.environ.setdefault("STATEHUB_MCP_LEGACY_METER", "off")
_ASYNC_URL = os.getenv(
"TEST_DATABASE_URL",
"postgresql+asyncpg://custodian:changeme@127.0.0.1:5432/custodian_test",

30
tests/test_mcp_legacy_deprecation.py
View file

@ -27,22 +27,6 @@ def test_normalize_progress_body_prefers_workplan_id_only() -> None:
assert "workstream_id" not in legacy_only
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"}]
@ -70,20 +54,6 @@ def _fake_get(path: str, params: dict | None = None):
return []
def test_legacy_workstreams_resource_includes_deprecation(monkeypatch) -> None:
def fake_get(path: str, params: dict | None = None):
if path == "/topics":
return [{"id": "topic-1", "slug": "infotech", "domain_slug": "infotech"}]
if path == "/workstreams":
return [{"id": "wp-1", "title": "Demo", "slug": "demo-wp"}]
return []
monkeypatch.setattr(server, "_get", fake_get)
payload = json.loads(server.resource_workstreams("infotech"))
assert payload["_deprecation"]["replacement"] == "state://workplans/{topic_slug}"
assert payload["workplans"] == payload["workstreams"]
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"))

43
tests/test_mcp_write_tools.py
View file

@ -56,43 +56,6 @@ class TestMCPWriteTools:
assert calls[1][1]["workplan_id"] == "wp-1"
assert calls[1][1]["event_type"] == "workplan_created"
async def test_create_workstream_legacy_alias_uses_workplans_endpoint(self, monkeypatch):
calls: list[tuple[str, dict[str, Any]]] = []
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
calls.append((path, body))
if path == "/workplans":
return {"id": "wp-1", "repo_id": body["repo_id"], "title": body["title"], "slug": body["slug"], "status": "active"}
if path == "/progress":
return {"id": "event-1", **body}
raise AssertionError(f"unexpected POST {path}")
monkeypatch.setattr(server, "_post", fake_post)
body = await _call_tool(
"create_workstream",
{"repo_id": "repo-1", "title": "Legacy alias"},
)
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]]] = []
@ -306,7 +269,7 @@ class TestMCPWriteTools:
assert calls[1][1]["decision_id"] == "decision-1"
assert calls[1][1]["event_type"] == "decision_recorded"
async def test_create_workstream_api_error_skips_progress(self, monkeypatch):
async def test_create_workplan_api_error_skips_progress(self, monkeypatch):
calls: list[tuple[str, dict[str, Any]]] = []
def fake_post(path: str, body: dict[str, Any]) -> dict[str, Any]:
@ -316,11 +279,11 @@ class TestMCPWriteTools:
monkeypatch.setattr(server, "_post", fake_post)
body = await _call_tool(
"create_workstream",
"create_workplan",
{"repo_id": "bad-repo", "title": "No progress on failure"},
)
assert body["tool"] == "create_workstream"
assert body["tool"] == "create_workplan"
assert body["error"] == "API 422: invalid repo"
assert [path for path, _ in calls] == ["/workplans"]