feat(proposals): supersede non-winning chain links (feat_overnight_final_solution_phase3)

backend/app/api/v1/proposals.py

@@ -376,6 +376,7 @@ async def list_proposals_endpoint(
     cursor: Annotated[str | None, Query()] = None,
     limit: Annotated[int, Query(ge=1, le=MAX_PAGE_LIMIT)] = DEFAULT_PAGE_LIMIT,
     sort: Annotated[ProposalSortKey | None, Query()] = None,
+    include_superseded: Annotated[bool, Query()] = False,
 ) -> ProposalsListResponse:
     """List proposals with cursor pagination + filters.
 
@@ -384,6 +385,11 @@ async def list_proposals_endpoint(
     study-detail page's pending-proposal lookup). Both reject invalid
     UUIDs with 422 via FastAPI's UUID parsing. ``?sort=`` (Story 1.3) is
     a :data:`ProposalSortKey` value with sort-aware cursor.
+
+    Phase 3 D-15 revised: ``?include_superseded`` defaults to ``False``;
+    when ``False`` AND no ``?status=`` is set, the response omits
+    ``superseded`` rows. Explicit ``?status=`` always beats implicit
+    ``include_superseded`` (single-value backward compat preserved).
     """
     parsed_sort = parse_sort(sort, _PROPOSAL_SORT_COLUMNS)
     decoded_cursor: tuple[object, str] | None = None
@@ -408,6 +414,7 @@ async def list_proposals_endpoint(
             study_id=study_id_str,
             is_last_merged=is_last_merged,
             sort=sort,
+            include_superseded=include_superseded,
         )
     )
     has_more = len(rows) > limit
@@ -421,6 +428,7 @@ async def list_proposals_endpoint(
         template_id=template_id_str,
         study_id=study_id_str,
         is_last_merged=is_last_merged,
+        include_superseded=include_superseded,
     )
     response.headers["X-Total-Count"] = str(total)
     next_cursor: str | None = None
@@ -531,4 +539,53 @@ async def open_pr_endpoint(
     )
 
 
+# ---------------------------------------------------------------------------
+# POST /api/v1/proposals/{id}/reinstate  (Phase 3, FR-6)
+# ---------------------------------------------------------------------------
+
+
+@router.post(
+    "/proposals/{proposal_id}/reinstate",
+    response_model=ProposalDetail,
+    tags=["proposals"],
+)
+async def reinstate_proposal_endpoint(
+    proposal_id: str,
+    db: Annotated[AsyncSession, Depends(get_db)],
+) -> ProposalDetail:
+    """Phase 3 FR-6: ``superseded → pending`` transition.
+
+    Mirrors :func:`reject_proposal_endpoint` (D-17 — read-check-mutate so
+    404 vs 409 stays deterministic). Reuses ``INVALID_STATE_TRANSITION``
+    per D-16; emits ``chain_proposal_reinstated`` structlog AFTER commit
+    per D-19.
+    """
+    # Single read-check-mutate via the repo helper (Gemini perf finding):
+    # catch its LookupError / InvalidStateTransition directly and reuse the
+    # returned row — no separate pre-read or post-commit refresh needed.
+    try:
+        proposal = await repo.reinstate_from_superseded(db, proposal_id=proposal_id)
+    except LookupError as exc:
+        raise _err(404, "PROPOSAL_NOT_FOUND", f"proposal {proposal_id} not found", False) from exc
+    except InvalidStateTransition as exc:
+        raise _err(
+            409,
+            "INVALID_STATE_TRANSITION",
+            f"proposal {proposal_id} is in status {exc.current_status!r}; "
+            "only 'superseded' proposals can be reinstated",
+            False,
+        ) from exc
+    await db.commit()
+    # D-19: emit AFTER commit succeeds (pre-commit emission risks the
+    # transaction rolling back while the log claims a durable transition).
+    logger.info(
+        "chain_proposal_reinstated",
+        event_type="chain_proposal_reinstated",
+        proposal_id=proposal_id,
+        study_id=proposal.study_id,
+        prior_status="superseded",
+    )
+    return await _assemble_proposal_detail(db, proposal)
+
+
 __all__ = ["router"]

backend/app/api/v1/schemas.py

@@ -1376,7 +1376,7 @@ class CalibrationResponse(BaseModel):
 # feat_digest_proposal Epic 3 schemas (Stories 3.1-3.4)
 # ---------------------------------------------------------------------------
 
-ProposalStatusWire = Literal["pending", "pr_opened", "pr_merged", "rejected"]
+ProposalStatusWire = Literal["pending", "pr_opened", "pr_merged", "rejected", "superseded"]
 """Wire values for ``proposals.status`` filter on ``GET /api/v1/proposals``.
 
 Values must match backend/app/db/models/proposal.py CHECK

backend/app/db/models/proposal.py

@@ -39,7 +39,7 @@ class Proposal(Base):
     __tablename__ = "proposals"
     __table_args__ = (
         CheckConstraint(
-            "status IN ('pending', 'pr_opened', 'pr_merged', 'rejected')",
+            "status IN ('pending', 'pr_opened', 'pr_merged', 'rejected', 'superseded')",
             name="proposals_status_check",
         ),
         CheckConstraint(

backend/app/db/repo/__init__.py

@@ -77,6 +77,7 @@
 from backend.app.db.repo.proposal import (
     InvalidStateTransition,
     ProposalStatusFilter,
+    bulk_mark_superseded,
     count_proposals,
     create_proposal,
     get_proposal,
@@ -90,6 +91,7 @@
     mark_proposal_pr_merged_from_closed,
     mark_proposal_pr_opened,
     mark_proposal_pr_reopened,
+    reinstate_from_superseded,
     reject_proposal,
     set_proposal_pr_open_error,
     stamp_proposal_last_polled_at,
@@ -222,11 +224,13 @@
     # feat_digest_proposal Story 1.2 (digest repo + proposal repo extensions)
     "InvalidStateTransition",
     "ProposalStatusFilter",
+    "bulk_mark_superseded",
     "count_proposals",
     "create_digest",
     "get_digest_for_study",
     "list_pending_proposals_for_boot_scan",
     "list_proposals_paginated",
+    "reinstate_from_superseded",
     "reject_proposal",
     "update_proposal_for_digest",
     # feat_github_pr_worker Story 1.1 (config_repo list/count + proposal pr-transition helpers)

backend/app/db/repo/proposal.py

@@ -53,7 +53,7 @@
 
 # Wire values for `?status=` filter on `GET /api/v1/proposals`.
 # Values must match backend/app/db/models/proposal.py CHECK proposals_status_check.
-ProposalStatusFilter = Literal["pending", "pr_opened", "pr_merged", "rejected"]
+ProposalStatusFilter = Literal["pending", "pr_opened", "pr_merged", "rejected", "superseded"]
 # Per chore_proposals_source_filter_server_side: distinguishes proposals
 # derived from a completed study (study_id NOT NULL) from operator-authored
 # manual proposals (study_id NULL).
@@ -173,6 +173,7 @@ async def list_proposals_paginated(
     study_id: str | None = None,
     is_last_merged: bool | None = None,
     sort: str | None = None,
+    include_superseded: bool = False,
 ) -> Sequence[Proposal]:
     """Cursor-paginated proposal list. Sort-aware (Story 1.3).
 
@@ -185,11 +186,20 @@ async def list_proposals_paginated(
     (used by the study-detail page's pending-proposal lookup).
     ``is_last_merged`` (feat_config_repo_baseline_tracking FR-6) filters
     to proposals tracked (or not) as some config_repo's live pointer.
+
+    Phase 3 D-15 revised: ``include_superseded`` defaults to ``False``;
+    when ``False`` AND ``status is None``, the implicit filter
+    ``Proposal.status != 'superseded'`` is applied so the default list
+    omits non-winning chain links. Explicit ``status`` overrides this
+    (e.g., ``status='superseded'`` returns only superseded rows).
     """
     parsed_sort: ParsedSort | None = parse_sort(sort, _PROPOSAL_SORT_COLUMNS)
     stmt = select(Proposal)
     if status is not None:
         stmt = stmt.where(Proposal.status == status)
+    elif not include_superseded:
+        # Phase 3 D-15 revised: default response excludes superseded rows.
+        stmt = stmt.where(Proposal.status != "superseded")
     if cluster_id is not None:
         stmt = stmt.where(Proposal.cluster_id == cluster_id)
     if template_id is not None:
@@ -224,17 +234,21 @@ async def count_proposals(
     template_id: str | None = None,
     study_id: str | None = None,
     is_last_merged: bool | None = None,
+    include_superseded: bool = False,
 ) -> int:
     """COUNT(*) for the ``X-Total-Count`` header on ``GET /api/v1/proposals``.
 
     ``template_id`` filter (Story 1.5) narrows by FK. ``study_id`` filter
     narrows to a single study. ``is_last_merged``
     (feat_config_repo_baseline_tracking FR-6) restricts to the live-pointer
-    set or its complement.
+    set or its complement. ``include_superseded`` mirrors the rule on
+    :func:`list_proposals_paginated` — Phase 3 D-15 revised.
     """
     stmt = select(func.count()).select_from(Proposal)
     if status is not None:
         stmt = stmt.where(Proposal.status == status)
+    elif not include_superseded:
+        stmt = stmt.where(Proposal.status != "superseded")
     if cluster_id is not None:
         stmt = stmt.where(Proposal.cluster_id == cluster_id)
     if template_id is not None:
@@ -632,3 +646,67 @@ async def hard_delete_proposal(db: AsyncSession, proposal_id: str) -> bool:
     await db.delete(existing)
     await db.flush()
     return True
+
+
+# ---------------------------------------------------------------------------
+# feat_overnight_final_solution_phase3 Story 1.2 — supersession + reinstate
+# ---------------------------------------------------------------------------
+
+
+async def bulk_mark_superseded(
+    db: AsyncSession,
+    *,
+    study_ids: list[str],
+) -> list[str]:
+    """Conditional UPDATE for the chain-rollup loser supersession path.
+
+    Transitions ``pending → superseded`` for proposals whose ``study_id``
+    is in ``study_ids``. Idempotent. Silently skips rows whose status is not ``pending`` —
+    that includes already-superseded rows on a re-run, ``pr_opened`` /
+    ``pr_merged`` rows (operator already shipped), and ``rejected`` rows
+    (operator already rejected — D-6 / Q3 precedence). Returns the IDs
+    actually transitioned, or ``[]`` if no rows matched. Caller commits.
+    """
+    if not study_ids:
+        return []
+    stmt = (
+        update(Proposal)
+        .where(Proposal.study_id.in_(study_ids), Proposal.status == "pending")
+        .values(status="superseded")
+        .returning(Proposal.id)
+    )
+    result = await db.execute(stmt)
+    transitioned: list[str] = [row[0] for row in result]
+    if transitioned:
+        await db.flush()
+    return transitioned
+
+
+async def reinstate_from_superseded(
+    db: AsyncSession,
+    *,
+    proposal_id: str,
+) -> Proposal:
+    """Transition ``superseded → pending`` for the operator-initiated reinstate flow.
+
+    Mirrors the :func:`reject_proposal` read-check-mutate precedent
+    (spec D-17) — the conditional-UPDATE pattern would collapse 404
+    (unknown id) and 409 (wrong status) into one zero-row signal and
+    the API endpoint could not drive its deterministic 404-vs-409
+    contract.
+
+    Raises :class:`LookupError` if the proposal id does not exist
+    (API translates to HTTP 404 ``PROPOSAL_NOT_FOUND``). Raises
+    :class:`InvalidStateTransition` if the row is not in ``superseded``
+    status (API translates to HTTP 409 ``INVALID_STATE_TRANSITION`` —
+    D-16 reuses the existing reject endpoint's code).
+    Caller commits.
+    """
+    row = await get_proposal(db, proposal_id)
+    if row is None:
+        raise LookupError(f"proposal {proposal_id!r} not found")
+    if row.status != "superseded":
+        raise InvalidStateTransition(proposal_id, row.status)
+    row.status = "pending"
+    await db.flush()
+    return row

backend/app/db/repo/study.py

@@ -338,7 +338,14 @@ async def get_chain_for_study(
     proposal_rows = (
         await db.execute(
             select(Proposal.id, Proposal.study_id)
-            .where(Proposal.study_id.in_(link_ids), Proposal.status != "rejected")
+            # Phase 3 FR-4: filter widened from `!= "rejected"` to also
+            # exclude `superseded` so the chain panel's per-link proposal
+            # resolution honors the rollup. Cascades automatically to
+            # `list_recent_completed_chains` (which reuses this function).
+            .where(
+                Proposal.study_id.in_(link_ids),
+                Proposal.status.notin_(("rejected", "superseded")),
+            )
             .order_by(
                 Proposal.study_id,
                 Proposal.created_at.desc(),

backend/app/services/chain_rollup.py

@@ -0,0 +1,91 @@
+# SPDX-FileCopyrightText: 2026 soundminds.ai
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""Chain-rollup service — supersede non-winning chain links' proposals.
+
+feat_overnight_final_solution_phase3 Story 2.1 (FR-2).
+
+Walks the chain anchored at a given study, identifies the winner via
+:func:`select_best_link` (Phase 1 infra), and delegates the loser
+supersession to :func:`repo.bulk_mark_superseded`. Returns the
+``(superseded_count, superseded_ids)`` tuple so the caller can emit the
+``chain_proposals_superseded`` structlog event AFTER its commit succeeds
+(spec D-19). Does NOT commit; caller commits per the service-layer
+convention.
+
+The service is chain-scoped, not proposal-scoped — landing under
+``services/`` rather than appending to ``agent_proposals_dispatch.py``
+keeps it usable by both the autopilot path (``_stop`` in
+``backend/workers/orchestrator.py``) and any future chat-agent surface
+(spec D-4 / Q5 locked).
+"""
+
+from __future__ import annotations
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.db import repo
+from backend.app.db.repo import ChainTraversalResult
+from backend.app.domain.study.chain_summary import (
+    derive_chain_stop_reason,
+    select_best_link,
+)
+
+
+async def mark_non_winning_chain_proposals_superseded(
+    db: AsyncSession,
+    *,
+    study_id: str,
+    traversal: ChainTraversalResult | None = None,
+) -> tuple[int, list[str]]:
+    """Supersede ``pending`` proposals of all chain links other than the winner.
+
+    Walks the chain anchored at the study root of ``study_id`` and, when
+    the chain has terminated with at least 2 completed links and a clear
+    best link, conditional-UPDATEs all sibling losers' ``pending``
+    proposals to ``superseded`` (via :func:`repo.bulk_mark_superseded`).
+
+    Returns ``(count, ids)`` — the count and IDs actually transitioned —
+    so the caller can emit the post-commit ``chain_proposals_superseded``
+    structlog event with the full IDs payload per spec D-19.
+
+    ``traversal`` is an optional pre-fetched
+    :class:`~backend.app.db.repo.ChainTraversalResult`. When the caller
+    has already walked the chain (e.g. ``_stop`` reads it to capture the
+    anchor + winner for its log payload), passing it here avoids a
+    duplicate ``get_chain_for_study`` round-trip. When ``None`` the
+    function fetches it itself.
+
+    Idempotent. Re-running on the same chain after a successful first
+    call returns ``(0, [])`` (the losers are now ``superseded`` and the
+    repo helper's ``WHERE status='pending'`` clause excludes them).
+
+    Early-returns ``(0, [])`` when:
+        a. the chain is missing (study not found).
+        b. The chain has fewer than 2 links (single-link chain has no
+           siblings to supersede).
+        c. The derived ``stop_reason == "in_flight"`` (chain still
+           running; rollup deferred until termination).
+        d. :func:`select_best_link` returns ``None`` (no completed link →
+           no winner → nothing to supersede against).
+
+    Does NOT commit. Caller commits as part of its own transaction
+    boundary (e.g., ``_stop`` commits the link's ``pending`` proposal
+    insert and the rollup in the same transaction).
+    """
+    if traversal is None:
+        traversal = await repo.get_chain_for_study(db, study_id)
+    if traversal is None:
+        return (0, [])
+    if len(traversal.links) < 2:
+        return (0, [])
+    stop_reason = derive_chain_stop_reason(traversal.links, traversal.anchor_trials)
+    if stop_reason == "in_flight":
+        return (0, [])
+    best_link_id = select_best_link(traversal.links)
+    if best_link_id is None:
+        return (0, [])
+    loser_ids = [link.id for link in traversal.links if link.id != best_link_id]
+    transitioned = await repo.bulk_mark_superseded(db, study_ids=loser_ids)
+    return (len(transitioned), transitioned)

backend/tests/contract/test_openapi_surface.py

@@ -105,6 +105,7 @@
     ("get", "/api/v1/proposals/{proposal_id}", "200"),
     ("post", "/api/v1/proposals/{proposal_id}/reject", "200"),
     ("post", "/api/v1/proposals/{proposal_id}/open_pr", "202"),
+    ("post", "/api/v1/proposals/{proposal_id}/reinstate", "200"),
     # ----- /api/v1/conversations (feat_chat_agent) -----
     ("post", "/api/v1/conversations", "201"),
     ("get", "/api/v1/conversations", "200"),