STATE-WP-0101-T02/T03: decision classes, evidence refusals, write-back route

T02: additive migration d5e6f7a8b9c0 (decision_class, memo_id,
approval_id, resolution_source, resolution_evidence, evidence_refs,
CHECKs, partial unique memo_id); typed evidence_refs; 409
evidence_required for text resolve, resolving PATCH fields and resolved
create on C1/C2; raise-only class; C3 refused as reserved.

T03: POST /decisions/{id}/resolve-evidence (link/resolve), projected
token verified by TokenReview with audience
state-hub-decision-writeback and an exact subject allowlist; outcome
record v1 validation, 8 KiB bound, secret scan; idempotent on
(decision_id, memo_id). Off by default (DECISION_WRITEBACK_ENABLED).
Tests use a fake reviewer; full suite 1286 passed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 5502@bnt-lap001
Assistant-Session: 68b3b02f-5d0f-4e71-8421-796f62e50196

api/config.py

@@ -42,6 +42,16 @@ class Settings(BaseSettings):
     repository_rename_preflight_secret: str | None = None
     repository_rename_preflight_ttl_seconds: int = 900
     repository_rename_forge_timeout_seconds: float = 10.0
+    # STATE-WP-0101-T03: authenticated decision write-back route
+    # (POST /decisions/{id}/resolve-evidence). Off by default; enabling it needs
+    # the API ServiceAccount bound to system:auth-delegator (founder-run, cluster
+    # RBAC) so TokenReview works. Subjects: comma-separated exact allowlist.
+    decision_writeback_enabled: bool = False
+    decision_writeback_audience: str = "state-hub-decision-writeback"
+    decision_writeback_subjects: str = "system:serviceaccount:state-hub:state-hub"
+    kubernetes_api_url: str = "https://kubernetes.default.svc"
+    kubernetes_serviceaccount_dir: str = "/var/run/secrets/kubernetes.io/serviceaccount"
+    token_review_timeout_seconds: float = 5.0
 
 
 settings = Settings()

api/models/decision.py

@@ -2,8 +2,8 @@ import enum
 import uuid
 from datetime import datetime
 
-from sqlalchemy import CheckConstraint, DateTime, Enum, ForeignKey, String, Text
-from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy import CheckConstraint, DateTime, Enum, ForeignKey, Index, String, Text, text
+from sqlalchemy.dialects.postgresql import JSONB, UUID
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from api.models.base import Base, TimestampMixin, new_uuid
@@ -21,6 +21,41 @@ class DecisionStatus(str, enum.Enum):
     superseded = "superseded"
 
 
+class DecisionClass(str, enum.Enum):
+    """Gate-house decision classes (GH-DEC-2026-026). NULL on a row means legacy (as C0)."""
+
+    C0 = "C0"
+    C1 = "C1"
+    C2 = "C2"
+    C3 = "C3"  # reserved by GH-DEC-2026-026; stored values allowed, the API refuses it
+
+
+class ResolutionSource(str, enum.Enum):
+    text = "text"
+    informed_decision = "informed_decision"
+
+
+# Classes whose resolution needs informed-decision evidence (STATE-WP-0101, design section 3.1).
+# C3 is reserved and refused by the API; the database treats it fail-closed like C2.
+EVIDENCE_CLASSES = frozenset({DecisionClass.C1, DecisionClass.C2, DecisionClass.C3})
+APPROVAL_CLASSES = frozenset({DecisionClass.C2, DecisionClass.C3})
+CLASS_RANK = {None: 0, DecisionClass.C0: 0, DecisionClass.C1: 1, DecisionClass.C2: 2, DecisionClass.C3: 3}
+
+# Shared with the Alembic migration (STATE-WP-0101-T02); keep the two in step.
+CK_DECISION_CLASS = "decision_class IS NULL OR decision_class IN ('C0','C1','C2','C3')"
+CK_RESOLUTION_SOURCE = (
+    "resolution_source IS NULL OR resolution_source IN ('text','informed_decision')"
+)
+CK_CLASS_EVIDENCE = (
+    "NOT (status = 'resolved' AND decision_class IN ('C1','C2','C3')) OR "
+    "(resolution_source = 'informed_decision' AND memo_id IS NOT NULL "
+    "AND resolution_evidence IS NOT NULL)"
+)
+CK_CLASS_APPROVAL = (
+    "NOT (status = 'resolved' AND decision_class IN ('C2','C3')) OR approval_id IS NOT NULL"
+)
+
+
 class Decision(Base, TimestampMixin):
     __tablename__ = "decisions"
     __table_args__ = (
@@ -28,6 +63,16 @@ class Decision(Base, TimestampMixin):
             "topic_id IS NOT NULL OR workplan_id IS NOT NULL",
             name="ck_decisions_topic_or_workplan",
         ),
+        CheckConstraint(CK_DECISION_CLASS, name="ck_decisions_decision_class"),
+        CheckConstraint(CK_RESOLUTION_SOURCE, name="ck_decisions_resolution_source"),
+        CheckConstraint(CK_CLASS_EVIDENCE, name="ck_decisions_class_evidence"),
+        CheckConstraint(CK_CLASS_APPROVAL, name="ck_decisions_class_approval"),
+        Index(
+            "uq_decisions_memo_id",
+            "memo_id",
+            unique=True,
+            postgresql_where=text("memo_id IS NOT NULL"),
+        ),
     )
 
     id: Mapped[uuid.UUID] = mapped_column(
@@ -58,6 +103,15 @@ class Decision(Base, TimestampMixin):
     superseded_by: Mapped[uuid.UUID | None] = mapped_column(
         UUID(as_uuid=True), ForeignKey("decisions.id", ondelete="SET NULL"), nullable=True
     )
+    # STATE-WP-0101: decision class and informed-decision evidence links.
+    decision_class: Mapped[str | None] = mapped_column(String(2), nullable=True, index=True)
+    memo_id: Mapped[str | None] = mapped_column(Text, nullable=True)
+    approval_id: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), nullable=True)
+    resolution_source: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    resolution_evidence: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
+    evidence_refs: Mapped[list] = mapped_column(
+        JSONB, nullable=False, default=list, server_default=text("'[]'::jsonb")
+    )
 
     topic: Mapped["Topic | None"] = relationship("Topic", back_populates="decisions")  # noqa: F821
     workplan: Mapped["Workplan | None"] = relationship("Workplan", back_populates="decisions")  # noqa: F821

api/routers/decisions.py

@@ -7,15 +7,30 @@ from pathlib import Path
 from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.database import get_session
 from api.events import EventEnvelope, publish_event
-from api.models.decision import Decision, DecisionStatus, DecisionType
+from api.models.decision import (
+    APPROVAL_CLASSES,
+    CLASS_RANK,
+    EVIDENCE_CLASSES,
+    Decision,
+    DecisionClass,
+    DecisionStatus,
+    DecisionType,
+    ResolutionSource,
+)
 from api.models.progress_event import ProgressEvent
 from api.models.review_contract import ReviewContract
 from api.schemas.decision import DecisionCreate, DecisionRead, DecisionResolve, DecisionUpdate
 from api.services.legacy_compat import meter_legacy_body_from_model, meter_legacy_query_param
+from api.services.decision_writeback import (
+    DecisionEvidenceWrite,
+    require_writeback_caller,
+    stored_outcome,
+)
 from api.services.review_contracts import aggregate
 
 logger = logging.getLogger(__name__)
@@ -40,6 +55,82 @@ def _needs_escalation(body: DecisionCreate) -> str | None:
     return None
 
 
+# STATE-WP-0101: fields that resolve or describe a resolution. On C1/C2 they are
+# written only by POST /decisions/{id}/resolve-evidence.
+_RESOLVING_PATCH_FIELDS = ("decided_by", "decided_at", "rationale")
+
+
+def _class_of(decision: Decision) -> DecisionClass | None:
+    return DecisionClass(decision.decision_class) if decision.decision_class else None
+
+
+def _refuse_reserved(value: DecisionClass | None) -> None:
+    if value == DecisionClass.C3:
+        raise HTTPException(
+            status_code=422,
+            detail={
+                "error": "decision_class_reserved",
+                "message": "C3 is reserved and not usable (gate-house GH-DEC-2026-026)",
+            },
+        )
+
+
+def _evidence_required(decision_class: DecisionClass, action: str) -> HTTPException:
+    return HTTPException(
+        status_code=409,
+        detail={
+            "error": "evidence_required",
+            "message": (
+                f"{decision_class.value} decisions are resolved only by informed-decision "
+                f"evidence; {action} is refused"
+            ),
+            "decision_class": decision_class.value,
+            "resolve_with": "POST /decisions/{id}/resolve-evidence",
+        },
+    )
+
+
+async def _require_review_contracts(session: AsyncSession, decision: Decision) -> None:
+    review_rows = await session.execute(
+        select(ReviewContract).where(
+            ReviewContract.decision_id == decision.id,
+            ReviewContract.active.is_(True),
+            ReviewContract.required_for_decision.is_(True),
+        )
+    )
+    for contract in review_rows.scalars():
+        review_state = await aggregate(session, contract)
+        if not review_state.satisfied:
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "message": "required multi-owner review is not satisfied",
+                    "contract_key": contract.contract_key,
+                    "contract_digest": contract.contract_digest,
+                },
+            )
+
+
+def _publish_resolved(decision: Decision, decided_by: str, rationale: str | None) -> None:
+    subject = "org.statehub.decision.resolved"
+    envelope = EventEnvelope.new(
+        subject,
+        attributes={
+            "decision_id": str(decision.id),
+            "title": decision.title,
+            "topic_id": str(decision.topic_id) if decision.topic_id else None,
+            "workstream_id": str(decision.workplan_id) if decision.workplan_id else None,
+            "decided_by": decided_by,
+            "rationale_snippet": (rationale or "")[:240],
+            "decision_class": decision.decision_class,
+            "memo_id": decision.memo_id,
+            "approval_id": str(decision.approval_id) if decision.approval_id else None,
+            "resolution_source": decision.resolution_source,
+        },
+    )
+    asyncio.create_task(publish_event(subject, envelope))
+
+
 @router.get("/", response_model=list[DecisionRead])
 async def list_decisions(
     request: Request,
@@ -49,6 +140,7 @@ async def list_decisions(
     workstream_id: uuid.UUID | None = None,
     status: DecisionStatus | None = None,
     decision_type: DecisionType | None = None,
+    decision_class: DecisionClass | None = None,
     session: AsyncSession = Depends(get_session),
 ) -> list[Decision]:
     if workstream_id is not None and workplan_id is None:
@@ -70,6 +162,8 @@ async def list_decisions(
         q = q.where(Decision.status == status)
     if decision_type:
         q = q.where(Decision.decision_type == decision_type)
+    if decision_class:
+        q = q.where(Decision.decision_class == decision_class.value)
     q = q.order_by(Decision.created_at)
     result = await session.execute(q)
     return list(result.scalars().all())
@@ -91,7 +185,11 @@ async def create_decision(
         route="/decisions/",
         replacement_ref="POST /decisions/ with workplan_id",
     )
+    _refuse_reserved(body.decision_class)
+    if body.decision_class in EVIDENCE_CLASSES and body.status == DecisionStatus.resolved:
+        raise _evidence_required(body.decision_class, "creating it already resolved")
     data = body.model_dump()
+    data["decision_class"] = body.decision_class.value if body.decision_class else None
     note = _needs_escalation(body)
     if note:
         data["escalation_note"] = note
@@ -123,7 +221,52 @@ async def update_decision(
     decision = await session.get(Decision, decision_id)
     if decision is None:
         raise HTTPException(status_code=404, detail="Decision not found")
-    for field, value in body.model_dump(exclude_unset=True).items():
+    fields = body.model_dump(exclude_unset=True)
+    current = _class_of(decision)
+    effective = current
+    if "decision_class" in fields:
+        new = body.decision_class
+        _refuse_reserved(new)
+        if CLASS_RANK[new] < CLASS_RANK[current] or (new is None and current is not None):
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "error": "decision_class_lowering",
+                    "message": "a decision class may be raised, never lowered",
+                    "decision_class": current.value if current else None,
+                },
+            )
+        if new != current and decision.status in (
+            DecisionStatus.resolved, DecisionStatus.superseded
+        ):
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "error": "decision_class_frozen",
+                    "message": "the class of a resolved or superseded decision cannot change",
+                },
+            )
+        effective = new
+        fields["decision_class"] = new.value if new else None
+    if effective in EVIDENCE_CLASSES:
+        if fields.get("status") == DecisionStatus.resolved:
+            raise _evidence_required(effective, "setting status=resolved by PATCH")
+        touched = [f for f in _RESOLVING_PATCH_FIELDS if f in fields]
+        if touched:
+            raise _evidence_required(effective, f"setting {', '.join(touched)} by PATCH")
+        if (
+            decision.status == DecisionStatus.resolved
+            and "status" in fields
+            and fields["status"] not in (DecisionStatus.resolved, DecisionStatus.superseded)
+        ):
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "error": "evidence_bound",
+                    "message": "an evidence-resolved decision is not reopened by PATCH",
+                },
+            )
+    for field, value in fields.items():
         setattr(decision, field, value)
     await session.commit()
     await session.refresh(decision)
@@ -155,31 +298,18 @@ async def resolve_decision_action(
         raise HTTPException(status_code=404, detail="Decision not found")
     if decision.status == DecisionStatus.resolved:
         raise HTTPException(status_code=409, detail="Decision already resolved")
+    decision_class = _class_of(decision)
+    if decision_class in EVIDENCE_CLASSES:
+        raise _evidence_required(decision_class, "text resolution")
 
-    review_rows = await session.execute(
-        select(ReviewContract).where(
-            ReviewContract.decision_id == decision.id,
-            ReviewContract.active.is_(True),
-            ReviewContract.required_for_decision.is_(True),
-        )
-    )
-    for contract in review_rows.scalars():
-        review_state = await aggregate(session, contract)
-        if not review_state.satisfied:
-            raise HTTPException(
-                status_code=409,
-                detail={
-                    "message": "required multi-owner review is not satisfied",
-                    "contract_key": contract.contract_key,
-                    "contract_digest": contract.contract_digest,
-                },
-            )
+    await _require_review_contracts(session, decision)
 
     decision.status = DecisionStatus.resolved
     decision.decision_type = DecisionType.made
     decision.rationale = body.rationale
     decision.decided_by = body.decided_by
     decision.decided_at = datetime.now(tz=timezone.utc)
+    decision.resolution_source = ResolutionSource.text.value
     await session.commit()
     await session.refresh(decision)
 
@@ -198,23 +328,152 @@ async def resolve_decision_action(
     if body.write_log:
         await _write_project_log(decision, body.rationale, body.decided_by, session)
 
-    subject = "org.statehub.decision.resolved"
-    envelope = EventEnvelope.new(
-        subject,
-        attributes={
-            "decision_id": str(decision.id),
-            "title": decision.title,
-            "topic_id": str(decision.topic_id) if decision.topic_id else None,
-            "workstream_id": str(decision.workplan_id) if decision.workplan_id else None,
-            "decided_by": body.decided_by,
-            "rationale_snippet": (body.rationale or "")[:240],
-        },
-    )
-    asyncio.create_task(publish_event(subject, envelope))
+    _publish_resolved(decision, body.decided_by, body.rationale)
 
     return decision
 
 
+def _conflict(error: str, message: str, **extra) -> HTTPException:
+    return HTTPException(status_code=409, detail={"error": error, "message": message, **extra})
+
+
+@router.post("/{decision_id}/resolve-evidence", response_model=DecisionRead)
+async def resolve_decision_with_evidence(
+    decision_id: uuid.UUID,
+    body: DecisionEvidenceWrite,
+    caller: str = Depends(require_writeback_caller),
+    session: AsyncSession = Depends(get_session),
+) -> Decision:
+    """Link a memo to, or resolve, a C1/C2 decision with informed-decision evidence.
+
+    STATE-WP-0101-T03, design sections 5.1-5.3. Only the verified write-back
+    subject reaches this body. Idempotent on (decision_id, memo_id).
+    """
+    decision = await session.get(Decision, decision_id)
+    if decision is None:
+        raise HTTPException(status_code=404, detail="Decision not found")
+    decision_class = _class_of(decision)
+    if decision_class not in (DecisionClass.C1, DecisionClass.C2):
+        raise _conflict(
+            "evidence_not_applicable",
+            "evidence write-back applies to C1 and C2 decisions only",
+            decision_class=decision_class.value if decision_class else None,
+        )
+    if decision.status == DecisionStatus.superseded:
+        raise _conflict("decision_superseded", "the decision was superseded")
+    if decision.memo_id is not None and decision.memo_id != body.memo_id:
+        raise _conflict(
+            "memo_conflict", "the decision is linked to another memo", memo_id=decision.memo_id
+        )
+    if decision.memo_id is None:
+        other = await session.execute(
+            select(Decision.id).where(Decision.memo_id == body.memo_id, Decision.id != decision.id)
+        )
+        if other.scalar_one_or_none() is not None:
+            raise _conflict("memo_already_linked", "the memo answers another decision")
+
+    outcome = body.outcome
+    if outcome is not None:
+        if outcome.hub_decision_id != decision.id:
+            raise HTTPException(status_code=422, detail={"error": "hub_decision_mismatch"})
+        if outcome.memo_id != body.memo_id:
+            raise HTTPException(status_code=422, detail={"error": "memo_mismatch"})
+    approvals = {
+        a for a in (
+            body.approval_id,
+            outcome.approval_id if outcome else None,
+            decision.approval_id,
+        ) if a is not None
+    }
+    if len(approvals) > 1:
+        raise _conflict(
+            "approval_conflict",
+            "approval_id differs from the one linked",
+            approval_id=str(decision.approval_id) if decision.approval_id else None,
+        )
+    approval_id = next(iter(approvals), None)
+
+    if decision.status == DecisionStatus.resolved:
+        # Same memo (checked above): a repeated link or resolve is a no-op.
+        return decision
+
+    if body.mode == "link":
+        changed = decision.memo_id is None or (
+            approval_id is not None and decision.approval_id is None
+        )
+        decision.memo_id = body.memo_id
+        decision.approval_id = approval_id
+        if changed:
+            try:
+                await session.commit()
+            except IntegrityError:
+                await session.rollback()
+                raise _conflict("memo_already_linked", "the memo answers another decision")
+            await session.refresh(decision)
+        return decision
+
+    assert outcome is not None  # guaranteed by DecisionEvidenceWrite
+    if decision_class in APPROVAL_CLASSES and (
+        approval_id is None or not outcome.approval_entry_id
+    ):
+        raise HTTPException(
+            status_code=422,
+            detail={
+                "error": "approval_required",
+                "message": "C2 resolution needs approval_id and approval_entry_id",
+            },
+        )
+    record = stored_outcome(outcome)
+    await _require_review_contracts(session, decision)
+
+    decided_by = (outcome.display_name or f"{outcome.subject.iss}#{outcome.subject.sub}")[:100]
+    verdict = "accepted" if outcome.verb == "accept" else "declined"
+    rationale = (
+        f"{verdict} in informed-decision, memo {body.memo_id} revision {outcome.memo_revision}"
+        + (f", option {outcome.option}" if outcome.option else "")
+        + (f"\n\n{outcome.note}" if outcome.note else "")
+    )
+    decision.status = DecisionStatus.resolved
+    decision.decision_type = DecisionType.made
+    decision.memo_id = body.memo_id
+    decision.approval_id = approval_id
+    decision.rationale = rationale
+    decision.decided_by = decided_by
+    decision.decided_at = outcome.decided_at
+    decision.resolution_source = ResolutionSource.informed_decision.value
+    decision.resolution_evidence = record
+    session.add(
+        ProgressEvent(
+            topic_id=decision.topic_id,
+            workplan_id=decision.workplan_id,
+            decision_id=decision.id,
+            event_type="decision_resolved",
+            summary=f"Decision resolved ({verdict}, informed-decision): {decision.title}",
+            author=decided_by,
+            detail={
+                "resolution_source": ResolutionSource.informed_decision.value,
+                "decision_class": decision_class.value,
+                "memo_id": body.memo_id,
+                "memo_revision": outcome.memo_revision,
+                "verb": outcome.verb,
+                "subject": record["subject"],
+                "view_hash": outcome.view_hash,
+                "presentation_id": outcome.presentation_id,
+                "approval_id": str(approval_id) if approval_id else None,
+                "attested_via": caller,
+            },
+        )
+    )
+    try:
+        await session.commit()
+    except IntegrityError:
+        await session.rollback()
+        raise _conflict("memo_already_linked", "the memo answers another decision")
+    await session.refresh(decision)
+    _publish_resolved(decision, decided_by, rationale)
+    return decision
+
+
 async def _write_project_log(
     decision: Decision, rationale: str, decided_by: str, session: AsyncSession
 ) -> None:

api/schemas/decision.py

@@ -1,13 +1,42 @@
 import uuid
 from datetime import datetime
+from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, model_validator
+from pydantic import BaseModel, ConfigDict, field_validator, model_validator
 
-from api.models.decision import DecisionStatus, DecisionType
+from api.models.decision import DecisionClass, DecisionStatus, DecisionType
 from api.schemas.compat import LegacyWorkstreamIdBodyMixin, OptionalWorkplanIdCompatMixin
 from pydantic import AliasChoices, Field
 
 
+EvidenceRefType = Literal[
+    "presentation", "audit_record", "ccr", "workplan_task", "adr", "commit", "document"
+]
+
+# Written only through POST /decisions/{id}/resolve-evidence (STATE-WP-0101).
+EVIDENCE_LINK_FIELDS = ("memo_id", "approval_id", "resolution_source", "resolution_evidence")
+
+
+class EvidenceRef(BaseModel):
+    """Descriptive reference (design section 3.2). Resolves nothing; closed type vocabulary."""
+
+    model_config = ConfigDict(extra="forbid")
+    type: EvidenceRefType
+    ref: str = Field(min_length=1, max_length=512)
+    digest: str | None = Field(default=None, max_length=128)
+
+
+def _refuse_link_fields(data: Any) -> Any:
+    if isinstance(data, dict):
+        present = sorted(k for k in EVIDENCE_LINK_FIELDS if k in data)
+        if present:
+            raise ValueError(
+                f"{', '.join(present)} can only be set through "
+                "POST /decisions/{id}/resolve-evidence"
+            )
+    return data
+
+
 class DecisionCreate(LegacyWorkstreamIdBodyMixin, BaseModel):
     id: uuid.UUID | None = None
     topic_id: uuid.UUID | None = None
@@ -24,6 +53,13 @@ class DecisionCreate(LegacyWorkstreamIdBodyMixin, BaseModel):
     decided_at: datetime | None = None
     deadline: datetime | None = None
     escalation_note: str | None = None
+    decision_class: DecisionClass | None = None
+    evidence_refs: list[EvidenceRef] = Field(default_factory=list, max_length=64)
+
+    @model_validator(mode="before")
+    @classmethod
+    def no_link_fields(cls, data: Any) -> Any:
+        return _refuse_link_fields(data)
 
     @model_validator(mode="after")
     def topic_or_workplan_required(self) -> "DecisionCreate":
@@ -49,6 +85,13 @@ class DecisionUpdate(BaseModel):
     deadline: datetime | None = None
     escalation_note: str | None = None
     superseded_by: uuid.UUID | None = None
+    decision_class: DecisionClass | None = None
+    evidence_refs: list[EvidenceRef] | None = Field(default=None, max_length=64)
+
+    @model_validator(mode="before")
+    @classmethod
+    def no_link_fields(cls, data: Any) -> Any:
+        return _refuse_link_fields(data)
 
 
 class DecisionRead(OptionalWorkplanIdCompatMixin, BaseModel):
@@ -65,5 +108,16 @@ class DecisionRead(OptionalWorkplanIdCompatMixin, BaseModel):
     deadline: datetime | None = None
     escalation_note: str | None = None
     superseded_by: uuid.UUID | None = None
+    decision_class: DecisionClass | None = None
+    memo_id: str | None = None
+    approval_id: uuid.UUID | None = None
+    resolution_source: str | None = None
+    resolution_evidence: dict | None = None
+    evidence_refs: list[dict] = Field(default_factory=list)
     created_at: datetime
     updated_at: datetime
+
+    @field_validator("evidence_refs", mode="before")
+    @classmethod
+    def _refs_default(cls, value: Any) -> Any:
+        return [] if value is None else value

api/services/decision_writeback.py

@@ -0,0 +1,236 @@
+"""Authenticated decision write-back (STATE-WP-0101-T03).
+
+Design: docs/decision-services-alignment-answer.md sections 5.2 and 5.3; founder
+decisions 1 and 2 of 2026-11-01 (section 8.1). The dispatch worker pulls decided
+outcomes from informed-decision and posts them here. Exactly one caller is
+admitted: a Kubernetes projected ServiceAccount token with the dedicated
+audience, verified by TokenReview, whose subject is on an exact allowlist.
+
+The hub records the human subject as attested by informed-decision through the
+authenticated worker; it does not authenticate the human itself.
+"""
+from __future__ import annotations
+
+import json
+import uuid
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Literal, Protocol
+
+import httpx
+from fastapi import Depends, HTTPException, Request
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from api.config import settings
+from api.services.informed_decision_contract import ContractError, scan_secrets
+
+OUTCOME_KIND = "informed-decision.decision-outcome"
+MAX_OUTCOME_BYTES = 8192
+MAX_NOTE_BYTES = 2048
+
+_MEMO_ID_PATTERN = r"^[a-z0-9][a-z0-9-]{2,62}$"
+_VIEW_HASH_PATTERN = r"^sha256:[0-9a-f]{64}$"
+
+
+# ---------------------------------------------------------------------------
+# Outcome record, informed-decision.decision-outcome v1 (design section 5.2)
+# ---------------------------------------------------------------------------
+
+
+class OutcomeSubject(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    iss: str = Field(min_length=1, max_length=512)
+    sub: str = Field(min_length=1, max_length=255)
+
+
+class OutcomeAssurance(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    acr: str | None = Field(default=None, max_length=128)
+    aal: str | None = Field(default=None, max_length=32)
+    auth_time: datetime
+
+    @model_validator(mode="after")
+    def _level_present(self) -> "OutcomeAssurance":
+        if not self.acr and not self.aal:
+            raise ValueError("assurance needs acr or aal")
+        if self.auth_time.tzinfo is None:
+            raise ValueError("auth_time must carry a timezone")
+        return self
+
+
+class DecisionOutcome(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    kind: Literal["informed-decision.decision-outcome"]
+    version: Literal[1]
+    hub_decision_id: uuid.UUID
+    memo_id: str = Field(pattern=_MEMO_ID_PATTERN)
+    memo_revision: int = Field(ge=0)
+    verb: Literal["accept", "decline"]
+    option: str | None = Field(default=None, max_length=64)
+    subject: OutcomeSubject
+    assurance: OutcomeAssurance
+    decided_at: datetime
+    view_hash: str = Field(pattern=_VIEW_HASH_PATTERN)
+    presentation_id: str = Field(min_length=1, max_length=128)
+    approval_id: uuid.UUID | None = None
+    approval_entry_id: str | None = Field(default=None, max_length=128)
+    audit_ref: str | None = Field(default=None, max_length=256)
+    display_name: str | None = Field(default=None, max_length=100)
+    note: str | None = None
+
+    @field_validator("decided_at")
+    @classmethod
+    def _utc(cls, value: datetime) -> datetime:
+        if value.tzinfo is None:
+            raise ValueError("decided_at must carry a timezone")
+        return value
+
+    @field_validator("note")
+    @classmethod
+    def _bounded_note(cls, value: str | None) -> str | None:
+        if value is not None and len(value.encode("utf-8")) > MAX_NOTE_BYTES:
+            raise ValueError(f"note exceeds {MAX_NOTE_BYTES} bytes")
+        return value
+
+
+class DecisionEvidenceWrite(BaseModel):
+    """Body of POST /decisions/{id}/resolve-evidence."""
+
+    model_config = ConfigDict(extra="forbid")
+    mode: Literal["link", "resolve"]
+    memo_id: str = Field(pattern=_MEMO_ID_PATTERN)
+    approval_id: uuid.UUID | None = None
+    outcome: DecisionOutcome | None = None
+
+    @model_validator(mode="after")
+    def _mode_shape(self) -> "DecisionEvidenceWrite":
+        if self.mode == "resolve" and self.outcome is None:
+            raise ValueError("resolve mode requires an outcome record")
+        if self.mode == "link" and self.outcome is not None:
+            raise ValueError("link mode carries no outcome record")
+        return self
+
+
+def stored_outcome(outcome: DecisionOutcome) -> dict:
+    """The record as received (unset optional fields omitted), bounded and secret-scanned."""
+    record = outcome.model_dump(mode="json", exclude_unset=True)
+    encoded = json.dumps(record, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+    if len(encoded.encode("utf-8")) > MAX_OUTCOME_BYTES:
+        raise HTTPException(
+            status_code=422,
+            detail={"error": "outcome_too_large", "limit_bytes": MAX_OUTCOME_BYTES},
+        )
+    try:
+        scan_secrets(record)
+    except ContractError:
+        # The matched text is never echoed.
+        raise HTTPException(status_code=422, detail={"error": "secret_marker"}) from None
+    return record
+
+
+# ---------------------------------------------------------------------------
+# Caller verification (design section 5.3)
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class TokenReviewResult:
+    authenticated: bool
+    username: str | None = None
+    audiences: tuple[str, ...] = ()
+
+
+class TokenReviewer(Protocol):
+    async def review(self, token: str, audience: str) -> TokenReviewResult: ...
+
+
+class TokenReviewUnavailable(RuntimeError):
+    pass
+
+
+class KubernetesTokenReviewer:
+    """TokenReview against the cluster API with the API pod's own ServiceAccount.
+
+    Needs the API ServiceAccount bound to ``system:auth-delegator`` (founder-run
+    cluster RBAC). Any transport or API failure raises TokenReviewUnavailable;
+    the route then refuses (fail closed).
+    """
+
+    def __init__(
+        self,
+        api_url: str | None = None,
+        serviceaccount_dir: str | None = None,
+        timeout: float | None = None,
+    ) -> None:
+        self.api_url = (api_url or settings.kubernetes_api_url).rstrip("/")
+        self.sa_dir = Path(serviceaccount_dir or settings.kubernetes_serviceaccount_dir)
+        self.timeout = timeout if timeout is not None else settings.token_review_timeout_seconds
+
+    async def review(self, token: str, audience: str) -> TokenReviewResult:
+        try:
+            own_token = (self.sa_dir / "token").read_text().strip()
+            ca = str(self.sa_dir / "ca.crt")
+            async with httpx.AsyncClient(verify=ca, timeout=self.timeout) as client:
+                response = await client.post(
+                    f"{self.api_url}/apis/authentication.k8s.io/v1/tokenreviews",
+                    headers={"Authorization": f"Bearer {own_token}"},
+                    json={
+                        "apiVersion": "authentication.k8s.io/v1",
+                        "kind": "TokenReview",
+                        "spec": {"token": token, "audiences": [audience]},
+                    },
+                )
+        except (OSError, httpx.HTTPError) as exc:
+            raise TokenReviewUnavailable(type(exc).__name__) from None
+        if response.status_code != 201 and response.status_code != 200:
+            raise TokenReviewUnavailable(f"tokenreview_http_{response.status_code}")
+        status = (response.json() or {}).get("status") or {}
+        return TokenReviewResult(
+            authenticated=bool(status.get("authenticated")),
+            username=(status.get("user") or {}).get("username"),
+            audiences=tuple(status.get("audiences") or ()),
+        )
+
+
+def get_token_reviewer() -> TokenReviewer:
+    """FastAPI dependency; tests override it with a fake reviewer."""
+    return KubernetesTokenReviewer()
+
+
+def allowed_subjects() -> frozenset[str]:
+    return frozenset(
+        s.strip() for s in settings.decision_writeback_subjects.split(",") if s.strip()
+    )
+
+
+async def require_writeback_caller(
+    request: Request, reviewer: TokenReviewer = Depends(get_token_reviewer)
+) -> str:
+    """Return the verified caller subject or refuse. Feature flag first, then token."""
+    if not settings.decision_writeback_enabled:
+        raise HTTPException(status_code=503, detail={"error": "decision_writeback_disabled"})
+    header = request.headers.get("authorization", "")
+    scheme, _, token = header.partition(" ")
+    if scheme.lower() != "bearer" or not token.strip():
+        raise HTTPException(
+            status_code=401,
+            detail={"error": "token_required"},
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    audience = settings.decision_writeback_audience
+    try:
+        result = await reviewer.review(token.strip(), audience)
+    except TokenReviewUnavailable:
+        raise HTTPException(
+            status_code=503, detail={"error": "token_review_unavailable"}
+        ) from None
+    if not result.authenticated or audience not in result.audiences:
+        raise HTTPException(
+            status_code=401,
+            detail={"error": "token_invalid"},
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    if result.username not in allowed_subjects():
+        raise HTTPException(status_code=403, detail={"error": "subject_not_allowed"})
+    return result.username

migrations/versions/d5e6f7a8b9c0_decision_evidence_links.py

@@ -0,0 +1,87 @@
+"""decision classes and informed-decision evidence links (STATE-WP-0101-T02)
+
+Additive, per docs/decision-services-alignment-answer.md sections 3 and 6:
+
+- nullable ``decision_class``, ``memo_id``, ``approval_id``, ``resolution_source``,
+  ``resolution_evidence`` on ``decisions``; ``evidence_refs`` jsonb default ``[]``;
+- CHECK constraints: class and source vocabularies, resolved C1/C2 needs
+  informed-decision evidence, resolved C2 needs ``approval_id`` (C3 is reserved
+  and treated fail-closed like C2);
+- partial unique index on ``memo_id``: one memo answers at most one decision.
+
+Existing rows get ``decision_class = NULL`` (legacy, behaves as C0). No backfill.
+The CHECK texts are the ones in ``api.models.decision``.
+
+Revision ID: d5e6f7a8b9c0
+Revises: c4d5e6f7a8b9
+"""
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+
+revision = "d5e6f7a8b9c0"
+down_revision = "c4d5e6f7a8b9"
+branch_labels = None
+depends_on = None
+
+CK_DECISION_CLASS = "decision_class IS NULL OR decision_class IN ('C0','C1','C2','C3')"
+CK_RESOLUTION_SOURCE = (
+    "resolution_source IS NULL OR resolution_source IN ('text','informed_decision')"
+)
+CK_CLASS_EVIDENCE = (
+    "NOT (status = 'resolved' AND decision_class IN ('C1','C2','C3')) OR "
+    "(resolution_source = 'informed_decision' AND memo_id IS NOT NULL "
+    "AND resolution_evidence IS NOT NULL)"
+)
+CK_CLASS_APPROVAL = (
+    "NOT (status = 'resolved' AND decision_class IN ('C2','C3')) OR approval_id IS NOT NULL"
+)
+
+
+def upgrade() -> None:
+    op.add_column("decisions", sa.Column("decision_class", sa.String(2), nullable=True))
+    op.add_column("decisions", sa.Column("memo_id", sa.Text(), nullable=True))
+    op.add_column("decisions", sa.Column("approval_id", UUID(as_uuid=True), nullable=True))
+    op.add_column("decisions", sa.Column("resolution_source", sa.String(32), nullable=True))
+    op.add_column("decisions", sa.Column("resolution_evidence", JSONB(), nullable=True))
+    op.add_column(
+        "decisions",
+        sa.Column(
+            "evidence_refs", JSONB(), nullable=False, server_default=sa.text("'[]'::jsonb")
+        ),
+    )
+    op.create_check_constraint("ck_decisions_decision_class", "decisions", CK_DECISION_CLASS)
+    op.create_check_constraint(
+        "ck_decisions_resolution_source", "decisions", CK_RESOLUTION_SOURCE
+    )
+    op.create_check_constraint("ck_decisions_class_evidence", "decisions", CK_CLASS_EVIDENCE)
+    op.create_check_constraint("ck_decisions_class_approval", "decisions", CK_CLASS_APPROVAL)
+    op.create_index("ix_decisions_decision_class", "decisions", ["decision_class"])
+    op.create_index(
+        "uq_decisions_memo_id",
+        "decisions",
+        ["memo_id"],
+        unique=True,
+        postgresql_where=sa.text("memo_id IS NOT NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("uq_decisions_memo_id", table_name="decisions")
+    op.drop_index("ix_decisions_decision_class", table_name="decisions")
+    for name in (
+        "ck_decisions_class_approval",
+        "ck_decisions_class_evidence",
+        "ck_decisions_resolution_source",
+        "ck_decisions_decision_class",
+    ):
+        op.drop_constraint(name, "decisions", type_="check")
+    for column in (
+        "evidence_refs",
+        "resolution_evidence",
+        "resolution_source",
+        "approval_id",
+        "memo_id",
+        "decision_class",
+    ):
+        op.drop_column("decisions", column)

tests/test_decision_evidence.py

@@ -0,0 +1,472 @@
+"""STATE-WP-0101-T02/T03: decision classes, evidence refusals and the write-back route.
+
+Design: docs/decision-services-alignment-answer.md sections 3-6, founder decisions
+of 2026-11-01 (section 8.1). The token reviewer is a fake; no live token is used.
+"""
+from __future__ import annotations
+
+import importlib.util
+import uuid
+from pathlib import Path
+
+import pytest
+import sqlalchemy
+
+from api.config import settings
+from api.services.decision_writeback import TokenReviewResult, TokenReviewUnavailable, get_token_reviewer
+from tests.conftest import _SYNC_URL, create_test_domain, create_test_topic
+
+WORKER = "system:serviceaccount:state-hub:state-hub"
+AUDIENCE = "state-hub-decision-writeback"
+GOOD_TOKEN = "worker-token"
+OTHER_TOKEN = "other-sa-token"
+WRONG_AUD_TOKEN = "wrong-audience-token"
+
+
+class FakeReviewer:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str]] = []
+        self.down = False
+
+    async def review(self, token: str, audience: str) -> TokenReviewResult:
+        self.calls.append((token, audience))
+        if self.down:
+            raise TokenReviewUnavailable("down")
+        if token == GOOD_TOKEN:
+            return TokenReviewResult(True, WORKER, (audience,))
+        if token == OTHER_TOKEN:
+            return TokenReviewResult(True, "system:serviceaccount:other:agent", (audience,))
+        if token == WRONG_AUD_TOKEN:
+            return TokenReviewResult(True, WORKER, ())
+        return TokenReviewResult(False)
+
+
+@pytest.fixture
+def reviewer(client, monkeypatch):
+    from api.main import app
+
+    fake = FakeReviewer()
+    app.dependency_overrides[get_token_reviewer] = lambda: fake
+    monkeypatch.setattr(settings, "decision_writeback_enabled", True)
+    return fake
+
+
+async def _topic(client):
+    await create_test_domain(client)
+    return await create_test_topic(client)
+
+
+async def _decision(client, topic, **extra):
+    r = await client.post(
+        "/decisions/", json={"title": "Pick a vendor", "topic_id": topic["id"], **extra}
+    )
+    assert r.status_code == 201, r.text
+    return r.json()
+
+
+def _outcome(decision_id, memo_id="memo-abc", **extra):
+    record = {
+        "kind": "informed-decision.decision-outcome",
+        "version": 1,
+        "hub_decision_id": decision_id,
+        "memo_id": memo_id,
+        "memo_revision": 2,
+        "verb": "accept",
+        "subject": {"iss": "https://id.example/realm", "sub": "u-123"},
+        "assurance": {"acr": "aal2", "auth_time": "2026-11-01T09:00:00Z"},
+        "decided_at": "2026-11-01T09:05:00Z",
+        "view_hash": "sha256:" + "a" * 64,
+        "presentation_id": "pres-1",
+        "display_name": "Bernd",
+    }
+    record.update(extra)
+    return record
+
+
+def _auth(token=GOOD_TOKEN):
+    return {"Authorization": f"Bearer {token}"}
+
+
+async def _write(client, decision_id, body, token=GOOD_TOKEN):
+    return await client.post(
+        f"/decisions/{decision_id}/resolve-evidence", json=body, headers=_auth(token)
+    )
+
+
+# ---------------------------------------------------------------------------
+# T02: legacy / C0 unchanged
+# ---------------------------------------------------------------------------
+
+
+async def test_legacy_decision_reads_new_fields_and_resolves_by_text(client):
+    topic = await _topic(client)
+    d = await _decision(client, topic)
+    assert d["decision_class"] is None
+    assert d["evidence_refs"] == []
+    assert d["memo_id"] is None and d["resolution_evidence"] is None
+    r = await client.post(f"/decisions/{d['id']}/resolve", json={"rationale": "ok", "decided_by": "b"})
+    assert r.status_code == 200
+    assert r.json()["status"] == "resolved"
+    assert r.json()["resolution_source"] == "text"
+
+
+async def test_c0_patch_and_resolved_create_as_today(client):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C0")
+    r = await client.patch(
+        f"/decisions/{d['id']}", json={"status": "resolved", "decided_by": "b", "rationale": "x"}
+    )
+    assert r.status_code == 200 and r.json()["status"] == "resolved"
+    created = await _decision(client, topic, decision_class="C0", status="resolved")
+    assert created["status"] == "resolved"
+
+
+async def test_evidence_refs_typed_and_any_class(client):
+    topic = await _topic(client)
+    d = await _decision(
+        client,
+        topic,
+        decision_class="C1",
+        evidence_refs=[{"type": "ccr", "ref": "CCR-0058"}, {"type": "commit", "ref": "abc", "digest": "sha1:abc"}],
+    )
+    assert [x["type"] for x in d["evidence_refs"]] == ["ccr", "commit"]
+    r = await client.patch(
+        f"/decisions/{d['id']}", json={"evidence_refs": [{"type": "workplan_task", "ref": "STATE-WP-0101-T02"}]}
+    )
+    assert r.status_code == 200
+    assert r.json()["evidence_refs"][0]["ref"] == "STATE-WP-0101-T02"
+    bad = await client.post(
+        "/decisions/",
+        json={"title": "x", "topic_id": topic["id"], "evidence_refs": [{"type": "tweet", "ref": "y"}]},
+    )
+    assert bad.status_code == 422
+
+
+async def test_list_filters_by_class(client):
+    topic = await _topic(client)
+    await _decision(client, topic)
+    c1 = await _decision(client, topic, decision_class="C1")
+    r = await client.get("/decisions/", params={"decision_class": "C1"})
+    assert [x["id"] for x in r.json()] == [c1["id"]]
+
+
+# ---------------------------------------------------------------------------
+# T02: refusals (design section 4)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("klass", ["C1", "C2"])
+async def test_text_resolve_refused_for_c1_c2(client, klass):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class=klass)
+    r = await client.post(f"/decisions/{d['id']}/resolve", json={"rationale": "ok", "decided_by": "b"})
+    assert r.status_code == 409
+    assert r.json()["detail"]["error"] == "evidence_required"
+    again = await client.get(f"/decisions/{d['id']}")
+    assert again.json()["status"] == "open"
+
+
+@pytest.mark.parametrize(
+    "patch",
+    [
+        {"status": "resolved"},
+        {"decided_by": "b"},
+        {"decided_at": "2026-11-01T00:00:00Z"},
+        {"rationale": "because"},
+    ],
+)
+async def test_resolving_patch_fields_refused_for_c1(client, patch):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    r = await client.patch(f"/decisions/{d['id']}", json=patch)
+    assert r.status_code == 409
+    assert r.json()["detail"]["error"] == "evidence_required"
+
+
+async def test_raise_and_resolve_in_one_patch_refused(client):
+    topic = await _topic(client)
+    d = await _decision(client, topic)
+    r = await client.patch(f"/decisions/{d['id']}", json={"decision_class": "C2", "status": "resolved"})
+    assert r.status_code == 409
+    assert r.json()["detail"]["error"] == "evidence_required"
+
+
+@pytest.mark.parametrize("field", ["memo_id", "approval_id", "resolution_source", "resolution_evidence"])
+async def test_link_fields_not_writable_by_patch_or_create(client, field):
+    topic = await _topic(client)
+    d = await _decision(client, topic)
+    value = str(uuid.uuid4()) if field == "approval_id" else ({} if field == "resolution_evidence" else "x")
+    r = await client.patch(f"/decisions/{d['id']}", json={field: value})
+    assert r.status_code == 422
+    r = await client.post("/decisions/", json={"title": "t", "topic_id": topic["id"], field: value})
+    assert r.status_code == 422
+
+
+async def test_class_raise_allowed_lowering_refused(client):
+    topic = await _topic(client)
+    d = await _decision(client, topic)
+    assert (await client.patch(f"/decisions/{d['id']}", json={"decision_class": "C1"})).status_code == 200
+    assert (await client.patch(f"/decisions/{d['id']}", json={"decision_class": "C2"})).status_code == 200
+    for lower in ("C1", "C0", None):
+        r = await client.patch(f"/decisions/{d['id']}", json={"decision_class": lower})
+        assert r.status_code == 409
+        assert r.json()["detail"]["error"] == "decision_class_lowering"
+
+
+async def test_class_of_resolved_decision_frozen(client):
+    topic = await _topic(client)
+    d = await _decision(client, topic)
+    await client.post(f"/decisions/{d['id']}/resolve", json={"rationale": "ok", "decided_by": "b"})
+    r = await client.patch(f"/decisions/{d['id']}", json={"decision_class": "C1"})
+    assert r.status_code == 409
+    assert r.json()["detail"]["error"] == "decision_class_frozen"
+
+
+async def test_c3_reserved(client):
+    topic = await _topic(client)
+    r = await client.post("/decisions/", json={"title": "t", "topic_id": topic["id"], "decision_class": "C3"})
+    assert r.status_code == 422
+    assert r.json()["detail"]["error"] == "decision_class_reserved"
+    d = await _decision(client, topic)
+    r = await client.patch(f"/decisions/{d['id']}", json={"decision_class": "C3"})
+    assert r.status_code == 422
+
+
+@pytest.mark.parametrize("klass", ["C1", "C2"])
+async def test_resolved_create_refused_for_c1_c2(client, klass):
+    topic = await _topic(client)
+    r = await client.post(
+        "/decisions/",
+        json={"title": "t", "topic_id": topic["id"], "decision_class": klass, "status": "resolved"},
+    )
+    assert r.status_code == 409
+    assert r.json()["detail"]["error"] == "evidence_required"
+
+
+async def test_supersede_allowed_for_c2(client):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C2")
+    r = await client.delete(f"/decisions/{d['id']}")
+    assert r.status_code == 200 and r.json()["status"] == "superseded"
+
+
+async def test_database_check_refuses_resolved_c1_without_evidence(client):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    engine = sqlalchemy.create_engine(_SYNC_URL)
+    try:
+        with pytest.raises(sqlalchemy.exc.IntegrityError):
+            with engine.begin() as conn:
+                conn.execute(
+                    sqlalchemy.text("UPDATE decisions SET status='resolved' WHERE id=:id"),
+                    {"id": d["id"]},
+                )
+    finally:
+        engine.dispose()
+
+
+def test_migration_constraints_match_model():
+    from api.models import decision as model
+
+    path = Path(__file__).parent.parent / "migrations/versions/d5e6f7a8b9c0_decision_evidence_links.py"
+    spec = importlib.util.spec_from_file_location("mig_d5e6", path)
+    mig = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mig)
+    for name in ("CK_DECISION_CLASS", "CK_RESOLUTION_SOURCE", "CK_CLASS_EVIDENCE", "CK_CLASS_APPROVAL"):
+        assert getattr(mig, name) == getattr(model, name)
+    assert mig.down_revision == "c4d5e6f7a8b9"
+
+
+# ---------------------------------------------------------------------------
+# T03: caller verification
+# ---------------------------------------------------------------------------
+
+
+async def test_route_disabled_by_default(client, monkeypatch):
+    monkeypatch.setattr(settings, "decision_writeback_enabled", False)
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    r = await _write(client, d["id"], {"mode": "link", "memo_id": "memo-abc"})
+    assert r.status_code == 503
+    assert r.json()["detail"]["error"] == "decision_writeback_disabled"
+
+
+@pytest.mark.parametrize(
+    ("headers", "status", "error"),
+    [
+        ({}, 401, "token_required"),
+        (_auth("garbage"), 401, "token_invalid"),
+        (_auth(WRONG_AUD_TOKEN), 401, "token_invalid"),
+        (_auth(OTHER_TOKEN), 403, "subject_not_allowed"),
+    ],
+)
+async def test_route_refuses_unverified_callers(client, reviewer, headers, status, error):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    r = await client.post(
+        f"/decisions/{d['id']}/resolve-evidence",
+        json={"mode": "link", "memo_id": "memo-abc"},
+        headers=headers,
+    )
+    assert r.status_code == status
+    assert r.json()["detail"]["error"] == error
+    assert (await client.get(f"/decisions/{d['id']}")).json()["memo_id"] is None
+
+
+async def test_token_review_audience_and_fail_closed(client, reviewer):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    reviewer.down = True
+    r = await _write(client, d["id"], {"mode": "link", "memo_id": "memo-abc"})
+    assert r.status_code == 503
+    assert reviewer.calls == [(GOOD_TOKEN, AUDIENCE)]
+
+
+# ---------------------------------------------------------------------------
+# T03: link and resolve
+# ---------------------------------------------------------------------------
+
+
+async def test_c1_link_then_resolve_idempotent(client, reviewer):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    r = await _write(client, d["id"], {"mode": "link", "memo_id": "memo-abc"})
+    assert r.status_code == 200
+    assert r.json()["memo_id"] == "memo-abc" and r.json()["status"] == "open"
+    assert (await _write(client, d["id"], {"mode": "link", "memo_id": "memo-abc"})).status_code == 200
+
+    body = {"mode": "resolve", "memo_id": "memo-abc", "outcome": _outcome(d["id"], note="go ahead")}
+    r = await _write(client, d["id"], body)
+    assert r.status_code == 200, r.text
+    got = r.json()
+    assert got["status"] == "resolved"
+    assert got["resolution_source"] == "informed_decision"
+    assert got["decided_by"] == "Bernd"
+    assert got["resolution_evidence"]["subject"] == {"iss": "https://id.example/realm", "sub": "u-123"}
+    assert got["resolution_evidence"]["view_hash"] == "sha256:" + "a" * 64
+    assert "accepted in informed-decision, memo memo-abc revision 2" in got["rationale"]
+
+    again = await _write(client, d["id"], body)
+    assert again.status_code == 200
+    events = (await client.get("/progress/", params={"decision_id": d["id"]})).json()
+    resolved = [e for e in events if e.get("event_type") == "decision_resolved"]
+    assert len(resolved) == 1
+    assert resolved[0]["detail"]["attested_via"] == WORKER
+
+
+async def test_decline_resolves_without_prior_link(client, reviewer):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    body = {"mode": "resolve", "memo_id": "memo-dec", "outcome": _outcome(d["id"], "memo-dec", verb="decline")}
+    r = await _write(client, d["id"], body)
+    assert r.status_code == 200
+    assert r.json()["rationale"].startswith("declined")
+
+
+async def test_different_memo_conflicts(client, reviewer):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    await _write(client, d["id"], {"mode": "link", "memo_id": "memo-abc"})
+    r = await _write(client, d["id"], {"mode": "link", "memo_id": "memo-xyz"})
+    assert r.status_code == 409
+    assert r.json()["detail"] == {
+        "error": "memo_conflict",
+        "message": "the decision is linked to another memo",
+        "memo_id": "memo-abc",
+    }
+
+
+async def test_memo_answers_one_decision_only(client, reviewer):
+    topic = await _topic(client)
+    a = await _decision(client, topic, decision_class="C1")
+    b = await _decision(client, topic, decision_class="C1")
+    assert (await _write(client, a["id"], {"mode": "link", "memo_id": "memo-abc"})).status_code == 200
+    r = await _write(client, b["id"], {"mode": "link", "memo_id": "memo-abc"})
+    assert r.status_code == 409
+    assert r.json()["detail"]["error"] == "memo_already_linked"
+
+
+async def test_c0_and_legacy_refused_on_evidence_route(client, reviewer):
+    topic = await _topic(client)
+    for extra in ({}, {"decision_class": "C0"}):
+        d = await _decision(client, topic, **extra)
+        r = await _write(client, d["id"], {"mode": "link", "memo_id": "memo-abc"})
+        assert r.status_code == 409
+        assert r.json()["detail"]["error"] == "evidence_not_applicable"
+
+
+async def test_c2_requires_approval(client, reviewer):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C2")
+    body = {"mode": "resolve", "memo_id": "memo-c2", "outcome": _outcome(d["id"], "memo-c2")}
+    r = await _write(client, d["id"], body)
+    assert r.status_code == 422
+    assert r.json()["detail"]["error"] == "approval_required"
+
+    approval = str(uuid.uuid4())
+    r = await _write(client, d["id"], {"mode": "link", "memo_id": "memo-c2", "approval_id": approval})
+    assert r.status_code == 200 and r.json()["approval_id"] == approval
+    other = _outcome(d["id"], "memo-c2", approval_id=str(uuid.uuid4()), approval_entry_id="e1")
+    r = await _write(client, d["id"], {"mode": "resolve", "memo_id": "memo-c2", "outcome": other})
+    assert r.status_code == 409 and r.json()["detail"]["error"] == "approval_conflict"
+
+    good = _outcome(d["id"], "memo-c2", approval_id=approval, approval_entry_id="e1")
+    r = await _write(client, d["id"], {"mode": "resolve", "memo_id": "memo-c2", "outcome": good})
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "resolved" and r.json()["approval_id"] == approval
+
+
+@pytest.mark.parametrize(
+    ("mutate", "error"),
+    [
+        (lambda o, d: o.update(hub_decision_id=str(uuid.uuid4())), "hub_decision_mismatch"),
+        (lambda o, d: o.update(memo_id="memo-other"), "memo_mismatch"),
+        (lambda o, d: o.update(note="password: hunter2hunter2hunter2"), "secret_marker"),
+    ],
+)
+async def test_outcome_checks(client, reviewer, mutate, error):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    outcome = _outcome(d["id"])
+    mutate(outcome, d)
+    r = await _write(client, d["id"], {"mode": "resolve", "memo_id": "memo-abc", "outcome": outcome})
+    assert r.status_code == 422
+    assert r.json()["detail"]["error"] == error
+    assert (await client.get(f"/decisions/{d['id']}")).json()["status"] == "open"
+
+
+@pytest.mark.parametrize(
+    "mutate",
+    [
+        lambda o: o.update(verb="comment"),
+        lambda o: o.update(view_hash="sha256:short"),
+        lambda o: o.update(packet={"content": "x"}),
+        lambda o: o.update(note="x" * 2049),
+        lambda o: o.update(decided_at="2026-11-01T09:05:00"),
+        lambda o: o.update(assurance={"auth_time": "2026-11-01T09:00:00Z"}),
+        lambda o: o.pop("subject"),
+    ],
+)
+async def test_outcome_record_shape(client, reviewer, mutate):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    outcome = _outcome(d["id"])
+    mutate(outcome)
+    r = await _write(client, d["id"], {"mode": "resolve", "memo_id": "memo-abc", "outcome": outcome})
+    assert r.status_code == 422
+
+
+async def test_superseded_refused(client, reviewer):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    await client.delete(f"/decisions/{d['id']}")
+    r = await _write(client, d["id"], {"mode": "link", "memo_id": "memo-abc"})
+    assert r.status_code == 409 and r.json()["detail"]["error"] == "decision_superseded"
+
+
+async def test_evidence_resolved_not_reopened_by_patch(client, reviewer):
+    topic = await _topic(client)
+    d = await _decision(client, topic, decision_class="C1")
+    await _write(client, d["id"], {"mode": "resolve", "memo_id": "memo-abc", "outcome": _outcome(d["id"])})
+    r = await client.patch(f"/decisions/{d['id']}", json={"status": "open"})
+    assert r.status_code == 409 and r.json()["detail"]["error"] == "evidence_bound"

workplans/STATE-WP-0101-decision-evidence-links.md

@@ -52,7 +52,7 @@ recorded in section 8.1 of the design:
 
 ```task
 id: STATE-WP-0101-T02
-status: wait
+status: done
 priority: high
 state_hub_task_id: "67e64f95-4869-5aef-9f44-5fcc688a1012"
 ```
@@ -63,11 +63,19 @@ index on `memo_id`). Optional fields on `DecisionRead`; typed `evidence_refs` va
 text resolution, resolving PATCH fields, class lowering and resolved creation for C1/C2 with 409
 `evidence_required`. Tests for C0/legacy unchanged behaviour and every refusal in section 4.
 
+Done 2026-11-01: migration `d5e6f7a8b9c0` (upgrade/downgrade/upgrade checked on a scratch local
+database only; not applied to the live hub), model, schemas, refusals, `decision_class` list
+filter; `tests/test_decision_evidence.py`. Additions beyond the text: C3 is refused with 422
+(GH-DEC-2026-026, reserved) and treated fail-closed like C2 by the CHECK constraints;
+`memo_id`, `approval_id` and `resolution_*` are writable only through the evidence route (422 on
+PATCH/create for any class, so a C0 cannot be pre-filled and then raised); the class of a resolved
+or superseded decision is frozen; an evidence-resolved decision is not reopened by PATCH.
+
 ## Authenticated evidence route
 
 ```task
 id: STATE-WP-0101-T03
-status: wait
+status: done
 priority: high
 state_hub_task_id: "fed28083-1eda-5f99-abf7-39c02b45988e"
 ```
@@ -78,6 +86,12 @@ Prerequisite: T02 and the RBAC decision in T01 (decision 2, accepted 2026-11-01)
 5.2, idempotency on `(decision_id, memo_id)`, progress event and enriched resolved event. Tests use
 a fake token reviewer; no live token.
 
+Done 2026-11-01: `api/services/decision_writeback.py` and the route, behind
+`DECISION_WRITEBACK_ENABLED` (default off, 503 `decision_writeback_disabled`). TokenReview runs
+with the API pod's own ServiceAccount; review failure is 503, fail closed. The `option`
+requirement for C1 with options is not enforced (the hub does not know a memo's options).
+Enabling needs the founder-run `system:auth-delegator` binding and T05's cluster prerequisites.
+
 ## Dispatch worker: outcome pull and write-back
 
 ```task