diff --git a/backend/src/apis/app_api/admin/costs/models.py b/backend/src/apis/app_api/admin/costs/models.py index 980840464..38f57501b 100644 --- a/backend/src/apis/app_api/admin/costs/models.py +++ b/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): diff --git a/backend/src/apis/app_api/admin/costs/service.py b/backend/src/apis/app_api/admin/costs/service.py index 2fee25977..b53d3929d 100644 --- a/backend/src/apis/app_api/admin/costs/service.py +++ b/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 diff --git a/backend/src/apis/app_api/admin/feedback/__init__.py b/backend/src/apis/app_api/admin/feedback/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/src/apis/app_api/admin/feedback/models.py b/backend/src/apis/app_api/admin/feedback/models.py new file mode 100644 index 000000000..c711f2f9d --- /dev/null +++ b/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 diff --git a/backend/src/apis/app_api/admin/feedback/routes.py b/backend/src/apis/app_api/admin/feedback/routes.py new file mode 100644 index 000000000..7c0446436 --- /dev/null +++ b/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.", + ) diff --git a/backend/src/apis/app_api/admin/routes.py b/backend/src/apis/app_api/admin/routes.py index 2e4956931..8b3711f88 100644 --- a/backend/src/apis/app_api/admin/routes.py +++ b/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 diff --git a/backend/src/apis/shared/feature_flags.py b/backend/src/apis/shared/feature_flags.py index 41a1305c9..140cb0264 100644 --- a/backend/src/apis/shared/feature_flags.py +++ b/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" diff --git a/backend/src/apis/shared/feedback_eval/__init__.py b/backend/src/apis/shared/feedback_eval/__init__.py new file mode 100644 index 000000000..f170aa11b --- /dev/null +++ b/backend/src/apis/shared/feedback_eval/__init__.py @@ -0,0 +1 @@ +"""Eval sampling: down-thumbed turns as the evaluation harness's input queue.""" diff --git a/backend/src/apis/shared/feedback_eval/sampler.py b/backend/src/apis/shared/feedback_eval/sampler.py new file mode 100644 index 000000000..29fde6115 --- /dev/null +++ b/backend/src/apis/shared/feedback_eval/sampler.py @@ -0,0 +1,230 @@ +"""Eval sampling — response-feedback spec §11 PR-4. + +A down-thumb is a *sampler*: it marks the small subset of turns worth +spending judge tokens on. This module turns the queue of recent down-thumbs +into AgentCore Evaluations calls, routed by the thumb's reason code (spec +§6), and writes a **content-free** verdict back onto the thumb row. + +Three rules, from the spec and the evaluations spike: + +* **Offline batch only.** ``run_sampling_batch`` is driven by an admin + request (or a future schedule), never by a turn. An inline per-turn judge + is exactly what the cost tenet exists to stop. +* **Numbers leave, prose does not.** The Evaluate API returns an + ``explanation`` that quotes the conversation. ``summarize`` drops it; the + storage write refuses anything that still carries one; the content-policy + denylist names it. +* **The judge is a seam.** :class:`Judge` is a one-method protocol so tests + and forks without AgentCore Evaluations inject their own; + :class:`AgentCoreJudge` is the SDK adapter, imported lazily. + +Routing (spec §6 → the 16 built-in evaluators the spike verified): + + wrong → Builtin.Correctness, Builtin.Faithfulness + instructions → Builtin.InstructionFollowing + length → Builtin.Conciseness (the style signal) + tool_failed → no judge — ops, corroborated against the call's tool census + outdated → no judge yet — KB-freshness join is a follow-up + other / none → Builtin.Helpfulness (the generic judge) +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass, field +from datetime import timedelta +from typing import Any, Dict, List, Optional, Protocol, Sequence + +logger = logging.getLogger(__name__) + +RUNTIME_LOG_GROUP_ENV = "AGENTCORE_RUNTIME_LOG_GROUP" + +#: Reason code → built-in evaluator ids. An empty tuple means "no judge for +#: this bucket" (it is an ops signal, or not yet joinable), not "skip". +EVALUATORS_BY_REASON: Dict[Optional[str], tuple] = { + "wrong": ("Builtin.Correctness", "Builtin.Faithfulness"), + "instructions": ("Builtin.InstructionFollowing",), + "length": ("Builtin.Conciseness",), + "tool_failed": (), + "outdated": (), + "other": ("Builtin.Helpfulness",), + None: ("Builtin.Helpfulness",), +} + + +def evaluators_for(reason: Optional[str]) -> tuple: + """The evaluator ids a reason bucket routes to (unknown codes → generic).""" + return EVALUATORS_BY_REASON.get(reason, EVALUATORS_BY_REASON[None]) + + +class Judge(Protocol): + """Anything that can score one conversation with a set of evaluators.""" + + def judge(self, session_id: str, evaluator_ids: Sequence[str]) -> List[Dict[str, Any]]: + """Raw ``evaluationResults`` items for the session, or ``[]``.""" + + +class AgentCoreJudge: + """``bedrock_agentcore.evaluation.EvaluationClient`` over the runtime log + group. Lazy imports keep ``apis.shared`` importable in images without the + SDK. The session id sent is the *runtime* session id + (``sid-``), which is how the chat proxy pins spans.""" + + def __init__(self, log_group_name: Optional[str] = None, region_name: Optional[str] = None, + look_back: timedelta = timedelta(days=7)): + self.log_group_name = log_group_name or os.environ.get(RUNTIME_LOG_GROUP_ENV, "").strip() + self.region_name = region_name or os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION") + self.look_back = look_back + self._client = None + + def judge(self, session_id: str, evaluator_ids: Sequence[str]) -> List[Dict[str, Any]]: + if not self.log_group_name: + raise RuntimeError(f"{RUNTIME_LOG_GROUP_ENV} is not configured") + if self._client is None: + from bedrock_agentcore.evaluation import EvaluationClient # lazy: heavy, optional + + self._client = EvaluationClient(region_name=self.region_name) + from apis.shared.harness.runner import runtime_session_id_for + + return self._client.run( + evaluator_ids=list(evaluator_ids), + session_id=runtime_session_id_for(session_id), + log_group_name=self.log_group_name, + look_back_time=self.look_back, + ) + + +def summarize(results: Sequence[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: + """Collapse raw ``evaluationResults`` into ``{evaluatorId: {value, rating, + n, tokens}}`` — numbers and the evaluator's own rating vocabulary (its + ``label``, stored as ``rating`` because ``label`` is a content-bearing + path elsewhere in the table and the denylist is by name). The + ``explanation`` (prose quoting the conversation) and the span context + are dropped here and nowhere later. TRACE-level evaluators return one + result per trace; ``value`` is their mean and ``n`` says how many.""" + out: Dict[str, Dict[str, Any]] = {} + for item in results: + evaluator = item.get("evaluatorId") or item.get("evaluatorName") + if not evaluator or item.get("errorCode"): + continue + value = item.get("value") + if not isinstance(value, (int, float)): + continue + slot = out.setdefault(evaluator, {"value": 0.0, "n": 0, "tokens": 0}) + n = slot["n"] + slot["value"] = round((slot["value"] * n + float(value)) / (n + 1), 4) + slot["n"] = n + 1 + slot["tokens"] += int((item.get("tokenUsage") or {}).get("totalTokens") or 0) + rating = item.get("label") + if isinstance(rating, str) and rating: + slot["rating"] = rating + return out + + +def corroborate_tool_failure(cost_row: Optional[Dict[str, Any]]) -> Optional[bool]: + """For a ``tool_failed`` thumb: did the call's tool census (``toolCalls`` + on the ``C#`` row) actually record an error? ``None`` when the row or the + census is missing — "could not check", not "no".""" + if not cost_row: + return None + census = cost_row.get("toolCalls") + if not isinstance(census, dict): + return None + for entry in census.values(): + errors = entry.get("errors") if isinstance(entry, dict) else None + try: + if int(errors or 0) > 0: + return True + except (TypeError, ValueError): + continue + return False + + +@dataclass +class SamplingReport: + """What one batch did — counts only, for the log line and tests.""" + candidates: int = 0 + judged: int = 0 + corroborated: int = 0 + skipped_no_judge: int = 0 + skipped_already: int = 0 + failed: int = 0 + tokens: int = 0 + errors: List[str] = field(default_factory=list) + + +def build_verdict( + reason: Optional[str], + results: Sequence[Dict[str, Any]], + cost_row: Optional[Dict[str, Any]], +) -> Dict[str, Any]: + """The content-free ``evaluation`` map for one thumb.""" + verdict: Dict[str, Any] = {"reason": reason or "none", "evaluators": list(evaluators_for(reason))} + if results: + verdict["scores"] = summarize(results) + if reason == "tool_failed": + corroborated = corroborate_tool_failure(cost_row) + if corroborated is not None: + verdict["toolFailureCorroborated"] = corroborated + return verdict + + +async def run_sampling_batch( + table, + judge: Judge, + *, + limit: int = 10, + cost_row_lookup=None, +) -> SamplingReport: + """Judge up to ``limit`` recent, not-yet-judged down-thumbs. + + ``cost_row_lookup(session_id, message_id) -> Optional[dict]`` supplies the + call's ``C#`` row for tool-failure corroboration; optional. Every thumb is + independent: one judge failure is recorded and the batch continues. + """ + from apis.shared.sessions.feedback import is_explicit, list_recent_down_thumbs, store_evaluation + + report = SamplingReport() + rows = [r for r in list_recent_down_thumbs(table, limit=limit * 3) if is_explicit(r)] + for row in rows: + if report.judged + report.skipped_no_judge + report.failed >= limit: + break + if row.get("evaluatedAt"): + report.skipped_already += 1 + continue + report.candidates += 1 + session_id = str(row.get("sessionId") or "") + user_id = str(row.get("userId") or "") + try: + message_id = int(row.get("messageId")) + except (TypeError, ValueError): + report.failed += 1 + continue + reason = row.get("reason") if isinstance(row.get("reason"), str) else None + evaluator_ids = evaluators_for(reason) + cost_row = cost_row_lookup(session_id, message_id) if cost_row_lookup else None + results: List[Dict[str, Any]] = [] + try: + if evaluator_ids: + results = judge.judge(session_id, evaluator_ids) + verdict = build_verdict(reason, results, cost_row) + store_evaluation(table, user_id, session_id, message_id, verdict) + except Exception as e: # noqa: BLE001 - one bad thumb must not stop the batch + report.failed += 1 + report.errors.append(type(e).__name__) + logger.warning("eval sampling failed for a thumb: %s", type(e).__name__) + continue + if evaluator_ids: + report.judged += 1 + report.tokens += sum(s.get("tokens", 0) for s in verdict.get("scores", {}).values()) + else: + report.skipped_no_judge += 1 + if verdict.get("toolFailureCorroborated"): + report.corroborated += 1 + logger.info( + "eval sampling batch: candidates=%d judged=%d no_judge=%d already=%d failed=%d tokens=%d", + report.candidates, report.judged, report.skipped_no_judge, report.skipped_already, + report.failed, report.tokens, + ) + return report diff --git a/backend/src/apis/shared/observability/content_policy.py b/backend/src/apis/shared/observability/content_policy.py index eb0ccde61..e1f016158 100644 --- a/backend/src/apis/shared/observability/content_policy.py +++ b/backend/src/apis/shared/observability/content_policy.py @@ -60,6 +60,8 @@ "stateReason", "lastError", "errorDetail", + # F# feedback rows, judged: the evaluator's prose quotes the conversation + "explanation", # FILE# upload rows "filename", # user-chosen "s3Key", # embeds the filename @@ -213,6 +215,8 @@ def is_content_bearing(path: str) -> bool: "reason", "signal", "retryMessageId", # a message index, the retry-with-correction link + "evaluation", # judged verdict: per-evaluator value/label/n/tokens, never the explanation + "evaluatedAt", "updatedAt", ) diff --git a/backend/src/apis/shared/sessions/feedback.py b/backend/src/apis/shared/sessions/feedback.py index 15ed464e2..16d4f383a 100644 --- a/backend/src/apis/shared/sessions/feedback.py +++ b/backend/src/apis/shared/sessions/feedback.py @@ -22,6 +22,13 @@ this same row family under ``signal: "implicit"``; every reader here filters to explicit rows so that phase needs no backfill and the two are never summed. +Thumb rows also carry ``GSI1PK = FEEDBACK#down`` / ``FEEDBACK#up`` and +``GSI1SK = updatedAt`` on the existing ``UserTimestampIndex``, so "recent +down-thumbs across the fleet" is one query with no new index — the eval +sampler's input queue (spec §11 PR-4). A judged row gains ``evaluation`` +(content-free: per-evaluator value / label / tokens, never the judge's +explanation) and ``evaluatedAt``. + ``retryMessageId`` is the index of the user message the SPA sent as a *retry with correction* after a down-thumb (response-feedback spec §7 "the retry loop", §11 PR-1's consequence). It is a link, never the correction's @@ -51,7 +58,7 @@ import logging import os from datetime import datetime, timedelta, timezone -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional from .models import FEEDBACK_REASONS, MessageFeedback from .preview import is_preview_session @@ -165,6 +172,9 @@ async def put_message_feedback( sets = { "GSI_PK": f"SESSION#{session_id}", "GSI_SK": f"F#{message_id}", + # Fleet queue on UserTimestampIndex: recent down-thumbs in one query. + "GSI1PK": f"FEEDBACK#{'down' if value == -1 else 'up'}", + "GSI1SK": now, "sessionId": session_id, "messageId": int(message_id), "userId": user_id, @@ -235,6 +245,53 @@ async def delete_message_feedback(session_id: str, user_id: str, message_id: int return True +def list_recent_down_thumbs(table, limit: int = 20) -> List[Dict[str, Any]]: + """Newest-first down-thumb rows across the fleet (``UserTimestampIndex``, + partition ``FEEDBACK#down``). Raw rows — the caller is the sampler, which + needs ``userId`` to write the verdict back; the admin list goes through + the projected storage reader instead.""" + from boto3.dynamodb.conditions import Key + + response = table.query( + IndexName="UserTimestampIndex", + KeyConditionExpression=Key("GSI1PK").eq("FEEDBACK#down"), + ScanIndexForward=False, + Limit=max(1, int(limit)), + ) + return list(response.get("Items", [])) + + +def store_evaluation( + table, + user_id: str, + session_id: str, + message_id: int, + evaluation: Dict[str, Any], +) -> None: + """Attach a content-free judged result to a thumb row. ``evaluation`` is + the sampler's summary (``apis.shared.feedback_eval.sampler.build_verdict``); + this function refuses anything carrying an ``explanation`` so the judge's + prose about the conversation can never land beside the cost row.""" + if _carries_explanation(evaluation): + raise ValueError("evaluation summaries must not carry the judge's explanation") + from .metadata import _convert_floats_to_decimal + + table.update_item( + Key=_keys(user_id, session_id, message_id), + UpdateExpression="SET evaluation = :e, evaluatedAt = :t", + ExpressionAttributeValues={":e": _convert_floats_to_decimal(evaluation), ":t": _now()}, + ConditionExpression="attribute_exists(PK)", + ) + + +def _carries_explanation(obj: Any) -> bool: + if isinstance(obj, dict): + return any(k == "explanation" or _carries_explanation(v) for k, v in obj.items()) + if isinstance(obj, list): + return any(_carries_explanation(v) for v in obj) + return False + + def query_session_feedback(table, session_id: str, user_id: Optional[str] = None) -> Dict[str, MessageFeedback]: """All ``F#`` rows for a session via ``SessionLookupIndex``, keyed by message id (as ``str``, matching the metadata index). With ``user_id`` diff --git a/backend/src/apis/shared/storage/dynamodb_storage.py b/backend/src/apis/shared/storage/dynamodb_storage.py index 6b03ba1ff..cb26945ed 100644 --- a/backend/src/apis/shared/storage/dynamodb_storage.py +++ b/backend/src/apis/shared/storage/dynamodb_storage.py @@ -380,6 +380,30 @@ async def get_session_feedback_rows( return [strip_content(self._convert_decimal_to_float(item)) for item in items] + async def get_recent_down_thumbs(self, limit: int = 50) -> List[Dict[str, Any]]: + """Newest-first down-thumb rows across the fleet — the eval sampler's + queue as an admin sees it: content-free by projection, no user id.""" + from boto3.dynamodb.conditions import Key + from apis.shared.observability.content_policy import ( + FEEDBACK_ROW_PROJECTION, + build_projection, + strip_content, + ) + + projection, names = build_projection(FEEDBACK_ROW_PROJECTION) + try: + response = self.sessions_metadata_table.query( + IndexName="UserTimestampIndex", + KeyConditionExpression=Key("GSI1PK").eq("FEEDBACK#down"), + ScanIndexForward=False, + Limit=max(1, min(int(limit), 200)), + ProjectionExpression=projection, + ExpressionAttributeNames=names, + ) + except ClientError as e: + raise Exception(f"Failed to list recent down-thumbs: {e}") + return [strip_content(self._convert_decimal_to_float(item)) for item in response.get("Items", [])] + async def get_session_diagnostic_row( self, session_id: str, diff --git a/backend/tests/apis/app_api/admin/costs/test_drilldown_routes.py b/backend/tests/apis/app_api/admin/costs/test_drilldown_routes.py index 74cc883a4..58dd5ac38 100644 --- a/backend/tests/apis/app_api/admin/costs/test_drilldown_routes.py +++ b/backend/tests/apis/app_api/admin/costs/test_drilldown_routes.py @@ -108,7 +108,7 @@ def test_session_profile_returns_200(): "feedback": False, "documents": False, } - assert body["feedback"] == {"up": 0, "down": 0, "byTurnClass": None, "unjoined": 0, "retried": 0, "reworkUsd": None} + assert body["feedback"] == {"up": 0, "down": 0, "byTurnClass": None, "unjoined": 0, "retried": 0, "reworkUsd": None, "evaluations": None} service.get_session_profile.assert_awaited_once_with("s1") diff --git a/backend/tests/apis/app_api/admin/costs/test_session_profile_service.py b/backend/tests/apis/app_api/admin/costs/test_session_profile_service.py index 5a5b75be9..98f494d84 100644 --- a/backend/tests/apis/app_api/admin/costs/test_session_profile_service.py +++ b/backend/tests/apis/app_api/admin/costs/test_session_profile_service.py @@ -372,3 +372,36 @@ async def test_feedback_reader_failure_never_breaks_the_profile(): service.storage.get_session_feedback_rows = AsyncMock(side_effect=RuntimeError("boom")) p = await service.get_session_profile("s1") assert p is not None and p.feedback.up == 0 and p.data_coverage.feedback is False + + +@pytest.mark.asyncio +async def test_judged_thumbs_aggregate_per_evaluator_and_corroboration(): + def judged(message_id, reason, scores=None, corroborated=None): + row = _feedback(message_id, -1, reason) + row["evaluatedAt"] = "t" + row["evaluation"] = {"reason": reason, "evaluators": list((scores or {}).keys())} + if scores: + row["evaluation"]["scores"] = {k: {"value": v, "n": 1} for k, v in scores.items()} + if corroborated is not None: + row["evaluation"]["toolFailureCorroborated"] = corroborated + return row + + feedback = [ + judged(0, "wrong", {"Builtin.Correctness": 1.0, "Builtin.Faithfulness": 0.5}), + judged(1, "wrong", {"Builtin.Correctness": 0.0}), + judged(2, "tool_failed", corroborated=True), + judged(3, "tool_failed", corroborated=False), + _feedback(4, -1, "other"), # not judged yet + ] + p = await _service_with_feedback(_row(), [_call(i) for i in range(5)], feedback).get_session_profile("s1") + ev = p.feedback.evaluations + assert ev is not None and ev.judged == 4 + assert ev.by_evaluator["Builtin.Correctness"].n == 2 and ev.by_evaluator["Builtin.Correctness"].mean == 0.5 + assert ev.by_evaluator["Builtin.Faithfulness"].mean == 0.5 + assert (ev.tool_failures_reported, ev.tool_failures_corroborated) == (2, 1) + assert p.feedback.down == 5 + wire = p.model_dump(by_alias=True)["feedback"]["evaluations"] + assert wire["byEvaluator"]["Builtin.Correctness"] == {"n": 2, "mean": 0.5} + + p = await _service_with_feedback(_row(), [_call(0)], [_feedback(0, -1)]).get_session_profile("s1") + assert p.feedback.evaluations is None diff --git a/backend/tests/apis/app_api/admin/test_feedback_eval_routes.py b/backend/tests/apis/app_api/admin/test_feedback_eval_routes.py new file mode 100644 index 000000000..6f453d94a --- /dev/null +++ b/backend/tests/apis/app_api/admin/test_feedback_eval_routes.py @@ -0,0 +1,76 @@ +"""Admin eval-sampling routes (spec §11 PR-4): + +- GET /admin/feedback/evaluations → 200, content-free queue with verdicts +- POST /admin/feedback/evaluations/run → 202 and a background batch with the injected judge; + 404 while FEEDBACK_EVAL_SAMPLING_ENABLED is off +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from apis.app_api.admin.feedback import routes as feedback_routes +from apis.shared.observability.content_policy import content_bearing_paths +from tests.conftest import override_admin_auth + + +def _app(storage, judge=None): + app = FastAPI() + app.include_router(feedback_routes.router) + override_admin_auth(app, lambda: SimpleNamespace(user_id="admin", email="a@x", roles=["system_admin"])) + app.dependency_overrides[feedback_routes.get_storage] = lambda: storage + app.dependency_overrides[feedback_routes.get_judge] = lambda: judge or SimpleNamespace(judge=lambda *a, **k: []) + return app + + +def test_queue_lists_recent_down_thumbs_with_verdicts(monkeypatch): + monkeypatch.delenv("FEEDBACK_EVAL_SAMPLING_ENABLED", raising=False) + storage = SimpleNamespace(get_recent_down_thumbs=AsyncMock(return_value=[ + {"sessionId": "s1", "messageId": 3, "reason": "wrong", "updatedAt": "t2", "evaluatedAt": "t3", + "evaluation": {"reason": "wrong", "evaluators": ["Builtin.Correctness"], + "scores": {"Builtin.Correctness": {"value": 0.5, "n": 2, "tokens": 100, "rating": "Meh"}}}}, + {"sessionId": "s2", "messageId": 1, "updatedAt": "t1"}, + {"garbage": True}, + ])) + resp = TestClient(_app(storage)).get("/feedback/evaluations", params={"limit": 5}) + assert resp.status_code == 200 + body = resp.json() + assert body["samplingEnabled"] is False and body["pending"] == 1 + assert [i["sessionId"] for i in body["items"]] == ["s1", "s2"] + assert body["items"][0]["evaluation"]["scores"]["Builtin.Correctness"]["rating"] == "Meh" + assert content_bearing_paths(body) == [] + storage.get_recent_down_thumbs.assert_awaited_once_with(limit=5) + + +def test_run_is_404_while_off_and_202_with_a_batch_when_on(monkeypatch): + calls = {} + + async def fake_batch(table, judge, *, limit, cost_row_lookup=None): + calls["limit"] = limit + calls["judge"] = judge + return SimpleNamespace() + + monkeypatch.setattr("apis.shared.feedback_eval.sampler.run_sampling_batch", fake_batch) + storage = SimpleNamespace(sessions_metadata_table=object(), get_recent_down_thumbs=AsyncMock(return_value=[])) + judge = SimpleNamespace(judge=lambda *a, **k: []) + + monkeypatch.setenv("FEEDBACK_EVAL_SAMPLING_ENABLED", "false") + assert TestClient(_app(storage, judge)).post("/feedback/evaluations/run").status_code == 404 + assert calls == {} + + monkeypatch.setenv("FEEDBACK_EVAL_SAMPLING_ENABLED", "true") + resp = TestClient(_app(storage, judge)).post("/feedback/evaluations/run", params={"limit": 7}) + assert resp.status_code == 202 + assert resp.json()["accepted"] is True and resp.json()["limit"] == 7 + # TestClient runs background tasks before returning. + assert calls["limit"] == 7 and calls["judge"] is judge + + +def test_run_limit_is_bounded(monkeypatch): + monkeypatch.setenv("FEEDBACK_EVAL_SAMPLING_ENABLED", "true") + storage = SimpleNamespace(sessions_metadata_table=object()) + assert TestClient(_app(storage)).post("/feedback/evaluations/run", params={"limit": 500}).status_code == 422 diff --git a/backend/tests/architecture/test_admin_scope_coverage.py b/backend/tests/architecture/test_admin_scope_coverage.py index b93e4a1af..eacc7e57e 100644 --- a/backend/tests/architecture/test_admin_scope_coverage.py +++ b/backend/tests/architecture/test_admin_scope_coverage.py @@ -44,6 +44,7 @@ "routes.py": "admin.models", "quota/routes.py": "admin.quota", "costs/routes.py": "admin.costs", + "feedback/routes.py": "admin.costs", # eval-sampling queue sits beside the cost rows "users/routes.py": "admin.users", "tools/routes.py": "admin.tools", "skills/routes.py": "admin.skills", diff --git a/backend/tests/costs/test_content_free_projections.py b/backend/tests/costs/test_content_free_projections.py index ae86d1389..fc5d60303 100644 --- a/backend/tests/costs/test_content_free_projections.py +++ b/backend/tests/costs/test_content_free_projections.py @@ -201,3 +201,23 @@ async def test_feedback_rows_come_back_content_free_and_keyed_to_the_call(storag records = await storage.get_session_cost_records(SESSION_ID) assert records[0]["messageId"] == row["messageId"] assert await storage.get_session_feedback_rows("no-such-session") == [] + + +@pytest.mark.asyncio +async def test_recent_down_thumbs_queue_is_content_free_and_newest_first(storage): + _seed(storage) + for i, ts in ((1, "2026-09-16T00:00:01Z"), (2, "2026-09-16T00:00:02Z")): + storage.sessions_metadata_table.put_item(Item={ + "PK": f"USER#{USER_ID}", "SK": f"F#{SESSION_ID}#{i}", + "GSI_PK": f"SESSION#{SESSION_ID}", "GSI_SK": f"F#{i}", + "GSI1PK": "FEEDBACK#down", "GSI1SK": ts, + "sessionId": SESSION_ID, "messageId": Decimal(i), "userId": USER_ID, + "value": Decimal(-1), "reason": "wrong", "signal": "explicit", "updatedAt": ts, + "evaluation": {"reason": "wrong", "scores": {"Builtin.Correctness": {"value": Decimal("0.5"), "n": Decimal(1), "explanation": "SECRET"}}}, + "displayText": "SECRET", + }) + rows = await storage.get_recent_down_thumbs(limit=10) + assert [r["messageId"] for r in rows] == [2, 1] + assert all(content_bearing_paths(r) == [] for r in rows) + assert "userId" not in rows[0] and "displayText" not in rows[0] + assert rows[0]["evaluation"]["scores"]["Builtin.Correctness"] == {"value": 0.5, "n": 1} diff --git a/backend/tests/shared/test_feedback_eval_sampler.py b/backend/tests/shared/test_feedback_eval_sampler.py new file mode 100644 index 000000000..2b06d0ce6 --- /dev/null +++ b/backend/tests/shared/test_feedback_eval_sampler.py @@ -0,0 +1,193 @@ +"""Eval sampling (response-feedback spec §11 PR-4): routing by reason, the +content-free verdict, tool-failure corroboration, and the batch over a real +(moto) table with a fake judge — the AgentCore adapter is a seam, not a +dependency of these tests. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Sequence + +import boto3 +import pytest +from boto3.dynamodb.conditions import Key + +from apis.shared.feedback_eval import sampler +from apis.shared.sessions import feedback as fb + +OWNER = "user-owner" + + +def _table(): + return boto3.resource("dynamodb", region_name="us-east-1").Table("test-sessions-metadata") + + +def _seed_session(session_id, user_id=OWNER): + _table().put_item(Item={ + "PK": f"USER#{user_id}", "SK": f"S#{session_id}", "GSI_PK": f"SESSION#{session_id}", "GSI_SK": "META", + "sessionId": session_id, "userId": user_id, "status": "active", + }) + + +def _row(session_id, message_id, user_id=OWNER): + return _table().get_item(Key={"PK": f"USER#{user_id}", "SK": f"F#{session_id}#{message_id}"}).get("Item") + + +class FakeJudge: + def __init__(self, results: List[Dict[str, Any]] | None = None, fail_for: str | None = None): + self.results = results or [] + self.calls: List[tuple] = [] + self.fail_for = fail_for + + def judge(self, session_id: str, evaluator_ids: Sequence[str]) -> List[Dict[str, Any]]: + self.calls.append((session_id, tuple(evaluator_ids))) + if session_id == self.fail_for: + raise RuntimeError("judge exploded") + return self.results + + +RAW = [ + {"evaluatorId": "Builtin.Correctness", "value": 1.0, "label": "Perfectly Correct", + "explanation": "The agent correctly reported that SECRET tool call 502'd", "tokenUsage": {"totalTokens": 1728}, + "context": {"spanContext": {"sessionId": "sid-x", "traceId": "t1"}}}, + {"evaluatorId": "Builtin.Correctness", "value": 0.0, "label": "Incorrect", "explanation": "SECRET", "tokenUsage": {"totalTokens": 900}}, + {"evaluatorId": "Builtin.Faithfulness", "value": 0.5, "label": "Partly", "explanation": "SECRET"}, + {"evaluatorId": "Builtin.Faithfulness", "errorCode": "ThrottlingException", "errorMessage": "slow down"}, +] + + +def test_routing_follows_the_spec_buckets(): + assert sampler.evaluators_for("wrong") == ("Builtin.Correctness", "Builtin.Faithfulness") + assert sampler.evaluators_for("instructions") == ("Builtin.InstructionFollowing",) + assert sampler.evaluators_for("length") == ("Builtin.Conciseness",) + assert sampler.evaluators_for("tool_failed") == () + assert sampler.evaluators_for("outdated") == () + assert sampler.evaluators_for("other") == ("Builtin.Helpfulness",) + assert sampler.evaluators_for(None) == ("Builtin.Helpfulness",) + assert sampler.evaluators_for("not-a-code") == ("Builtin.Helpfulness",) + + +def test_summarize_keeps_numbers_and_labels_and_drops_the_explanation(): + scores = sampler.summarize(RAW) + assert scores == { + "Builtin.Correctness": {"value": 0.5, "n": 2, "tokens": 2628, "rating": "Incorrect"}, + "Builtin.Faithfulness": {"value": 0.5, "n": 1, "tokens": 0, "rating": "Partly"}, + } + assert "SECRET" not in repr(scores) and "explanation" not in repr(scores) + + +def test_corroboration_reads_the_call_census(): + assert sampler.corroborate_tool_failure(None) is None + assert sampler.corroborate_tool_failure({"cost": {"total": 1}}) is None + assert sampler.corroborate_tool_failure({"toolCalls": {"web_search": {"calls": 2, "errors": 0}}}) is False + assert sampler.corroborate_tool_failure({"toolCalls": {"web_search": {"calls": 2, "errors": 1}}}) is True + + +def test_build_verdict_shapes(): + v = sampler.build_verdict("wrong", RAW, None) + assert v["reason"] == "wrong" and v["evaluators"] == ["Builtin.Correctness", "Builtin.Faithfulness"] + assert set(v["scores"]) == {"Builtin.Correctness", "Builtin.Faithfulness"} + assert "toolFailureCorroborated" not in v + v = sampler.build_verdict("tool_failed", [], {"toolCalls": {"x": {"calls": 1, "errors": 1}}}) + assert v == {"reason": "tool_failed", "evaluators": [], "toolFailureCorroborated": True} + assert sampler.build_verdict(None, [], None) == {"reason": "none", "evaluators": ["Builtin.Helpfulness"]} + + +@pytest.fixture() +def table(sessions_metadata_table, monkeypatch): + monkeypatch.delenv("COST_DIAGNOSTICS_ENABLED", raising=False) + return _table() + + +@pytest.mark.asyncio +async def test_down_thumbs_are_queued_on_the_timestamp_index_and_up_thumbs_leave_it(table): + _seed_session("s1") + await fb.put_message_feedback("s1", OWNER, 1, -1, reason="wrong") + await fb.put_message_feedback("s1", OWNER, 3, -1) + await fb.put_message_feedback("s1", OWNER, 5, 1) + queue = fb.list_recent_down_thumbs(table, limit=10) + assert sorted(int(r["messageId"]) for r in queue) == [1, 3] + assert all(r["GSI1PK"] == "FEEDBACK#down" and r["GSI1SK"] == r["updatedAt"] for r in queue) + # Flipping to up moves the row out of the down queue. + await fb.put_message_feedback("s1", OWNER, 1, 1) + assert [int(r["messageId"]) for r in fb.list_recent_down_thumbs(table, limit=10)] == [3] + + +@pytest.mark.asyncio +async def test_store_evaluation_refuses_prose_and_requires_the_row(table): + _seed_session("s1") + await fb.put_message_feedback("s1", OWNER, 1, -1, reason="wrong") + with pytest.raises(ValueError): + fb.store_evaluation(table, OWNER, "s1", 1, {"scores": {"x": {"value": 1, "explanation": "SECRET"}}}) + fb.store_evaluation(table, OWNER, "s1", 1, {"reason": "wrong", "scores": {"Builtin.Correctness": {"value": 1, "n": 1}}}) + row = _row("s1", 1) + assert row["evaluatedAt"] and row["evaluation"]["reason"] == "wrong" + from botocore.exceptions import ClientError + with pytest.raises(ClientError): + fb.store_evaluation(table, OWNER, "s1", 99, {"reason": "wrong"}) + + +@pytest.mark.asyncio +async def test_batch_judges_by_reason_skips_judged_and_survives_one_failure(table): + for sid in ("s-wrong", "s-tool", "s-boom", "s-done"): + _seed_session(sid) + await fb.put_message_feedback("s-wrong", OWNER, 1, -1, reason="wrong") + await fb.put_message_feedback("s-tool", OWNER, 2, -1, reason="tool_failed") + await fb.put_message_feedback("s-boom", OWNER, 3, -1, reason="instructions") + await fb.put_message_feedback("s-done", OWNER, 4, -1, reason="length") + fb.store_evaluation(table, OWNER, "s-done", 4, {"reason": "length"}) + judge = FakeJudge(results=RAW, fail_for="s-boom") + + def lookup(session_id, message_id): + return {"toolCalls": {"web_search": {"calls": 1, "errors": 1}}} if session_id == "s-tool" else None + + report = await sampler.run_sampling_batch(table, judge, limit=10, cost_row_lookup=lookup) + + assert (report.candidates, report.judged, report.skipped_no_judge, report.skipped_already, report.failed) == (3, 1, 1, 1, 1) + assert report.corroborated == 1 and report.errors == ["RuntimeError"] + assert report.tokens == 2628 + assert sorted(judge.calls) == [("s-boom", ("Builtin.InstructionFollowing",)), ("s-wrong", ("Builtin.Correctness", "Builtin.Faithfulness"))] + + wrong = _row("s-wrong", 1) + assert wrong["evaluation"]["scores"]["Builtin.Correctness"]["rating"] == "Incorrect" + assert "SECRET" not in repr(wrong) + tool = _row("s-tool", 2) + assert tool["evaluation"] == {"reason": "tool_failed", "evaluators": [], "toolFailureCorroborated": True} + assert "evaluation" not in _row("s-boom", 3) + # The judged row was left alone (its evaluatedAt is the earlier one). + assert _row("s-done", 4)["evaluation"] == {"reason": "length"} + + # A second batch finds nothing new to judge except the failed one. + report = await sampler.run_sampling_batch(table, FakeJudge(results=RAW), limit=10) + assert (report.candidates, report.judged, report.skipped_already) == (1, 1, 3) + + +def test_agentcore_judge_requires_the_log_group(monkeypatch): + monkeypatch.delenv(sampler.RUNTIME_LOG_GROUP_ENV, raising=False) + with pytest.raises(RuntimeError): + sampler.AgentCoreJudge().judge("s1", ["Builtin.Helpfulness"]) + + +def test_agentcore_judge_sends_the_runtime_session_id(monkeypatch): + calls = {} + + class FakeClient: + def __init__(self, region_name=None): + calls["region"] = region_name + + def run(self, **kwargs): + calls.update(kwargs) + return [{"evaluatorId": "Builtin.Helpfulness", "value": 1.0}] + + import types, sys + fake_mod = types.ModuleType("bedrock_agentcore.evaluation") + fake_mod.EvaluationClient = FakeClient + monkeypatch.setitem(sys.modules, "bedrock_agentcore.evaluation", fake_mod) + from apis.shared.harness.runner import runtime_session_id_for + + judge = sampler.AgentCoreJudge(log_group_name="/aws/bedrock-agentcore/runtimes/rt-DEFAULT", region_name="us-west-2") + out = judge.judge("sess-1", ["Builtin.Helpfulness"]) + assert out[0]["value"] == 1.0 + assert calls["session_id"] == runtime_session_id_for("sess-1") + assert calls["log_group_name"] == "/aws/bedrock-agentcore/runtimes/rt-DEFAULT" and calls["region"] == "us-west-2" + assert calls["evaluator_ids"] == ["Builtin.Helpfulness"] diff --git a/backend/tests/shared/test_message_feedback.py b/backend/tests/shared/test_message_feedback.py index aae508dd5..815f883b9 100644 --- a/backend/tests/shared/test_message_feedback.py +++ b/backend/tests/shared/test_message_feedback.py @@ -73,7 +73,7 @@ async def test_put_writes_one_row_keyed_beside_the_cost_row(table): assert "ttl" in row # Content-free: nothing on the row but ids, a number, a timestamp and keys. assert row["signal"] == "explicit" - assert set(row) <= {"PK", "SK", "GSI_PK", "GSI_SK", "sessionId", "messageId", "userId", "value", "signal", "retryMessageId", "updatedAt", "ttl"} + assert set(row) <= {"PK", "SK", "GSI_PK", "GSI_SK", "GSI1PK", "GSI1SK", "sessionId", "messageId", "userId", "value", "signal", "retryMessageId", "updatedAt", "ttl"} @pytest.mark.asyncio diff --git a/docs/specs/response-feedback.md b/docs/specs/response-feedback.md index 8c1ab1169..e9c9fd692 100644 --- a/docs/specs/response-feedback.md +++ b/docs/specs/response-feedback.md @@ -3,8 +3,9 @@ **Status:** PARTIALLY BUILT — capture and the read model shipped in PR #1142 (2026-09-16, as document-context-offload PR-7) and the consequence (retry-with-correction) in the PR stacked on it; §11 PR-1 is therefore -complete. Implicit signals and eval sampling are not built. See §13. Written -2026-09-04 from the "how would we benefit?" conversation. +complete. Eval sampling (§11 PR-4) followed, opt-in per environment; implicit +signals (PR-2) are in review. See §13. Written 2026-09-04 from the "how would +we benefit?" conversation. **Refs:** `docs/specs/agentcore-evaluations-spike-findings.md` (the eval harness this feeds), `docs/specs/mid-turn-steering.md` (the injection path Phase 1 reuses), `docs/specs/agent-marketplace.md` D15 (the *other* feedback @@ -368,7 +369,32 @@ points: so Phase 6 can find it once the consent decision is made. The profile reports `feedback.retried` and `reworkUsd` (§7 "rework cost": the thumbed call rows plus the retry turn's consecutive assistant rows). -- **Not built**: implicit signals (PR-2), eval sampling (PR-4), - author/marketplace surfaces (PR-5), the report-dialog escape hatch in the - reason row. +- **Eval sampling (§11 PR-4), fourth PR — opt-in.** Down-thumbs carry + `GSI1PK = FEEDBACK#down` / `GSI1SK = updatedAt` on the existing + `UserTimestampIndex`, so the fleet's recent down-thumbs are one query with + no new index. `POST /admin/feedback/evaluations/run` (scope `admin.costs`) + judges up to N not-yet-judged ones in a background task through + `bedrock_agentcore.evaluation.EvaluationClient` over the runtime log + group, keyed by the runtime session id the chat proxy already pins; + `GET /admin/feedback/evaluations` is the queue with verdicts. Routing is + §6's table: `wrong` → Correctness + Faithfulness, `instructions` → + InstructionFollowing, `length` → Conciseness, `other`/none → Helpfulness; + `tool_failed` gets **no judge** and is corroborated against the call's + tool census on the `C#` row (ops, not model); `outdated` gets no judge yet + (the KB-freshness join is a follow-up). The verdict stored on the `F#` row + is per-evaluator value / rating / n / tokens — **the judge's `explanation` + is dropped at summarisation, refused at the storage write, and denylisted + in the content policy**, because it quotes the conversation. The profile + shows `feedback.evaluations` (judged, mean per evaluator, tool failures + corroborated). **Default OFF** (`FEEDBACK_EVAL_SAMPLING_ENABLED`, + `CDK_FEEDBACK_EVAL_SAMPLING_ENABLED=true` to enable): the managed judge + reads the sampled conversation's spans, which is the scoping decision the + evaluations spike (§2) says to make explicitly per environment, and §8 + rule 1 here. The IAM grant (Logs Insights on the runtime log group and + `aws/spans`; Evaluate / GetEvaluator) is wired but inert until an + environment opts in. Runs are admin-triggered; a schedule can follow once + a week of verdicts says the token spend is worth it. +- **Not built**: author/marketplace surfaces (PR-5), the report-dialog + escape hatch in the reason row, abandonment, the `outdated` → KB-freshness + join, a schedule for the sampler. diff --git a/frontend/ai.client/src/app/admin/costs/models/admin-cost.models.ts b/frontend/ai.client/src/app/admin/costs/models/admin-cost.models.ts index f58fc1847..08234735d 100644 --- a/frontend/ai.client/src/app/admin/costs/models/admin-cost.models.ts +++ b/frontend/ai.client/src/app/admin/costs/models/admin-cost.models.ts @@ -442,6 +442,13 @@ export interface FeedbackProfile { /** Down-thumbs followed by a retry-with-correction, and what the rework cost. */ retried?: number; reworkUsd?: number | null; + /** Judged down-thumbs (eval sampling): counts and means only; null when none judged. */ + evaluations?: { + judged: number; + byEvaluator: Record; + toolFailuresReported: number; + toolFailuresCorroborated: number; + } | null; } /** The content-free diagnostic profile of one conversation. */ diff --git a/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.ts b/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.ts index f12d35a0f..0788611cd 100644 --- a/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.ts +++ b/frontend/ai.client/src/app/admin/costs/pages/session-cost-anatomy.page.ts @@ -38,6 +38,7 @@ import { buildDiagnosticJson, downRate, feedbackByTurnClassLine, + feedbackEvaluationsLine, feedbackRetryLine, formatBytes, formatEvidenceValue, @@ -257,6 +258,10 @@ import {

{{ retries }}

} + @if (feedbackEvaluationsLine(); as judged) { + +

{{ judged }}

+ } } @else {

—

not tracked

@@ -843,6 +848,10 @@ export class SessionCostAnatomyPage { return feedback ? downRate(feedback) : null; }); + readonly feedbackEvaluationsLine = computed(() => + this.profileResource.hasValue() ? feedbackEvaluationsLine(this.profileResource.value().feedback) : null, + ); + readonly feedbackRetryLine = computed(() => this.profileResource.hasValue() ? feedbackRetryLine(this.profileResource.value().feedback) : null, ); diff --git a/frontend/ai.client/src/app/admin/costs/pages/session-profile.util.spec.ts b/frontend/ai.client/src/app/admin/costs/pages/session-profile.util.spec.ts index ec9ad88ac..cdf666641 100644 --- a/frontend/ai.client/src/app/admin/costs/pages/session-profile.util.spec.ts +++ b/frontend/ai.client/src/app/admin/costs/pages/session-profile.util.spec.ts @@ -4,6 +4,7 @@ import { cleanCeiling, downRate, feedbackByTurnClassLine, + feedbackEvaluationsLine, feedbackRetryLine, formatBytes, formatEvidenceValue, @@ -50,6 +51,23 @@ describe('session-profile.util', () => { expect(feedbackRetryLine({ up: 0, down: 2, retried: 2, reworkUsd: 0.351 })).toBe('2 retried · $0.35 rework'); }); + it('feedbackEvaluationsLine reports judged means and corroboration, null when nothing judged', () => { + expect(feedbackEvaluationsLine({ up: 0, down: 2 })).toBeNull(); + expect(feedbackEvaluationsLine({ up: 0, down: 2, evaluations: null })).toBeNull(); + expect( + feedbackEvaluationsLine({ + up: 0, + down: 3, + evaluations: { + judged: 3, + byEvaluator: { 'Builtin.Faithfulness': { n: 1, mean: 0.5 }, 'Builtin.Correctness': { n: 2, mean: 0.25 } }, + toolFailuresReported: 2, + toolFailuresCorroborated: 1, + }, + }), + ).toBe('judged 3 · Correctness 0.25 · Faithfulness 0.50 · tool failures 1/2 confirmed'); + }); + it('feedbackByTurnClassLine is null when the turn class is not tracked', () => { expect(feedbackByTurnClassLine({ up: 1, down: 1, byTurnClass: null })).toBeNull(); expect(feedbackByTurnClassLine({ up: 1, down: 1 })).toBeNull(); diff --git a/frontend/ai.client/src/app/admin/costs/pages/session-profile.util.ts b/frontend/ai.client/src/app/admin/costs/pages/session-profile.util.ts index eb7a63ffe..ccd7f0b5c 100644 --- a/frontend/ai.client/src/app/admin/costs/pages/session-profile.util.ts +++ b/frontend/ai.client/src/app/admin/costs/pages/session-profile.util.ts @@ -241,6 +241,24 @@ export function feedbackRetryLine(feedback: FeedbackProfile | null | undefined): return parts.join(' · '); } +/** + * `judged 3 · Correctness 0.50 · tool failures 1/2 confirmed` — what the eval + * sampler concluded. Evaluator names lose their `Builtin.` prefix; null when + * nothing was judged. + */ +export function feedbackEvaluationsLine(feedback: FeedbackProfile | null | undefined): string | null { + const ev = feedback?.evaluations; + if (!ev || ev.judged === 0) return null; + const parts = [`judged ${ev.judged}`]; + for (const [name, agg] of Object.entries(ev.byEvaluator ?? {}).sort(([a], [b]) => a.localeCompare(b))) { + parts.push(`${name.replace(/^Builtin\./, '')} ${agg.mean.toFixed(2)}`); + } + if (ev.toolFailuresReported > 0) { + parts.push(`tool failures ${ev.toolFailuresCorroborated}/${ev.toolFailuresReported} confirmed`); + } + return parts.join(' · '); +} + export function feedbackByTurnClassLine(feedback: FeedbackProfile | null | undefined): string | null { const by = feedback?.byTurnClass; if (!by) return null; diff --git a/infrastructure/lib/config.ts b/infrastructure/lib/config.ts index f1b531fe0..024700d47 100644 --- a/infrastructure/lib/config.ts +++ b/infrastructure/lib/config.ts @@ -52,6 +52,7 @@ export interface AppConfig { managedKb: ManagedKbConfig; scheduledRuns: ScheduledRunsConfig; memorySpaces: MemorySpacesConfig; + feedbackEvalSampling: FeedbackEvalSamplingConfig; skills: SkillsConfig; agents: AgentsConfig; agentMarketplace: AgentMarketplaceConfig; @@ -274,6 +275,16 @@ export interface MemorySpacesConfig { enabled: boolean; } +/** + * Feedback eval sampling (response-feedback spec §11 PR-4): lets an admin + * send down-thumbed conversations to AgentCore Evaluations. **Opt-in** — + * the managed evaluator reads the conversation's spans (system prompt and + * user messages), so each environment turns it on deliberately. + */ +export interface FeedbackEvalSamplingConfig { + enabled: boolean; +} + /** * Skills feature flag (Skills v2). Default ON with a kill switch — the epic is * complete and dogfooded, so it ships enabled for every deployer (opt-out), @@ -872,6 +883,17 @@ export function loadConfig(scope: cdk.App): AppConfig { ? process.env.CDK_SCHEDULED_RUNS_ENABLED !== 'false' : scope.node.tryGetContext('scheduledRuns')?.enabled ?? true, }, + feedbackEvalSampling: { + // Default OFF, opt-in (the `fineTuning`-style deferred pattern inverted): + // only the literal "true" enables. Sending real conversations to an + // AWS-managed judge is the scoping decision the evaluations spike says to + // make explicitly per environment — a workflow's empty/unset variable must + // never make it. A `feedbackEvalSampling.enabled: true` cdk.json context + // also enables it. + enabled: process.env.CDK_FEEDBACK_EVAL_SAMPLING_ENABLED + ? process.env.CDK_FEEDBACK_EVAL_SAMPLING_ENABLED === 'true' + : scope.node.tryGetContext('feedbackEvalSampling')?.enabled ?? false, + }, memorySpaces: { // Default ON with a kill switch: Memory Spaces is a complete feature and // ships enabled for every deployer (opt-out, not opt-in — matches kbSync / diff --git a/infrastructure/lib/constructs/app-api/app-api-environment.ts b/infrastructure/lib/constructs/app-api/app-api-environment.ts index 3bfbabd7f..576208db6 100644 --- a/infrastructure/lib/constructs/app-api/app-api-environment.ts +++ b/infrastructure/lib/constructs/app-api/app-api-environment.ts @@ -81,6 +81,8 @@ export interface AppApiSsmParams { voiceTicketSigningSecretArn: string; // Inference inferenceApiRuntimeEndpointUrl: string; + /** AgentCore Runtime CloudWatch log group (spans + content log records) for eval sampling. */ + agentCoreRuntimeLogGroupName: string; // File uploads userFilesBucketName: string; userFilesBucketArn: string; @@ -117,6 +119,8 @@ export interface AppApiBackendOverrides { memoryId: string; /** AgentCore Runtime endpoint URL (from InferenceAgentCoreConstruct.runtimeEndpointUrl). */ inferenceApiRuntimeEndpointUrl: string; + /** AgentCore Runtime log group name (from InferenceAgentCoreConstruct.runtimeLogGroupName). */ + agentCoreRuntimeLogGroupName: string; } /** Resolve every value the App API construct needs. @@ -200,6 +204,7 @@ export function resolveAppApiParams( voiceTicketSigningSecretArn: refs.voiceTicketSigningSecret.secretArn, // Inference inferenceApiRuntimeEndpointUrl: overrides.inferenceApiRuntimeEndpointUrl, + agentCoreRuntimeLogGroupName: overrides.agentCoreRuntimeLogGroupName, // File uploads userFilesBucketName: refs.fileUploadBucket.bucketName, userFilesBucketArn: refs.fileUploadBucket.bucketArn, @@ -338,6 +343,13 @@ export function buildAppApiEnvironment( // every read 502s (ResourceNotFoundException). inference-api already sets // the identical trio — app-api owns the CRUD surface, so it needs them too. MEMORY_SPACES_ENABLED: config.memorySpaces.enabled ? 'true' : 'false', + // Feedback eval sampling (response-feedback spec §11 PR-4): OPT-IN per + // environment — the admin batch sends down-thumbed conversations' spans to + // an AWS-managed evaluator. The runtime log group is where those spans and + // the content-bearing log records live (evaluations spike §1); it is wired + // regardless so turning the flag on is a one-variable change. + FEEDBACK_EVAL_SAMPLING_ENABLED: config.feedbackEvalSampling.enabled ? 'true' : 'false', + AGENTCORE_RUNTIME_LOG_GROUP: params.agentCoreRuntimeLogGroupName, DYNAMODB_MEMORY_SPACES_TABLE_NAME: params.memorySpacesTableName, S3_MEMORY_SPACES_BUCKET_NAME: params.memorySpacesBucketName, // Skills v2 (default ON with a kill switch per env). Skills live in the diff --git a/infrastructure/lib/constructs/app-api/app-api-iam-grants.ts b/infrastructure/lib/constructs/app-api/app-api-iam-grants.ts index 695cf6abe..7ededd2d7 100644 --- a/infrastructure/lib/constructs/app-api/app-api-iam-grants.ts +++ b/infrastructure/lib/constructs/app-api/app-api-iam-grants.ts @@ -48,6 +48,12 @@ export interface AppApiIamGrantsProps { * `refs` later if convenient. */ agentCoreMemoryArn: string; + /** + * AgentCore Runtime CloudWatch log group name. Feedback eval sampling runs + * Logs Insights queries against it (and `aws/spans`) to collect a + * conversation's spans for AgentCore Evaluations. + */ + agentCoreRuntimeLogGroupName: string; /** * SageMaker fine-tuning execution role ARN. Created by a sibling * construct in wireCompute() — passed in here. @@ -508,6 +514,44 @@ export function grantAppApiPermissions(props: AppApiIamGrantsProps): void { }), ); + // ── AgentCore Evaluations (feedback eval sampling, spec §11 PR-4) ── + // The admin batch judges down-thumbed conversations with the built-in + // evaluators. Two halves: the SDK's span collector runs Logs Insights + // queries over the runtime log group and `aws/spans` (StartQuery is + // resource-scoped; GetQueryResults/StopQuery are not), then calls the + // data-plane Evaluate with the spans and the control-plane GetEvaluator + // to learn each evaluator's level. Built-in evaluators are AWS-owned, so + // the bedrock-agentcore actions cannot be resource-scoped. The flag + // (FEEDBACK_EVAL_SAMPLING_ENABLED) defaults OFF; the grant is inert until + // an environment opts in. + taskRole.addToPrincipalPolicy( + new iam.PolicyStatement({ + sid: 'FeedbackEvalSpanQueries', + effect: iam.Effect.ALLOW, + actions: ['logs:StartQuery'], + resources: [ + `arn:aws:logs:${config.awsRegion}:${config.awsAccount}:log-group:${props.agentCoreRuntimeLogGroupName}:*`, + `arn:aws:logs:${config.awsRegion}:${config.awsAccount}:log-group:aws/spans:*`, + ], + }), + ); + taskRole.addToPrincipalPolicy( + new iam.PolicyStatement({ + sid: 'FeedbackEvalSpanQueryResults', + effect: iam.Effect.ALLOW, + actions: ['logs:GetQueryResults', 'logs:StopQuery'], + resources: ['*'], + }), + ); + taskRole.addToPrincipalPolicy( + new iam.PolicyStatement({ + sid: 'FeedbackEvalEvaluate', + effect: iam.Effect.ALLOW, + actions: ['bedrock-agentcore:Evaluate', 'bedrock-agentcore:GetEvaluator', 'bedrock-agentcore:ListEvaluators'], + resources: ['*'], + }), + ); + // ── Bedrock model invocation ── // Used by both title generation and the API-key `/chat/api-converse` // handler (apis/app_api/chat/converse_routes.py), which calls Bedrock diff --git a/infrastructure/lib/constructs/app-api/app-api-service-construct.ts b/infrastructure/lib/constructs/app-api/app-api-service-construct.ts index e681a25af..a317f034d 100644 --- a/infrastructure/lib/constructs/app-api/app-api-service-construct.ts +++ b/infrastructure/lib/constructs/app-api/app-api-service-construct.ts @@ -41,6 +41,12 @@ export interface AppApiServiceConstructProps { * `INFERENCE_API_URL` env var. */ inferenceApiRuntimeEndpointUrl: string; + /** + * AgentCore Runtime CloudWatch log group name (same-stack ref via + * InferenceAgentCoreConstruct.runtimeLogGroupName). The App API reads it + * for feedback eval sampling and is granted Logs Insights queries on it. + */ + agentCoreRuntimeLogGroupName: string; /** * Artifacts iframe origin URL (https://artifacts.{domain}). Same-stack * ref via ArtifactsDistributionConstruct; used as the App API @@ -95,6 +101,7 @@ export class AppApiServiceConstruct extends Construct { const params = resolveAppApiParams(props.refs, { memoryId: props.agentCoreMemoryId, inferenceApiRuntimeEndpointUrl: props.inferenceApiRuntimeEndpointUrl, + agentCoreRuntimeLogGroupName: props.agentCoreRuntimeLogGroupName, }); // ── Network resources (typed refs from PlatformStack) ── @@ -247,6 +254,7 @@ export class AppApiServiceConstruct extends Construct { taskRole: taskDefinition.taskRole, refs: props.refs, agentCoreMemoryArn: props.agentCoreMemoryArn, + agentCoreRuntimeLogGroupName: props.agentCoreRuntimeLogGroupName, sagemakerExecutionRoleArn: props.sagemakerExecutionRoleArn, }); diff --git a/infrastructure/lib/platform-stack.ts b/infrastructure/lib/platform-stack.ts index 042db7383..27e8ae120 100644 --- a/infrastructure/lib/platform-stack.ts +++ b/infrastructure/lib/platform-stack.ts @@ -898,6 +898,7 @@ export class PlatformStack extends cdk.Stack { agentCoreMemoryArn: this.agentCoreMemoryArn, agentCoreMemoryId: this.agentCoreMemoryId, inferenceApiRuntimeEndpointUrl: inferenceApi.runtimeEndpointUrl, + agentCoreRuntimeLogGroupName: inferenceApi.runtimeLogGroupName, artifactsOrigin: this.artifactsOriginUrl, sagemakerExecutionRoleArn: sagemaker.executionRole.roleArn, sagemakerSecurityGroupId: sagemaker.securityGroup.securityGroupId, diff --git a/infrastructure/test/config.test.ts b/infrastructure/test/config.test.ts index 423f0534c..5b67f7d4f 100644 --- a/infrastructure/test/config.test.ts +++ b/infrastructure/test/config.test.ts @@ -426,6 +426,31 @@ describe('RAG Ingestion Configuration', () => { // (same ternary as kbSync; empty workflow var must not disable) // ============================================================ + describe('Feedback eval sampling flag (opt-in)', () => { + test('defaults to DISABLED when CDK_FEEDBACK_EVAL_SAMPLING_ENABLED is unset', () => { + delete process.env.CDK_FEEDBACK_EVAL_SAMPLING_ENABLED; + expect(loadConfig(app).feedbackEvalSampling.enabled).toBe(false); + }); + + test('an empty string (unset workflow variable) stays disabled', () => { + process.env.CDK_FEEDBACK_EVAL_SAMPLING_ENABLED = ''; + expect(loadConfig(app).feedbackEvalSampling.enabled).toBe(false); + }); + + test('only the literal "true" enables it', () => { + process.env.CDK_FEEDBACK_EVAL_SAMPLING_ENABLED = 'true'; + expect(loadConfig(app).feedbackEvalSampling.enabled).toBe(true); + process.env.CDK_FEEDBACK_EVAL_SAMPLING_ENABLED = 'yes'; + expect(loadConfig(app).feedbackEvalSampling.enabled).toBe(false); + }); + + test('cdk.json context feedbackEvalSampling.enabled=true enables when env is unset', () => { + delete process.env.CDK_FEEDBACK_EVAL_SAMPLING_ENABLED; + app.node.setContext('feedbackEvalSampling', { enabled: true }); + expect(loadConfig(app).feedbackEvalSampling.enabled).toBe(true); + }); + }); + describe('Scheduled Runs feature flag', () => { test('defaults to enabled when CDK_SCHEDULED_RUNS_ENABLED is unset', () => { delete process.env.CDK_SCHEDULED_RUNS_ENABLED; diff --git a/infrastructure/test/helpers/mock-config.ts b/infrastructure/test/helpers/mock-config.ts index 947317a20..ff5c75d0e 100644 --- a/infrastructure/test/helpers/mock-config.ts +++ b/infrastructure/test/helpers/mock-config.ts @@ -134,6 +134,7 @@ export function createMockConfig(overrides: Partial = {}): AppConfig memorySpaces: { enabled: false, }, + feedbackEvalSampling: { enabled: false }, skills: { enabled: false, },