Finish STATE-WP-0069: retire legacy completion event and DELETE /workstreams

Stop dual-publishing org.statehub.workstream.completed; return 410 Gone for
legacy DELETE /workstreams/{id}. Migrate fix-consistency, MCP adhoc bootstrap,
and dashboard token summary to /workplans/. Add legacy-meter evidence capture
script and pytest snapshot; update docs and close out the workplan.

tests/test_consistency_check.py

@@ -967,17 +967,17 @@ class TestLifecycleRenormalization:
                     "local_path": str(repo),
                     "host_paths": {socket.gethostname(): str(repo)},
                 }
-            if path == "/workstreams/ws-1":
+            if path == "/workplans/ws-1":
                 return ws
             if path == "/tasks/task-1":
                 return task
             if path == "/tasks" and params == {"workstream_id": "ws-1"}:
                 return [task]
-            if path == "/workstreams/ws-1/dependencies":
+            if path == "/workplans/ws-1/dependencies":
                 return []
-            if path == "/workstreams" and params == {"repo_id": "repo-1"}:
+            if path == "/workplans" and params == {"repo_id": "repo-1"}:
                 return [ws]
-            if path == "/workstreams" and params and params.get("topic_id") == "topic-1":
+            if path == "/workplans" and params and params.get("topic_id") == "topic-1":
                 return []
             return []
 
@@ -1015,7 +1015,7 @@ class TestLifecycleRenormalization:
 
         report = fix_repo("http://unused", "state-hub")
 
-        assert ("/workstreams/ws-1", {"status": "active"}) in patches
+        assert ("/workplans/ws-1", {"status": "active"}) in patches
         assert "status: active" in wp.read_text(encoding="utf-8")
         assert any("C-23 fixed" in fix for fix in report.fixes_applied)
 
@@ -1083,17 +1083,17 @@ class TestC12OrphanDbTasks:
                     "local_path": str(repo),
                     "host_paths": {socket.gethostname(): str(repo)},
                 }
-            if path == "/workstreams/ws-1":
+            if path == "/workplans/ws-1":
                 return ws
             if path == "/tasks/task-linked":
                 return linked
             if path == "/tasks" and params == {"workstream_id": "ws-1"}:
                 return [linked, orphan]
-            if path == "/workstreams/ws-1/dependencies":
+            if path == "/workplans/ws-1/dependencies":
                 return []
-            if path == "/workstreams" and params == {"repo_id": "repo-1"}:
+            if path == "/workplans" and params == {"repo_id": "repo-1"}:
                 return [ws]
-            if path == "/workstreams" and params and params.get("topic_id") == "topic-1":
+            if path == "/workplans" and params and params.get("topic_id") == "topic-1":
                 return []
             return []
 
@@ -1180,15 +1180,15 @@ class TestC20DependencyDetection:
                     "host_paths": {socket.gethostname(): str(repo)},
                     "domain_slug": "financials",
                 }
-            if path == "/workstreams/base-ws":
+            if path == "/workplans/base-ws":
                 return {"id": "base-ws", "repo_id": "repo-1", "slug": "state-wp-0001", "title": "Base", "status": "active"}
-            if path == "/workstreams/dependent-ws":
+            if path == "/workplans/dependent-ws":
                 return {"id": "dependent-ws", "repo_id": "repo-1", "slug": "state-wp-0002", "title": "Dependent", "status": "active"}
             if path == "/tasks" and params and params.get("workstream_id") in {"base-ws", "dependent-ws"}:
                 return []
-            if path == "/workstreams/base-ws/dependencies":
+            if path == "/workplans/base-ws/dependencies":
                 return []
-            if path == "/workstreams/dependent-ws/dependencies":
+            if path == "/workplans/dependent-ws/dependencies":
                 return [
                     {
                         "id": "dep-1",
@@ -1198,7 +1198,7 @@ class TestC20DependencyDetection:
                         "relationship_type": "blocks",
                     }
                 ]
-            if path == "/workstreams" and params == {"repo_id": "repo-1"}:
+            if path == "/workplans" and params == {"repo_id": "repo-1"}:
                 return []
             return []
 
@@ -1250,18 +1250,18 @@ class TestC06WorkstreamCreation:
                 }
             if path == "/topics":
                 return [{"id": "topic-1", "domain_slug": "financials"}]
-            if path == "/workstreams" and params == {"slug": "state-wp-0001"}:
+            if path == "/workplans" and params == {"slug": "state-wp-0001"}:
                 return [{"id": "old-ws", "repo_id": "other-repo", "title": "Old Workplan"}]
-            if path == "/workstreams" and params == {"slug": "demo-repo-state-wp-0001"}:
+            if path == "/workplans" and params == {"slug": "demo-repo-state-wp-0001"}:
                 return []
-            if path == "/workstreams" and params == {"repo_id": "repo-1"}:
+            if path == "/workplans" and params == {"repo_id": "repo-1"}:
                 return []
-            if path == "/workstreams" and params and params.get("topic_id") == "topic-1":
+            if path == "/workplans" and params and params.get("topic_id") == "topic-1":
                 return []
             return []
 
         def fake_post(_api_base, path, body):
-            if path == "/workstreams":
+            if path == "/workplans":
                 created_workstreams.append(body)
                 return {"id": "new-ws", **body}
             if path == "/tasks":

tests/test_legacy_meter.py

@@ -1,5 +1,10 @@
 from __future__ import annotations
 
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+
 
 async def _create_domain(client, slug="legacy-domain", name="Legacy Domain"):
     r = await client.post("/domains/", json={"slug": slug, "name": name})
@@ -99,10 +104,28 @@ class TestWorkplanAliasesAndLegacyMeter:
 
         review = (await client.get("/legacy-meter/weekly-review")).json()
         assert review["activity_core_handoff"]["scheduler_owner"] == "activity-core"
+        assert review["activity_core_handoff"]["source_endpoint"] == "/legacy-meter/weekly-review"
+        assert "window_start" in review
+        assert "window_end" in review
+        assert isinstance(review["interfaces"], list)
+        assert isinstance(review["retirement_candidates"], list)
         candidates = _summary_by_key({"interfaces": review["retirement_candidates"]})
         assert "rest_api:GET /obsolete" in candidates
         assert candidates["rest_api:GET /obsolete"]["retirement_reason"] == "no measured usage in review window"
 
+        if os.environ.get("CAPTURE_LEGACY_METER_EVIDENCE"):
+            evidence_dir = Path(__file__).resolve().parents[1] / "docs" / "evidence"
+            evidence_dir.mkdir(parents=True, exist_ok=True)
+            today = datetime.now(tz=timezone.utc).strftime("%Y%m%d")
+            out = evidence_dir / f"legacy-meter-weekly-review-{today}.json"
+            payload = {
+                "captured_at": datetime.now(tz=timezone.utc).isoformat(),
+                "source": "pytest",
+                "workplan": "STATE-WP-0069",
+                "weekly_review": review,
+            }
+            out.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
+
     async def test_recent_usage_blocks_weekly_retirement_candidate(self, client):
         r = await client.post("/legacy-meter/usage", json={
             "interface_key": "rest_api:GET /old-but-used",
@@ -122,7 +145,7 @@ class TestWorkplanAliasesAndLegacyMeter:
         assert item["retirement_candidate"] is False
         assert item["retirement_reason"] == "1 call(s) in review window"
 
-    async def test_legacy_completion_event_is_metered_and_workplan_event_is_preferred(self, client):
+    async def test_workplan_finish_does_not_meter_legacy_completion_event(self, client):
         await _create_domain(client)
         topic = await _create_topic(client)
         wp = await _create_workplan(client, topic["id"])
@@ -140,9 +163,26 @@ class TestWorkplanAliasesAndLegacyMeter:
         assert items["rest_api:PATCH /workstreams/{workstream_id}"]["window"]["components"] == {
             "old-client": 1
         }
-        event = items["event_subject:org.statehub.workstream.completed"]
-        assert event["interface"]["replacement_ref"] == "org.statehub.workplan.completed"
-        assert event["window"]["components"] == {"state-hub.events": 1}
+        assert items.get("event_subject:org.statehub.workstream.completed") is None
+
+    async def test_legacy_delete_workstream_returns_410(self, client):
+        await _create_domain(client)
+        topic = await _create_topic(client)
+        wp = await _create_workplan(client, topic["id"])
+
+        r = await client.delete(
+            f"/workstreams/{wp['id']}",
+            headers={"X-StateHub-Component": "old-archiver"},
+        )
+        assert r.status_code == 410
+        assert "retired" in r.json()["detail"]
+        assert r.headers["Deprecation"] == "true"
+        assert r.headers["X-StateHub-Replacement"] == "DELETE /workplans/{workplan_id}"
+
+        summary = (await client.get("/legacy-meter/summary")).json()
+        item = _summary_by_key(summary)["rest_api:DELETE /workstreams/{workstream_id}"]
+        assert item["window"]["calls"] == 1
+        assert item["window"]["components"] == {"old-archiver": 1}
 
     async def test_legacy_workstream_id_query_param_on_tasks_is_metered(self, client):
         await _create_domain(client)