feat(feedback): eval sampling — down-thumbs feed AgentCore Evaluations (response-feedback PR-4)

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

@@ -487,6 +487,27 @@ class FeedbackByTurnClass(BaseModel):
     none: FeedbackCounts = Field(default_factory=FeedbackCounts)
 
 
+class EvaluatorAggregate(BaseModel):
+    """Mean judged score for one evaluator over this session's sampled thumbs."""
+    model_config = ConfigDict(populate_by_name=True)
+
+    n: int = 0
+    mean: float = 0.0
+
+
+class FeedbackEvaluations(BaseModel):
+    """What the eval sampler (spec §11 PR-4) concluded about this session's
+    down-thumbs: how many were judged, the mean per evaluator, and for
+    ``tool_failed`` thumbs whether the call's tool census corroborated them.
+    Scores and counts only — the judge's explanation is never stored."""
+    model_config = ConfigDict(populate_by_name=True)
+
+    judged: int = 0
+    by_evaluator: Dict[str, EvaluatorAggregate] = Field(default_factory=dict, alias="byEvaluator")
+    tool_failures_reported: int = Field(0, alias="toolFailuresReported")
+    tool_failures_corroborated: int = Field(0, alias="toolFailuresCorroborated")
+
+
 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
@@ -505,6 +526,8 @@ class FeedbackProfile(BaseModel):
     # a cost row to price.
     retried: int = 0
     rework_usd: Optional[float] = Field(None, alias="reworkUsd")
+    # Judged down-thumbs, or None when the sampler has not touched this session.
+    evaluations: Optional[FeedbackEvaluations] = None
 
 
 class DataCoverage(BaseModel):

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

@@ -28,6 +28,8 @@
     DataCoverage,
     FeedbackByTurnClass,
     FeedbackCounts,
+    EvaluatorAggregate,
+    FeedbackEvaluations,
     FeedbackProfile,
     FingerprintChanges,
     SessionDiagnosis,
@@ -126,6 +128,8 @@ def _join_feedback(
     any_turn_class = any(_turn_class(r) is not None for r in records)
     buckets = FeedbackByTurnClass() if any_turn_class else None
     profile = FeedbackProfile()
+    evaluations = FeedbackEvaluations()
+    evaluator_sums: Dict[str, List[float]] = {}
     # Every cost row per assistant message index, for pricing rework.
     cost_by_message: Dict[int, float] = {}
     for record in records:
@@ -146,6 +150,17 @@ def _join_feedback(
         else:
             profile.down += 1
         message_id = _as_int(row.get("messageId"))
+        verdict = row.get("evaluation")
+        if isinstance(verdict, dict):
+            evaluations.judged += 1
+            for evaluator, score in (verdict.get("scores") or {}).items():
+                value = _as_float(score.get("value")) if isinstance(score, dict) else None
+                if value is not None:
+                    evaluator_sums.setdefault(str(evaluator), []).append(value)
+            if verdict.get("reason") == "tool_failed":
+                evaluations.tool_failures_reported += 1
+                if verdict.get("toolFailureCorroborated") is True:
+                    evaluations.tool_failures_corroborated += 1
         retry_id = _as_int(row.get("retryMessageId"))
         if value == -1 and retry_id is not None:
             profile.retried += 1
@@ -170,6 +185,12 @@ def _join_feedback(
             bucket.down += 1
     profile.by_turn_class = buckets
     profile.rework_usd = round(rework_total, 6) if rework_total is not None else None
+    if evaluations.judged:
+        evaluations.by_evaluator = {
+            name: EvaluatorAggregate(n=len(values), mean=round(sum(values) / len(values), 4))
+            for name, values in sorted(evaluator_sums.items())
+        }
+        profile.evaluations = evaluations
     return profile
 
 

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

@@ -0,0 +1,57 @@
+"""Admin models for the feedback eval-sampling surface (spec §11 PR-4).
+Content-free: ids, codes, scores, counts. No conversation text, and never
+the judge's explanation (the content-policy walk covers this module)."""
+
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class EvaluatorScore(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    value: float
+    rating: Optional[str] = None
+    n: int = 1
+    tokens: int = 0
+
+
+class FeedbackVerdict(BaseModel):
+    """The judged result stored on a thumb row."""
+    model_config = ConfigDict(populate_by_name=True)
+
+    reason: str = "none"
+    evaluators: List[str] = Field(default_factory=list)
+    scores: Dict[str, EvaluatorScore] = Field(default_factory=dict)
+    tool_failure_corroborated: Optional[bool] = Field(None, alias="toolFailureCorroborated")
+
+
+class DownThumbQueueItem(BaseModel):
+    """One recent down-thumb as the sampler's queue sees it."""
+    model_config = ConfigDict(populate_by_name=True)
+
+    session_id: str = Field(..., alias="sessionId")
+    message_id: int = Field(..., alias="messageId")
+    reason: Optional[str] = None
+    updated_at: str = Field("", alias="updatedAt")
+    retry_message_id: Optional[int] = Field(None, alias="retryMessageId")
+    evaluated_at: Optional[str] = Field(None, alias="evaluatedAt")
+    evaluation: Optional[FeedbackVerdict] = None
+
+
+class DownThumbQueueResponse(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    items: List[DownThumbQueueItem]
+    pending: int = Field(0, description="Items in this page not yet judged")
+    sampling_enabled: bool = Field(False, alias="samplingEnabled")
+
+
+class SamplingRunResponse(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    accepted: bool
+    limit: int
+    note: str

backend/src/apis/app_api/admin/feedback/routes.py

@@ -0,0 +1,117 @@
+"""Admin feedback routes — the eval-sampling queue (spec §11 PR-4).
+
+    GET  /admin/feedback/evaluations         recent down-thumbs + any verdict
+    POST /admin/feedback/evaluations/run     judge up to `limit` of them, offline
+
+Scope: ``admin.costs`` — the judge spends tokens and the verdicts sit beside
+the cost rows. The run is a background task (the SDK waits on span
+ingestion; minutes, not milliseconds) and 404s while
+``FEEDBACK_EVAL_SAMPLING_ENABLED`` is off, per the flag's docstring.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, Optional
+
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
+
+from apis.shared.auth import User, require_admin_scope
+from apis.shared.feature_flags import feedback_eval_sampling_enabled
+from apis.shared.storage.dynamodb_storage import DynamoDBStorage
+
+from .models import DownThumbQueueItem, DownThumbQueueResponse, SamplingRunResponse
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/feedback", tags=["admin-feedback"])
+require_feedback_admin = require_admin_scope("admin.costs")
+
+
+def get_storage() -> DynamoDBStorage:
+    return DynamoDBStorage()
+
+
+def get_judge():
+    """The AgentCore Evaluations adapter. A dependency so tests inject a fake."""
+    from apis.shared.feedback_eval.sampler import AgentCoreJudge
+
+    return AgentCoreJudge()
+
+
+@router.get("/evaluations", response_model=DownThumbQueueResponse, response_model_by_alias=True)
+async def list_down_thumb_queue(
+    limit: int = Query(50, ge=1, le=200),
+    current_user: User = Depends(require_feedback_admin),
+    storage: DynamoDBStorage = Depends(get_storage),
+):
+    """Recent down-thumbs across the fleet, newest first, with the verdict
+    where one exists. Content-free by projection."""
+    try:
+        rows = await storage.get_recent_down_thumbs(limit=limit)
+    except Exception:
+        logger.error("Error listing the down-thumb queue", exc_info=True)
+        raise HTTPException(status_code=500, detail="Failed to list feedback queue")
+    items = []
+    for row in rows:
+        try:
+            items.append(DownThumbQueueItem(**row))
+        except Exception:  # noqa: BLE001 - a malformed row is skipped, not fatal
+            continue
+    return DownThumbQueueResponse(
+        items=items,
+        pending=sum(1 for i in items if not i.evaluated_at),
+        sampling_enabled=feedback_eval_sampling_enabled(),
+    )
+
+
+@router.post("/evaluations/run", response_model=SamplingRunResponse, status_code=202, response_model_by_alias=True)
+async def run_eval_sampling(
+    background: BackgroundTasks,
+    limit: int = Query(10, ge=1, le=50),
+    current_user: User = Depends(require_feedback_admin),
+    storage: DynamoDBStorage = Depends(get_storage),
+    judge=Depends(get_judge),
+):
+    """Judge up to ``limit`` recent, not-yet-judged down-thumbs in the
+    background. 202 immediately; results appear on the queue list and the
+    session profiles as they land."""
+    if not feedback_eval_sampling_enabled():
+        raise HTTPException(status_code=404, detail="Not found")
+
+    def cost_row_lookup(session_id: str, message_id: int) -> Optional[Dict[str, Any]]:
+        # Sync lookup for tool-failure corroboration: the call's C# row.
+        try:
+            from boto3.dynamodb.conditions import Key
+
+            response = storage.sessions_metadata_table.query(
+                IndexName="SessionLookupIndex",
+                KeyConditionExpression=Key("GSI_PK").eq(f"SESSION#{session_id}") & Key("GSI_SK").begins_with("C#"),
+            )
+            for item in response.get("Items", []):
+                try:
+                    if int(item.get("messageId")) == message_id:
+                        return storage._convert_decimal_to_float(item)
+                except (TypeError, ValueError):
+                    continue
+        except Exception:  # noqa: BLE001 - corroboration is best-effort
+            return None
+        return None
+
+    async def task() -> None:
+        from apis.shared.feedback_eval.sampler import run_sampling_batch
+
+        try:
+            await run_sampling_batch(
+                storage.sessions_metadata_table, judge, limit=limit, cost_row_lookup=cost_row_lookup,
+            )
+        except Exception:  # noqa: BLE001 - background; nothing to return to
+            logger.error("eval sampling batch failed", exc_info=True)
+
+    background.add_task(task)
+    logger.info("Admin queued an eval sampling batch (limit=%d)", limit)
+    return SamplingRunResponse(
+        accepted=True,
+        limit=limit,
+        note="Judging runs in the background; the SDK waits for span ingestion, so allow a few minutes.",
+    )

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

@@ -923,6 +923,11 @@ async def get_managed_model_roles(
 
 router.include_router(costs_router)
 
+# ========== Include Feedback Eval-Sampling Subrouter ==========
+from .feedback.routes import router as feedback_admin_router
+
+router.include_router(feedback_admin_router)
+
 # ========== Include User Admin Subrouter ==========
 from .users.routes import router as users_router
 

backend/src/apis/shared/feature_flags.py

@@ -405,3 +405,24 @@ def attachment_turn_guard_enabled() -> bool:
     live in ``apis.shared.files.models``.
     """
     return os.environ.get("ATTACHMENT_TURN_GUARD_ENABLED", "").strip().lower() != "false"
+
+
+def feedback_eval_sampling_enabled() -> bool:
+    """Whether down-thumbed turns may be sent to AgentCore Evaluations.
+
+    Covers ``POST /admin/feedback/evaluations/run`` (the offline batch that
+    judges recent down-thumbs, response-feedback spec §11 PR-4). **Defaults
+    OFF** (the ``FINE_TUNING_ENABLED``-style opt-in): set
+    ``FEEDBACK_EVAL_SAMPLING_ENABLED=true`` to turn it on.
+
+    Off by default on purpose, not by caution: the judge is an AWS-managed
+    evaluator that reads the conversation's spans — the full system prompt
+    and every user message of the sampled session. The evaluations spike
+    (``docs/specs/agentcore-evaluations-spike-findings.md`` §2) says to make
+    that decision explicitly per environment rather than let it happen as a
+    side effect, and the feedback spec's §8 puts conversation content behind
+    a scope. Flipping this flag is that decision. The read surfaces (the
+    queue list, the profile's judged aggregates) are not gated — they show
+    numbers only and tolerate the absence of any judged row.
+    """
+    return os.environ.get("FEEDBACK_EVAL_SAMPLING_ENABLED", "false").strip().lower() == "true"

backend/src/apis/shared/feedback_eval/__init__.py

@@ -0,0 +1 @@
+"""Eval sampling: down-thumbed turns as the evaluation harness's input queue."""