Finish CUST-WP-0011 and implement STATE-WP-0061 suggestion backlog.

Add persisted Suggestion entities with WSJF ranking, relevance bumps,
REST/MCP write surfaces, /suggestions dashboard page, and daily triage
digest integration. Document the cluster operating model, archive the
completed migration workplan, and seed WARDEN-WP-0012 routing scenarios.

api/routers/capability_requests.py

@@ -13,6 +13,7 @@ from api.models.capability_request import CapabilityRequest
 from api.models.domain import Domain
 from api.models.managed_repo import ManagedRepo
 from api.models.task import Task
+from api.services.suggestion_relevance import bump_matching_for_capability_request
 from api.schemas.capability_request import (
     CapabilityRequestAccept,
     CapabilityRequestCreate,
@@ -68,6 +69,23 @@ def _build_capability_request(
     )
 
 
+async def _on_capability_request_persisted(
+    session: AsyncSession,
+    req: CapabilityRequest,
+    body: CapabilityRequestCreate,
+) -> None:
+    await session.flush()
+    await bump_matching_for_capability_request(
+        session,
+        request_id=req.id,
+        title=body.title,
+        description=body.description or "",
+        capability_type=body.capability_type,
+        catalog_entry_id=req.catalog_entry_id,
+    )
+    await _notify_on_create(session, req, body)
+
+
 async def _notify_on_create(
     session: AsyncSession,
     req: CapabilityRequest,
@@ -405,7 +423,7 @@ router.include_router(
         request_read_schema=CapabilityRequestRead,
         route_request=_route_capability,
         build_request=_build_capability_request,
-        on_request_persisted=_notify_on_create,
+        on_request_persisted=_on_capability_request_persisted,
         check_transition=_check_transition,
         apply_accept_fields=_apply_accept_fields,
         after_accept=_notify_on_accept,

api/routers/state.py

@@ -21,8 +21,12 @@ from api.models.sbom_snapshot import SBOMSnapshot
 from api.models.task import Task, TaskPriority, TaskStatus
 from api.models.technical_debt import TechnicalDebt
 from api.models.topic import Topic, TopicStatus
+from api.models.suggestion import OPEN_SUGGESTION_STAGES, Suggestion
 from api.models.workplan import Workplan
 from api.models.workplan_dependency import WorkplanDependency
+from api.schemas.suggestion import RankedSuggestionDigest
+from api.services.suggestion_relevance import bump_suggestions_for_next_steps
+from api.services.suggestion_wsjf import compute_wsjf, suggestion_sort_key
 from api.schemas.decision import DecisionRead
 from api.schemas.domain import DomainSummary
 from api.schemas.progress_event import ProgressEventRead
@@ -329,7 +333,7 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
         ),
     )
 
-    next_steps = await _derive_next_steps(session)
+    next_steps, _ = await _derive_next_steps(session)
 
     # Domain summary stats
     domain_summaries = await _build_domain_summaries(session)
@@ -368,6 +372,8 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
         )
     )).scalar() or 0
 
+    ranked_suggestions = await _ranked_suggestion_digest(session, limit=10)
+
     result = StateSummary(
         generated_at=datetime.now(tz=timezone.utc),
         totals=totals,
@@ -387,6 +393,7 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
         contribution_counts=contribution_counts,
         licence_risk_count=licence_risk_count,
         open_capability_requests=open_cap_req_count,
+        ranked_suggestions=ranked_suggestions,
         open_workstreams=[
             WorkstreamWithDeps(
                 **{
@@ -700,7 +707,7 @@ async def _build_dashboard_overview(session: AsyncSession) -> DashboardOverview:
         waiting_tasks=[TaskRead.model_validate(t) for t in waiting],
         blocked_tasks=[TaskRead.model_validate(t) for t in waiting],
         recent_progress=[ProgressEventRead.model_validate(e) for e in recent],
-        next_steps=await _derive_next_steps(session),
+        next_steps=(await _derive_next_steps(session))[0],
         contribution_counts=contribution_counts,
         licence_risk_count=licence_risk_count,
         open_capability_requests=open_cap_req_count,
@@ -852,7 +859,33 @@ _PRIORITY_RANK = {
 }
 
 
-async def _derive_next_steps(session: AsyncSession) -> list[NextStep]:
+async def _ranked_suggestion_digest(
+    session: AsyncSession,
+    *,
+    limit: int = 10,
+) -> list[RankedSuggestionDigest]:
+    rows = (
+        await session.execute(
+            select(Suggestion).where(Suggestion.stage.in_(OPEN_SUGGESTION_STAGES))
+        )
+    ).scalars().all()
+    ranked = sorted(rows, key=suggestion_sort_key)[:limit]
+    return [
+        RankedSuggestionDigest(
+            id=s.id,
+            title=s.title,
+            stage=s.stage,
+            domain_slug=s.domain_slug,
+            origin_ref=s.origin_ref,
+            relevance=s.relevance,
+            wsjf=compute_wsjf(s),
+            last_requested_at=s.last_requested_at,
+        )
+        for s in ranked
+    ]
+
+
+async def _derive_next_steps(session: AsyncSession) -> tuple[list[NextStep], list[Suggestion]]:
     """Derive contextual next-action suggestions from current hub state.
 
     Two signal sources:
@@ -991,7 +1024,31 @@ async def _derive_next_steps(session: AsyncSession) -> list[NextStep]:
         ))
         seen_task_ids.add(task.id)
 
-    return steps
+    # Signal 3: open demand-weighted suggestions (needed-but-unmet backlog)
+    open_suggestions = sorted(
+        (
+            await session.execute(
+                select(Suggestion).where(Suggestion.stage.in_(OPEN_SUGGESTION_STAGES))
+            )
+        ).scalars().all(),
+        key=suggestion_sort_key,
+    )[:5]
+    for suggestion in open_suggestions:
+        steps.append(NextStep(
+            type="open_suggestion",
+            domain=suggestion.domain_slug,
+            workstream_id=suggestion.workplan_id,
+            workstream_title=None,
+            workstream_slug=suggestion.origin_ref,
+            task_id=None,
+            task_title=None,
+            message=(
+                f"Gated need '{suggestion.title}' (stage={suggestion.stage.value}, "
+                f"relevance={suggestion.relevance}) — vet or promote when unblocked"
+            ),
+        ))
+
+    return steps, open_suggestions
 
 
 async def _get_domain_slug_for_workstream(ws: Workplan | None, session: AsyncSession) -> str | None:
@@ -1019,8 +1076,12 @@ async def get_next_steps(session: AsyncSession = Depends(get_session)) -> list[N
     Returns suggestions based on:
     - Recently resolved decisions → first open task in the same workstream
     - Workstreams whose every dependency workstream is now finished -> first todo task
+    - Open demand-weighted suggestions (gated needs accruing relevance)
     """
-    return await _derive_next_steps(session)
+    steps, surfaced = await _derive_next_steps(session)
+    await bump_suggestions_for_next_steps(session, surfaced)
+    await session.commit()
+    return steps
 
 
 @router.get("/health")

api/routers/suggestions.py

@@ -0,0 +1,259 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.domain import Domain
+from api.models.suggestion import (
+    OPEN_SUGGESTION_STAGES,
+    Suggestion,
+    SuggestionNote,
+    SuggestionStage,
+)
+from api.models.task import Task, TaskPriority, TaskStatus
+from api.schemas.suggestion import (
+    SuggestionBumpRelevance,
+    SuggestionCreate,
+    SuggestionDecline,
+    SuggestionNoteRead,
+    SuggestionPromote,
+    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}
+
+
+async def _resolve_domain_id(slug: str, session: AsyncSession) -> uuid.UUID:
+    row = await session.execute(
+        select(Domain.id).where(Domain.slug == slug, Domain.status == "active")
+    )
+    domain_id = row.scalar_one_or_none()
+    if domain_id is None:
+        valid = [r[0] for r in (await session.execute(
+            select(Domain.slug).where(Domain.status == "active")
+        )).all()]
+        raise HTTPException(
+            status_code=422,
+            detail=f"Unknown domain '{slug}'. Valid domains: {sorted(valid)}",
+        )
+    return domain_id
+
+
+def _enrich_read(suggestion: Suggestion) -> SuggestionRead:
+    data = SuggestionRead.model_validate(suggestion)
+    data.cost_of_delay = cost_of_delay(suggestion)
+    data.wsjf = compute_wsjf(suggestion)
+    return data
+
+
+async def _get_suggestion_or_404(
+    suggestion_id: uuid.UUID,
+    session: AsyncSession,
+) -> Suggestion:
+    suggestion = await session.get(Suggestion, suggestion_id)
+    if suggestion is None:
+        raise HTTPException(status_code=404, detail="Suggestion not found")
+    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,
+    stage: SuggestionStage | None = None,
+    include_terminal: bool = Query(False),
+    rank: str | None = Query(None),
+    limit: int = Query(100, ge=1, le=500),
+    session: AsyncSession = Depends(get_session),
+) -> list[SuggestionRead]:
+    q = select(Suggestion)
+    if domain:
+        domain_id = await _resolve_domain_id(domain, session)
+        q = q.where(Suggestion.domain_id == domain_id)
+    if stage:
+        q = q.where(Suggestion.stage == stage)
+    elif not include_terminal:
+        q = q.where(Suggestion.stage.in_(OPEN_SUGGESTION_STAGES))
+    result = await session.execute(q)
+    suggestions = list(result.scalars().all())
+    if rank == "wsjf":
+        suggestions.sort(key=suggestion_sort_key)
+    else:
+        suggestions.sort(key=lambda s: s.created_at)
+    return [_enrich_read(s) for s in suggestions[:limit]]
+
+
+@router.post("/", response_model=SuggestionRead, status_code=status.HTTP_201_CREATED)
+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)
+
+
+@router.get("/{suggestion_id}", response_model=SuggestionRead)
+async def get_suggestion(
+    suggestion_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> SuggestionRead:
+    suggestion = await _get_suggestion_or_404(suggestion_id, session)
+    return _enrich_read(suggestion)
+
+
+@router.post("/{suggestion_id}/vet", response_model=SuggestionRead)
+async def vet_suggestion(
+    suggestion_id: uuid.UUID,
+    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)
+
+
+@router.post("/{suggestion_id}/decline", response_model=SuggestionRead)
+async def decline_suggestion(
+    suggestion_id: uuid.UUID,
+    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)
+
+
+@router.post("/{suggestion_id}/promote", response_model=SuggestionRead)
+async def promote_suggestion_to_task(
+    suggestion_id: uuid.UUID,
+    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)
+
+
+@router.post("/{suggestion_id}/bump-relevance", response_model=SuggestionRead)
+async def bump_suggestion_relevance(
+    suggestion_id: uuid.UUID,
+    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)
+
+
+@router.get("/{suggestion_id}/notes", response_model=list[SuggestionNoteRead])
+async def list_suggestion_notes(
+    suggestion_id: uuid.UUID,
+    session: AsyncSession = Depends(get_session),
+) -> list[SuggestionNote]:
+    await _get_suggestion_or_404(suggestion_id, session)
+    result = await session.execute(
+        select(SuggestionNote)
+        .where(SuggestionNote.suggestion_id == suggestion_id)
+        .order_by(SuggestionNote.created_at)
+    )
+    return list(result.scalars().all())