feat(feedback): content-free thumbs on assistant messages, joined to cost rows (offload PR-7)

backend/src/apis/app_api/admin/costs/models.py

@@ -463,6 +463,44 @@ class ToolCensusEntry(BaseModel):
     errors: int = 0
 
 
+class FeedbackCounts(BaseModel):
+    """Thumbs on one bucket of calls: ``up`` / ``down`` are counts of live
+    feedback rows, never the text of anything."""
+    model_config = ConfigDict(populate_by_name=True)
+
+    up: int = 0
+    down: int = 0
+
+
+class FeedbackByTurnClass(BaseModel):
+    """Feedback split by the call's document turn class (document-context
+    offload spec §6.1): *full* (``hasDocuments``), *retrieved*
+    (``documentReads.pages > 0``), *digestOnly* (``documentDigests > 0``),
+    else *none* — in that precedence, since a retrieving call still holds
+    the digest. ``n`` per class is ``up + down``; the down-thumb rate is
+    ``down / n``."""
+    model_config = ConfigDict(populate_by_name=True)
+
+    full: FeedbackCounts = Field(default_factory=FeedbackCounts)
+    digest_only: FeedbackCounts = Field(default_factory=FeedbackCounts, alias="digestOnly")
+    retrieved: FeedbackCounts = Field(default_factory=FeedbackCounts)
+    none: FeedbackCounts = Field(default_factory=FeedbackCounts)
+
+
+class FeedbackProfile(BaseModel):
+    """The outcome signal joined to the session's cost rows. ``byTurnClass``
+    is ``None`` when no cost row carries the turn-class fields (they arrive
+    with #1137; rows written before it have none) — "not tracked", not zero.
+    ``unjoined`` counts thumbs whose message has no cost row at all (the row
+    expired, or the call was never recorded)."""
+    model_config = ConfigDict(populate_by_name=True)
+
+    up: int = 0
+    down: int = 0
+    by_turn_class: Optional[FeedbackByTurnClass] = Field(None, alias="byTurnClass")
+    unjoined: int = 0
+
+
 class DataCoverage(BaseModel):
     """Which optional signals this session actually has, so the UI can say
     "not tracked" instead of rendering an honest-looking zero."""
@@ -475,6 +513,8 @@ class DataCoverage(BaseModel):
     prefix_tokens: bool = Field(False, alias="prefixTokens")
     window_trim: bool = Field(False, alias="windowTrim")
     compaction_events: bool = Field(False, alias="compactionEvents")
+    # Any F# row, or a session rollup written while diagnostics were on.
+    feedback: bool = False
     documents: bool = False
 
 
@@ -518,6 +558,8 @@ class SessionProfile(BaseModel):
     )
     # The summary's token size at the most recent compaction decision.
     last_summary_tokens: Optional[int] = Field(None, alias="lastSummaryTokens")
+    # Thumbs up/down joined to the cost rows by (sessionId, messageId).
+    feedback: FeedbackProfile = Field(default_factory=FeedbackProfile)
     # Document lifecycle across the session's calls: how many calls ran with
     # the full document inline vs. a digest only (the digest-vs-full turn
     # shares), the largest estimated document footprint seen, and what

backend/src/apis/app_api/admin/costs/service.py

@@ -26,6 +26,9 @@
     AttachmentProfile,
     ContextTrajectoryPoint,
     DataCoverage,
+    FeedbackByTurnClass,
+    FeedbackCounts,
+    FeedbackProfile,
     FingerprintChanges,
     SessionDiagnosis,
     SessionProfile,
@@ -78,6 +81,84 @@ def _record_cost(record: Dict[str, Any]) -> Optional[float]:
     return _as_float(raw)
 
 
+def _turn_class(record: Dict[str, Any]) -> Optional[str]:
+    """Turn class of one ``C#`` row from its document-context fields (spec
+    §6.1): ``full`` (``hasDocuments``), ``retrieved`` (``documentReads.pages
+    > 0``), ``digestOnly`` (``documentDigests > 0``), else ``none``.
+
+    Precedence is full > retrieved > digestOnly: a call that pulled pages back
+    still holds the digest, so testing the digest first would leave the
+    retrieved arm — the one the quality gate is about — permanently empty.
+    ``None`` when the row carries none of the fields (written before #1137
+    or with diagnostics off), so the caller says "not tracked", not "none".
+    """
+    has_documents = record.get("hasDocuments")
+    digests = record.get("documentDigests")
+    reads = record.get("documentReads")
+    if has_documents is None and digests is None and reads is None:
+        return None
+    if has_documents:
+        return "full"
+    pages = _as_int(reads.get("pages")) if isinstance(reads, dict) else _as_int(reads)
+    if (pages or 0) > 0:
+        return "retrieved"
+    if (_as_int(digests) or 0) > 0:
+        return "digestOnly"
+    return "none"
+
+
+def _join_feedback(
+    records: List[Dict[str, Any]],
+    feedback_rows: List[Dict[str, Any]],
+) -> FeedbackProfile:
+    """Join ``F#`` rows to ``C#`` rows on ``messageId`` and bucket by turn
+    class. Pure; the profile's numbers, never any content. Explicit thumbs
+    only (``signal`` absent or ``"explicit"``)."""
+    by_message: Dict[int, Dict[str, Any]] = {}
+    for record in records:
+        message_id = _as_int(record.get("messageId"))
+        if message_id is not None:
+            # The last call of a multi-call turn is the one the user thumbed;
+            # rows share a messageId only across the turn's tool round trips
+            # and later rows have the fuller context, so last write wins.
+            by_message[message_id] = record
+
+    any_turn_class = any(_turn_class(r) is not None for r in records)
+    buckets = FeedbackByTurnClass() if any_turn_class else None
+    profile = FeedbackProfile()
+    for row in feedback_rows:
+        # Explicit thumbs only — implicit signals (spec §10) share the row
+        # family but answer a different question and must never be summed in.
+        if row.get("signal") not in (None, "explicit"):
+            continue
+        value = _as_int(row.get("value"))
+        if value not in (1, -1):
+            continue
+        if value == 1:
+            profile.up += 1
+        else:
+            profile.down += 1
+        message_id = _as_int(row.get("messageId"))
+        record = by_message.get(message_id) if message_id is not None else None
+        if record is None:
+            profile.unjoined += 1
+            continue
+        if buckets is None:
+            continue
+        klass = _turn_class(record) or "none"
+        bucket: FeedbackCounts = {
+            "full": buckets.full,
+            "digestOnly": buckets.digest_only,
+            "retrieved": buckets.retrieved,
+        }.get(klass, buckets.none)
+        if value == 1:
+            bucket.up += 1
+        else:
+            bucket.down += 1
+    profile.by_turn_class = buckets
+    return profile
+
+
 def _context_tokens(record: Dict[str, Any]) -> int:
     """True context occupancy of one call: uncached input + cached prefix +
     newly cached tokens (Bedrock reports the three disjointly)."""
@@ -1036,6 +1117,20 @@ async def _attachment_profile(self, session_id: str) -> AttachmentProfile:
             digest_tokens=digest_tokens,
         )
 
+    async def _feedback_rows(self, session_id: str) -> List[Dict[str, Any]]:
+        """The session's ``F#`` rows. Best-effort: a storage fork without the
+        reader, or a transient error, yields none — the profile then falls
+        back to the session rollups and reports coverage honestly."""
+        reader = getattr(self.storage, "get_session_feedback_rows", None)
+        if reader is None:
+            return []
+        try:
+            rows = await reader(session_id)
+        except Exception as e:  # noqa: BLE001 - feedback is one signal of several
+            logger.debug("Feedback rows unavailable for session: %s", e)
+            return []
+        return list(rows or [])
+
     async def get_session_profile(self, session_id: str) -> Optional[SessionProfile]:
         """The content-free diagnostic profile of one conversation, or ``None``
         when the session has no metadata row.
@@ -1052,6 +1147,7 @@ async def get_session_profile(self, session_id: str) -> Optional[SessionProfile]
 
         records = await self.storage.get_session_cost_records(session_id)
         attachments = await self._attachment_profile(session_id)
+        feedback_rows = await self._feedback_rows(session_id)
         user_period_cost = (
             await self._user_period_cost(user_id, self._get_current_period())
             if user_id else None
@@ -1192,6 +1288,15 @@ async def get_session_profile(self, session_id: str) -> Optional[SessionProfile]
             facts.tool_call_count = sum(e.calls for e in census.values())
             facts.tool_error_count = sum(e.errors for e in census.values())
 
+        # Outcome signal: thumbs joined to the calls they rate. The rows are
+        # authoritative when present; the session rollups cover thumbs whose
+        # rows expired (they share the C# TTL, so this is rare).
+        feedback = _join_feedback(records, feedback_rows)
+        if not feedback_rows:
+            feedback.up = _as_int(row.get("thumbsUp")) or 0
+            feedback.down = _as_int(row.get("thumbsDown")) or 0
+        feedback_tracked = bool(feedback_rows) or row.get("thumbsUp") is not None
+
         findings = run_diagnoses(facts)
         summary = self._session_summary(row, findings, share)
         if any_census and summary.tool_call_count is None:
@@ -1229,8 +1334,10 @@ async def get_session_profile(self, session_id: str) -> Optional[SessionProfile]
                 compaction_events=(
                     any_compaction_events or row.get("compactionAppliedCount") is not None
                 ),
+                feedback=feedback_tracked,
                 documents=any_documents or row.get("fullDocumentCalls") is not None,
             ),
+            feedback=feedback,
             prefix_tokens=prefix_tokens,
             window_trim_calls=window_trim_calls,
             window_removed_messages=last_removed,

backend/src/apis/app_api/messages/models.py

@@ -117,6 +117,38 @@ class Citation(BaseModel):
     text: str = Field(..., description="Relevant text excerpt from the document")
 
 
+#: Reason codes a down-thumb may carry — the six buckets of
+#: ``docs/specs/response-feedback.md`` §6, each of which routes to an
+#: evaluator or an ops signal. A closed enum, never free text: the row is
+#: content-free by construction so it can sit beside the ``C#`` cost row and
+#: be read by the admin profile without reading the conversation. The spec's
+#: "something else → free text" is deliberately not here; that hand-off is
+#: the existing Agent report dialog (spec §3), which already has moderation.
+FEEDBACK_REASONS = ("wrong", "instructions", "length", "tool_failed", "outdated", "other")
+FeedbackReason = Literal["wrong", "instructions", "length", "tool_failed", "outdated", "other"]
+
+
+class MessageFeedback(BaseModel):
+    """One user's thumb on one assistant message (``F#`` row, see
+    ``apis.shared.sessions.metadata``). ``value`` is +1 (up) or -1 (down);
+    ``reason`` is an optional code from ``FEEDBACK_REASONS``."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    value: Literal[1, -1] = Field(..., description="+1 for thumbs up, -1 for thumbs down")
+    reason: Optional[FeedbackReason] = Field(None, description="Optional reason code (never free text)")
+    updated_at: str = Field(..., alias="updatedAt", description="ISO timestamp of the latest thumb")
+
+
+class MessageFeedbackRequest(BaseModel):
+    """Body of ``PUT /sessions/{id}/messages/{messageId}/feedback``."""
+
+    model_config = ConfigDict(populate_by_name=True)
+
+    value: Literal[1, -1] = Field(..., description="+1 for thumbs up, -1 for thumbs down")
+    reason: Optional[FeedbackReason] = Field(None, description="Optional reason code (never free text)")
+
+
 class MessageMetadata(BaseModel):
     """Metadata associated with a single message"""
 
@@ -129,8 +161,10 @@ class MessageMetadata(BaseModel):
     cost: Optional[Union[float, Dict[str, float]]] = Field(None, description="Cost for this message — either a total float (legacy) or a breakdown dict with total, inputCost, outputCost, cacheReadCost, cacheWriteCost")
     citations: Optional[List[Dict[str, str]]] = Field(None, description="RAG citations for this message (stored as dicts for flexible JSON storage)")
     display_text: Optional[str] = Field(None, alias="displayText", description="Original user message text before RAG augmentation (for clean UI display)")
-    # Note: Feedback will be added in future implementation
-    # feedback: Optional[Feedback] = None
+    # One user's thumb on this message, merged from the ``F#`` row on read
+    # (see ``apis.shared.sessions.metadata``). Content-free: a ±1, a timestamp
+    # and an optional reason code.
+    feedback: Optional[MessageFeedback] = Field(None, description="User thumbs up/down on this assistant message")
 
 
 class Message(BaseModel):

backend/src/apis/app_api/sessions/routes.py

@@ -3,7 +3,7 @@
 Provides endpoints for managing session metadata.
 """
 
-from fastapi import APIRouter, HTTPException, Depends, Query, Response, BackgroundTasks, status
+from fastapi import APIRouter, HTTPException, Depends, Path, Query, Response, BackgroundTasks, status
 from typing import Optional
 import logging
 from apis.shared.sessions.models import (
@@ -18,7 +18,14 @@
     BulkDeleteSessionsRequest,
     BulkDeleteSessionsResponse,
     BulkDeleteSessionResult,
-    MessagesListResponse
+    MessagesListResponse,
+    MessageFeedback,
+    MessageFeedbackRequest,
+)
+from apis.shared.sessions.feedback import (
+    SessionNotOwned,
+    delete_message_feedback,
+    put_message_feedback,
 )
 from apis.shared.sessions.messages import get_messages
 from apis.shared.sessions.metadata import (
@@ -35,7 +42,7 @@
 from apis.app_api.shares.service import get_share_service
 from apis.app_api.artifacts.service import get_artifact_share_service
 from apis.shared.auth.dependencies import get_current_user_from_session
-from apis.shared.feature_flags import mid_turn_steering_enabled
+from apis.shared.feature_flags import response_feedback_enabled, mid_turn_steering_enabled
 from apis.shared.auth.models import User
 from apis.shared.system_prompts.service import get_system_prompts_service
 
@@ -662,6 +669,80 @@ async def get_session_messages_endpoint(
         )
 
 
+def _require_message_feedback() -> None:
+    """404 while ``RESPONSE_FEEDBACK_ENABLED=false`` — the surface does not exist."""
+    if not response_feedback_enabled():
+        raise HTTPException(status_code=404, detail="Not found")
+
+
+@router.put(
+    "/{session_id}/messages/{message_id}/feedback",
+    response_model=MessageFeedback,
+    response_model_by_alias=True,
+    response_model_exclude_none=True,
+)
+async def put_message_feedback_endpoint(
+    session_id: str,
+    message_id: int = Path(..., ge=0, description="0-based message index"),
+    body: MessageFeedbackRequest = ...,
+    current_user: User = Depends(get_current_user_from_session),
+):
+    """Thumb an assistant message up (+1) or down (-1), optionally with a
+    reason code. Idempotent per (user, message): a second click replaces the
+    first. Content-free by construction — the body is a closed enum, so no
+    text can be stored (``apis.shared.sessions.feedback``).
+
+    ``message_id`` is the message's 0-based index in the conversation — the
+    trailing number of the SPA's ``msg-{sessionId}-{index}`` id, and the
+    ``messageId`` the message's cost row carries.
+    """
+    _require_message_feedback()
+    try:
+        return await put_message_feedback(
+            session_id=session_id,
+            user_id=current_user.user_id,
+            message_id=message_id,
+            value=body.value,
+            reason=body.reason,
+        )
+    except SessionNotOwned:
+        raise HTTPException(status_code=404, detail=f"Session not found: {session_id}")
+    except ValueError as e:
+        raise HTTPException(status_code=422, detail=str(e))
+    except RuntimeError as e:
+        # No metadata table configured (local dev without DynamoDB).
+        logger.warning("Message feedback unavailable: %s", scrub_log(str(e)))
+        raise HTTPException(status_code=503, detail="Message feedback is not available")
+    except Exception:
+        logger.error("Error storing message feedback", exc_info=True)
+        raise HTTPException(status_code=500, detail="Failed to store message feedback")
+
+
+@router.delete("/{session_id}/messages/{message_id}/feedback", status_code=204)
+async def delete_message_feedback_endpoint(
+    session_id: str,
+    message_id: int = Path(..., ge=0, description="0-based message index"),
+    current_user: User = Depends(get_current_user_from_session),
+):
+    """Withdraw this user's thumb on a message. 204 whether or not one existed."""
+    _require_message_feedback()
+    try:
+        await delete_message_feedback(
+            session_id=session_id,
+            user_id=current_user.user_id,
+            message_id=message_id,
+        )
+    except SessionNotOwned:
+        raise HTTPException(status_code=404, detail=f"Session not found: {session_id}")
+    except RuntimeError as e:
+        logger.warning("Message feedback unavailable: %s", scrub_log(str(e)))
+        raise HTTPException(status_code=503, detail="Message feedback is not available")
+    except Exception:
+        logger.error("Error deleting message feedback", exc_info=True)
+        raise HTTPException(status_code=500, detail="Failed to delete message feedback")
+    return Response(status_code=204)
+
+
 @router.post("/{session_id}/interrupt", status_code=204)
 async def signal_turn_interrupted_endpoint(
     session_id: str,