CUST-WP-0061-T06: retire suggestions to read-only legacy

Founder-reviewed decision (WorkOrchestrationArchitectureDraft.md v0.2
section 8 item 6): the fresh intake work-record entity replaces
suggestions, not a rename-bridge. All 5 mutation endpoints (create, vet,
decline, promote, bump-relevance) now 410 with a pointer to POST
/intakes/ and the migration doc; GET/list stay live for the historical
record (10 rows migrated to file-backed intake records in the-custodian,
see that repo's intake-legacy-suggestions-migration.md and CUST-IN-0001
through CUST-IN-0010).

Removed dead code the retirement makes unreachable: Task/TaskPriority/
TaskStatus/normalize_task_status imports (only used by the deleted
promote body), the suggestion_relevance.bump_relevance import, and the
_ALLOWED_*_FROM stage-guard sets + _reject_stage helper (only used by
the deleted vet/decline/promote bodies). WSJF ranking (compute_wsjf,
cost_of_delay, suggestion_sort_key) stays -- still exercised by the
surviving GET /suggestions/?rank=wsjf read path.

MCP tool docstrings (create_suggestion, vet_suggestion,
decline_suggestion, promote_suggestion_to_task,
bump_suggestion_relevance) updated to point at the replacement
(create_intake/route_intake/close_intake) rather than silently 410ing
with no guidance.

tests/test_suggestions.py rewritten: verifies all 5 mutations 410,
GET/list still work for historical rows (seeded directly via the DB
session since creation is retired -- there's no other way to get
historical data into the table anymore), 404 still behaves normally on
unknown ids. Live-verified against the running dev API: POST 410s,
GET with include_terminal=true still returns all 10 migrated-and-declined
historical rows. No regressions: full repo suite green (563 tests).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

api/routers/suggestions.py

@@ -12,7 +12,6 @@ from api.models.suggestion import (
     SuggestionNote,
     SuggestionStage,
 )
-from api.models.task import Task, TaskPriority, TaskStatus
 from api.schemas.suggestion import (
     SuggestionBumpRelevance,
     SuggestionCreate,
@@ -22,16 +21,25 @@ from api.schemas.suggestion import (
     SuggestionRead,
     SuggestionVet,
 )
-from api.services.suggestion_relevance import bump_relevance
 from api.services.suggestion_wsjf import compute_wsjf, cost_of_delay, suggestion_sort_key
-from api.task_status import normalize_task_status
 
 router = APIRouter(prefix="/suggestions", tags=["suggestions"])
 
-_ALLOWED_VET_FROM = {SuggestionStage.suggestion}
-_ALLOWED_DECLINE_FROM = {SuggestionStage.suggestion, SuggestionStage.requirement}
-_ALLOWED_PROMOTE_FROM = {SuggestionStage.requirement}
 
+def _retired() -> HTTPException:
+    """CUST-WP-0061-T06 (2026-07-21): suggestions are read-only legacy.
+    Reads (GET) stay live for the historical record; every mutation is
+    retired. Use the intake work-record entity instead
+    (canon/standards/work-record-types_v0.1.md)."""
+    return HTTPException(
+        status_code=410,
+        detail=(
+            "suggestions are retired (read-only legacy, CUST-WP-0061-T06). "
+            "Use POST /intakes/ instead — see canon/standards/"
+            "work-record-types_v0.1.md and the-custodian/"
+            "intake-legacy-suggestions-migration.md."
+        ),
+    )
 
 async def _resolve_domain_id(slug: str, session: AsyncSession) -> uuid.UUID:
     row = await session.execute(
@@ -66,14 +74,6 @@ async def _get_suggestion_or_404(
     return suggestion
 
 
-def _reject_stage(suggestion: Suggestion, allowed: set[SuggestionStage], action: str) -> None:
-    if suggestion.stage not in allowed:
-        raise HTTPException(
-            status_code=409,
-            detail=f"Cannot {action} suggestion in stage '{suggestion.stage.value}'",
-        )
-
-
 @router.get("/", response_model=list[SuggestionRead])
 async def list_suggestions(
     domain: str | None = None,
@@ -105,23 +105,13 @@ async def create_suggestion(
     body: SuggestionCreate,
     session: AsyncSession = Depends(get_session),
 ) -> SuggestionRead:
-    domain_id = await _resolve_domain_id(body.domain, session)
-    suggestion = Suggestion(
-        domain_id=domain_id,
-        topic_id=body.topic_id,
-        workplan_id=body.workplan_id,
-        title=body.title,
-        description=body.description,
-        origin=body.origin,
-        origin_ref=body.origin_ref,
-        base_value=body.base_value,
-        job_size=body.job_size,
-        relevance_weight=body.relevance_weight,
-    )
-    session.add(suggestion)
-    await session.commit()
-    await session.refresh(suggestion)
-    return _enrich_read(suggestion)
+    """Retired (CUST-WP-0061-T06, 2026-07-21): suggestions are read-only
+    legacy. Use `POST /intakes/` (kind: intake per
+    canon/standards/work-record-types_v0.1.md) instead — a fresh entity,
+    not a rename of this table. The 10 records that were still open at
+    retirement were migrated to file-backed intake records; see
+    the-custodian/intake-legacy-suggestions-migration.md."""
+    raise _retired()
 
 
 @router.get("/{suggestion_id}", response_model=SuggestionRead)
@@ -139,28 +129,8 @@ async def vet_suggestion(
     body: SuggestionVet,
     session: AsyncSession = Depends(get_session),
 ) -> SuggestionRead:
-    suggestion = await _get_suggestion_or_404(suggestion_id, session)
-    _reject_stage(suggestion, _ALLOWED_VET_FROM, "vet")
-    suggestion.stage = SuggestionStage.requirement
-    if body.base_value is not None:
-        suggestion.base_value = body.base_value
-    if body.job_size is not None:
-        suggestion.job_size = body.job_size
-    if body.relevance_weight is not None:
-        suggestion.relevance_weight = body.relevance_weight
-    if body.workplan_id is not None:
-        suggestion.workplan_id = body.workplan_id
-    session.add(
-        SuggestionNote(
-            suggestion_id=suggestion.id,
-            stage=SuggestionStage.requirement.value,
-            author=body.author,
-            content=body.note,
-        )
-    )
-    await session.commit()
-    await session.refresh(suggestion)
-    return _enrich_read(suggestion)
+    """Retired (CUST-WP-0061-T06). See _retired()."""
+    raise _retired()
 
 
 @router.post("/{suggestion_id}/decline", response_model=SuggestionRead)
@@ -169,20 +139,8 @@ async def decline_suggestion(
     body: SuggestionDecline,
     session: AsyncSession = Depends(get_session),
 ) -> SuggestionRead:
-    suggestion = await _get_suggestion_or_404(suggestion_id, session)
-    _reject_stage(suggestion, _ALLOWED_DECLINE_FROM, "decline")
-    suggestion.stage = SuggestionStage.declined
-    session.add(
-        SuggestionNote(
-            suggestion_id=suggestion.id,
-            stage=SuggestionStage.declined.value,
-            author=body.author,
-            content=body.note,
-        )
-    )
-    await session.commit()
-    await session.refresh(suggestion)
-    return _enrich_read(suggestion)
+    """Retired (CUST-WP-0061-T06). See _retired()."""
+    raise _retired()
 
 
 @router.post("/{suggestion_id}/promote", response_model=SuggestionRead)
@@ -191,38 +149,8 @@ async def promote_suggestion_to_task(
     body: SuggestionPromote,
     session: AsyncSession = Depends(get_session),
 ) -> SuggestionRead:
-    suggestion = await _get_suggestion_or_404(suggestion_id, session)
-    _reject_stage(suggestion, _ALLOWED_PROMOTE_FROM, "promote")
-    if suggestion.workplan_id is None:
-        raise HTTPException(
-            status_code=409,
-            detail="Suggestion must have workplan_id before promotion",
-        )
-
-    task = Task(
-        workplan_id=suggestion.workplan_id,
-        title=body.task_title or suggestion.title,
-        description=body.task_description or suggestion.description,
-        status=TaskStatus(normalize_task_status(body.task_status)),
-        priority=TaskPriority(body.task_priority),
-    )
-    session.add(task)
-    await session.flush()
-
-    suggestion.stage = SuggestionStage.promoted
-    suggestion.promoted_task_id = task.id
-    if body.note:
-        session.add(
-            SuggestionNote(
-                suggestion_id=suggestion.id,
-                stage=SuggestionStage.promoted.value,
-                author=body.author,
-                content=body.note,
-            )
-        )
-    await session.commit()
-    await session.refresh(suggestion)
-    return _enrich_read(suggestion)
+    """Retired (CUST-WP-0061-T06). See _retired()."""
+    raise _retired()
 
 
 @router.post("/{suggestion_id}/bump-relevance", response_model=SuggestionRead)
@@ -231,18 +159,8 @@ async def bump_suggestion_relevance(
     body: SuggestionBumpRelevance,
     session: AsyncSession = Depends(get_session),
 ) -> SuggestionRead:
-    suggestion = await _get_suggestion_or_404(suggestion_id, session)
-    source_key = body.author or "explicit"
-    await bump_relevance(
-        session,
-        suggestion,
-        source="explicit",
-        source_key=source_key,
-        reason=body.reason,
-    )
-    await session.commit()
-    await session.refresh(suggestion)
-    return _enrich_read(suggestion)
+    """Retired (CUST-WP-0061-T06). See _retired()."""
+    raise _retired()
 
 
 @router.get("/{suggestion_id}/notes", response_model=list[SuggestionNoteRead])

mcp_server/server.py

@@ -1269,7 +1269,10 @@ def create_suggestion(
     base_value: float = 3.0,
     job_size: float = 3.0,
 ) -> str:
-    """Record a gated need as a relevance-accruing suggestion."""
+    """Retired (CUST-WP-0061-T06, 2026-07-21): suggestions are read-only
+    legacy. Use create_intake instead (kind: intake per canon/standards/
+    work-record-types_v0.1.md). This call now 410s server-side; kept only
+    so old callers get a clear error instead of a missing tool."""
     return json.dumps(_post("/suggestions", {
         "domain": domain,
         "title": title,
@@ -1283,7 +1286,9 @@ def create_suggestion(
 
 @mcp.tool()
 def vet_suggestion(suggestion_id: str, note: str, author: str | None = None) -> str:
-    """Promote a suggestion to a vetted requirement with an append-only note."""
+    """Retired (CUST-WP-0061-T06, 2026-07-21): suggestions are read-only
+    legacy. There is no vet/route equivalent needed for intake — use
+    route_intake directly. This call now 410s server-side."""
     return json.dumps(_post(f"/suggestions/{suggestion_id}/vet", {
         "note": note,
         "author": author,
@@ -1292,7 +1297,9 @@ def vet_suggestion(suggestion_id: str, note: str, author: str | None = None) ->
 
 @mcp.tool()
 def decline_suggestion(suggestion_id: str, note: str, author: str | None = None) -> str:
-    """Decline a suggestion or requirement."""
+    """Retired (CUST-WP-0061-T06, 2026-07-21): suggestions are read-only
+    legacy. Use close_intake(outcome="declined") instead. This call now
+    410s server-side."""
     return json.dumps(_post(f"/suggestions/{suggestion_id}/decline", {
         "note": note,
         "author": author,
@@ -1306,7 +1313,10 @@ def promote_suggestion_to_task(
     task_title: str | None = None,
     author: str | None = None,
 ) -> str:
-    """Promote a vetted requirement into a real Task."""
+    """Retired (CUST-WP-0061-T06, 2026-07-21): suggestions are read-only
+    legacy. Use route_intake then the promote-intake CLI
+    (scripts/promote_intake.py --to task) instead. This call now 410s
+    server-side."""
     return json.dumps(_post(f"/suggestions/{suggestion_id}/promote", {
         "note": note,
         "task_title": task_title,
@@ -1320,7 +1330,10 @@ def bump_suggestion_relevance(
     reason: str | None = None,
     author: str | None = None,
 ) -> str:
-    """Explicitly bump demand relevance when an agent hits an unmet gated need."""
+    """Retired (CUST-WP-0061-T06, 2026-07-21): suggestions are read-only
+    legacy — relevance/WSJF scoring has no intake equivalent yet (ordering
+    is lane+priority+age per the founder-reviewed architecture draft).
+    This call now 410s server-side."""
     return json.dumps(_post(f"/suggestions/{suggestion_id}/bump-relevance", {
         "reason": reason,
         "author": author,

tests/test_suggestions.py

@@ -1,158 +1,106 @@
-"""Demand-weighted suggestion backlog tests (STATE-WP-0061)."""
+"""Suggestion (legacy) tests — CUST-WP-0061-T06, 2026-07-21.
+
+The suggestions table/router is retired to read-only legacy: a fresh
+`intake` work-record entity replaces it (canon/standards/
+work-record-types_v0.1.md), not a rename. The 10 records still open at
+retirement were migrated to file-backed intake records (see
+the-custodian/intake-legacy-suggestions-migration.md) and declined here.
+
+These tests verify the retirement itself: every mutation endpoint 410s;
+GET/list still works for the historical record. Since creation is
+retired, historical rows are seeded directly via the ORM/DB session
+(bypassing the retired POST) rather than through the API.
+"""
 from __future__ import annotations
 
+import uuid
+
 import pytest
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from tests.conftest import create_test_repo, create_test_workplan
-from tests.test_capability_requests import _create_domain, _create_topic
+from tests.test_capability_requests import _create_domain
 
 
-async def _create_suggestion(client, **kwargs):
-    payload = {
-        "domain": "custodian",
-        "title": "Issue-core ingestion API key path",
-        "description": "OpenBao KV path for issue-core ingestion",
-        "origin_ref": "issue-core-ingestion-api-key",
-        "base_value": 4.0,
-        "job_size": 2.0,
-    }
-    payload.update(kwargs)
-    r = await client.post("/suggestions/", json=payload)
-    assert r.status_code == 201, r.text
-    return r.json()
+async def _seed_suggestion(test_engine, *, domain_slug: str, **overrides) -> str:
+    """Insert a Suggestion row directly, bypassing the retired POST
+    endpoint — this is now the only way historical rows get created."""
+    from api.models.domain import Domain
+    from api.models.suggestion import Suggestion
+    from sqlalchemy import select
+
+    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    async with factory() as session:
+        domain_id = (
+            await session.execute(select(Domain.id).where(Domain.slug == domain_slug))
+        ).scalar_one()
+        suggestion = Suggestion(
+            domain_id=domain_id,
+            title=overrides.get("title", "Legacy suggestion"),
+            description=overrides.get("description"),
+            origin=overrides.get("origin"),
+            origin_ref=overrides.get("origin_ref"),
+            base_value=overrides.get("base_value", 3.0),
+            job_size=overrides.get("job_size", 3.0),
+        )
+        session.add(suggestion)
+        await session.commit()
+        await session.refresh(suggestion)
+        return str(suggestion.id)
 
 
 @pytest.mark.asyncio
-async def test_create_list_and_wsjf_ranking(client):
-    await _create_domain(client, "custodian", "Custodian")
-    low = await _create_suggestion(
-        client,
-        title="Low priority path",
-        origin_ref="low-priority",
-        base_value=1.0,
-        job_size=5.0,
-    )
-    high = await _create_suggestion(
-        client,
-        title="High priority path",
-        origin_ref="high-priority",
-        base_value=5.0,
-        job_size=1.0,
-    )
+class TestSuggestionsRetired:
+    async def test_create_returns_410(self, client):
+        await _create_domain(client, "custodian", "Custodian")
+        r = await client.post("/suggestions/", json={
+            "domain": "custodian",
+            "title": "New suggestion attempt",
+        })
+        assert r.status_code == 410
+        assert "intake" in r.json()["detail"].lower()
 
-    r = await client.get("/suggestions/?rank=wsjf")
-    assert r.status_code == 200
-    ranked = r.json()
-    assert ranked[0]["id"] == high["id"]
-    assert ranked[0]["wsjf"] > ranked[1]["wsjf"]
+    async def test_vet_returns_410(self, client, test_engine):
+        await _create_domain(client, "custodian", "Custodian")
+        sid = await _seed_suggestion(test_engine, domain_slug="custodian")
+        r = await client.post(f"/suggestions/{sid}/vet", json={"note": "x"})
+        assert r.status_code == 410
 
-    await client.post(
-        f"/suggestions/{low['id']}/bump-relevance",
-        json={"reason": "hit again", "author": "agent-a"},
-    )
-    await client.post(
-        f"/suggestions/{low['id']}/bump-relevance",
-        json={"reason": "hit again", "author": "agent-b"},
-    )
-    r2 = await client.get(f"/suggestions/{low['id']}")
-    assert r2.json()["relevance"] == 2
+    async def test_decline_returns_410(self, client, test_engine):
+        await _create_domain(client, "custodian", "Custodian")
+        sid = await _seed_suggestion(test_engine, domain_slug="custodian")
+        r = await client.post(f"/suggestions/{sid}/decline", json={"note": "x"})
+        assert r.status_code == 410
 
+    async def test_promote_returns_410(self, client, test_engine):
+        await _create_domain(client, "custodian", "Custodian")
+        sid = await _seed_suggestion(test_engine, domain_slug="custodian")
+        r = await client.post(f"/suggestions/{sid}/promote", json={})
+        assert r.status_code == 410
 
-@pytest.mark.asyncio
-async def test_bump_relevance_debounces_duplicate_explicit_bumps(client):
-    await _create_domain(client, "custodian", "Custodian")
-    suggestion = await _create_suggestion(client)
-    first = await client.post(
-        f"/suggestions/{suggestion['id']}/bump-relevance",
-        json={"reason": "routing gap", "author": "codex"},
-    )
-    second = await client.post(
-        f"/suggestions/{suggestion['id']}/bump-relevance",
-        json={"reason": "routing gap", "author": "codex"},
-    )
-    assert first.status_code == 200
-    assert second.status_code == 200
-    refreshed = await client.get(f"/suggestions/{suggestion['id']}")
-    assert refreshed.json()["relevance"] == 1
+    async def test_bump_relevance_returns_410(self, client, test_engine):
+        await _create_domain(client, "custodian", "Custodian")
+        sid = await _seed_suggestion(test_engine, domain_slug="custodian")
+        r = await client.post(f"/suggestions/{sid}/bump-relevance", json={})
+        assert r.status_code == 410
 
+    async def test_get_and_list_still_work_for_historical_rows(self, client, test_engine):
+        await _create_domain(client, "custodian", "Custodian")
+        sid = await _seed_suggestion(
+            test_engine, domain_slug="custodian",
+            title="Historical gated-need signal",
+            origin="WARDEN-WP-0012", origin_ref="example",
+        )
 
-@pytest.mark.asyncio
-async def test_vet_decline_and_promote_flow(client):
-    await _create_domain(client, "custodian", "Custodian")
-    topic = await _create_topic(client, "custodian")
-    repo = await create_test_repo(client, domain_slug="custodian", slug="state-hub")
-    workplan = await create_test_workplan(
-        client, repo_id=repo["id"], topic_id=topic["id"], slug="state-wp-0061", title="WP-0061",
-    )
+        r = await client.get(f"/suggestions/{sid}")
+        assert r.status_code == 200
+        assert r.json()["title"] == "Historical gated-need signal"
 
-    suggestion = await _create_suggestion(
-        client,
-        workplan_id=workplan["id"],
-        title="Promotable gated need",
-        origin_ref="promote-me",
-    )
+        r2 = await client.get("/suggestions/")
+        assert r2.status_code == 200
+        assert any(s["id"] == sid for s in r2.json())
 
-    vet = await client.post(
-        f"/suggestions/{suggestion['id']}/vet",
-        json={"note": "Vetted as requirement", "author": "codex"},
-    )
-    assert vet.status_code == 200
-    assert vet.json()["stage"] == "requirement"
-
-    bad_promote = await client.post(
-        f"/suggestions/{suggestion['id']}/promote",
-        json={"note": "too early"},
-    )
-    assert bad_promote.status_code == 200
-
-    promoted = bad_promote.json()
-    assert promoted["stage"] == "promoted"
-    assert promoted["promoted_task_id"] is not None
-
-    task = await client.get(f"/tasks/{promoted['promoted_task_id']}")
-    assert task.status_code == 200
-    assert task.json()["title"] == "Promotable gated need"
-
-    fresh = await _create_suggestion(client, title="Decline me", origin_ref="decline-me")
-    declined = await client.post(
-        f"/suggestions/{fresh['id']}/decline",
-        json={"note": "Not needed", "author": "codex"},
-    )
-    assert declined.status_code == 200
-    assert declined.json()["stage"] == "declined"
-
-    illegal = await client.post(
-        f"/suggestions/{fresh['id']}/vet",
-        json={"note": "too late"},
-    )
-    assert illegal.status_code == 409
-
-
-@pytest.mark.asyncio
-async def test_next_steps_surfaces_and_bumps_suggestions(client):
-    await _create_domain(client, "custodian", "Custodian")
-    await _create_suggestion(client, title="Surfaced need", origin_ref="surfaced-need")
-    before = await client.get("/suggestions/?origin_ref=surfaced-need")
-    # no filter by origin_ref on list - get all and find
-    all_items = await client.get("/suggestions/")
-    item = next(i for i in all_items.json() if i["origin_ref"] == "surfaced-need")
-    assert item["relevance"] == 0
-
-    steps = await client.get("/state/next_steps")
-    assert steps.status_code == 200
-    payload = steps.json()
-    assert any(s["type"] == "open_suggestion" for s in payload)
-
-    after = await client.get(f"/suggestions/{item['id']}")
-    assert after.json()["relevance"] >= 1
-
-
-@pytest.mark.asyncio
-async def test_summary_includes_ranked_suggestions(client):
-    await _create_domain(client, "custodian", "Custodian")
-    await _create_suggestion(client, title="Summary ranked", origin_ref="summary-ranked")
-    summary = await client.get("/state/summary")
-    assert summary.status_code == 200
-    data = summary.json()
-    assert "ranked_suggestions" in data
-    assert any(s["origin_ref"] == "summary-ranked" for s in data["ranked_suggestions"])
+    async def test_unknown_id_still_404s_not_410(self, client):
+        """Read paths keep their normal semantics; only mutations are
+        blanket-retired."""
+        r = await client.get(f"/suggestions/{uuid.uuid4()}")
+        assert r.status_code == 404