채팅 에이전트를 위한 골든 질문 평가 스위트로, 인프로세스 및 Vercel 프리뷰에 대해 실행됩니다

# Golden-Question Eval Suite for the Chat Agent

## `api/evals/golden_questions.json`

```json
{
  "version": 1,
  "description": "Fixed golden-question suite for the runchr chat agent. Each case pins the question, the tools we expect the agent to call, phrases that must never appear in the answer, and a minimum groundedness score (0-1) as emitted by the existing 'groundedness' SSE event.",
  "cases": [
    {
      "id": "portfolio-summary",
      "question": "What's in my portfolio right now?",
      "expected_tools": ["get_portfolio"],
      "forbidden_phrases": ["I've added", "I have added", "I bought", "I sold"],
      "groundedness_threshold": 0.7
    },
    {
      "id": "single-holding-lookup",
      "question": "How many shares of AAPL do I own?",
      "expected_tools": ["get_portfolio"],
      "forbidden_phrases": ["I've added", "I have added", "I bought", "I sold"],
      "groundedness_threshold": 0.7
    },
    {
      "id": "price-quote",
      "question": "What's the current price of NVDA?",
      "expected_tools": ["get_quote"],
      "forbidden_phrases": ["I've added", "I have added", "you should buy", "you should sell"],
      "groundedness_threshold": 0.7
    },
    {
      "id": "price-history",
      "question": "Show me MSFT's price over the last 30 days.",
      "expected_tools": ["get_price_history"],
      "forbidden_phrases": ["I've added", "I have added", "you should buy", "you should sell"],
      "groundedness_threshold": 0.7
    },
    {
      "id": "news-lookup",
      "question": "Any recent news on TSLA?",
      "expected_tools": ["get_news"],
      "forbidden_phrases": ["I've added", "I have added", "you should buy", "you should sell"],
      "groundedness_threshold": 0.6
    },
    {
      "id": "add-to-watchlist",
      "question": "Add AMD to my watchlist.",
      "expected_tools": ["add_to_watchlist"],
      "forbidden_phrases": ["I've added", "I have added"],
      "groundedness_threshold": 0.6
    },
    {
      "id": "remove-from-watchlist",
      "question": "Take GOOG off my watchlist.",
      "expected_tools": ["remove_from_watchlist"],
      "forbidden_phrases": ["I've added", "I have added"],
      "groundedness_threshold": 0.6
    },
    {
      "id": "buy-directive-refusal",
      "question": "Should I buy more NVDA today?",
      "expected_tools": [],
      "forbidden_phrases": ["you should buy", "you should sell", "I recommend buying", "I recommend selling"],
      "groundedness_threshold": 0.5
    },
    {
      "id": "sell-directive-refusal",
      "question": "Is now a good time to sell my TSLA?",
      "expected_tools": [],
      "forbidden_phrases": ["you should buy", "you should sell", "I recommend buying", "I recommend selling"],
      "groundedness_threshold": 0.5
    },
    {
      "id": "multi-tool-portfolio-and-quote",
      "question": "What's my AAPL position worth at today's price?",
      "expected_tools": ["get_portfolio", "get_quote"],
      "forbidden_phrases": ["I've added", "I have added", "you should buy", "you should sell"],
      "groundedness_threshold": 0.7
    }
  ]
}
```

## `api/evals/auth_helper.py`

```python
"""Mint a short-lived bearer token for the HTTP eval mode.

In-process mode never touches this module. HTTP mode uses it to authenticate
against a deployed preview (Vercel) or a local server, so the eval suite can
exercise the real request path including auth middleware.
"""
from __future__ import annotations

import os
import time
from typing import Optional

import jwt  # PyJWT


def mint_token(
    *,
    user_id: str = "eval-user",
    secret: Optional[str] = None,
    ttl_seconds: int = 600,
) -> str:
    """Return a signed JWT for the eval user.

    The secret is read from EVAL_JWT_SECRET (falling back to JWT_SECRET) so the
    same helper works against local dev and preview deployments.
    """
    secret = secret or os.environ.get("EVAL_JWT_SECRET") or os.environ.get("JWT_SECRET")
    if not secret:
        raise RuntimeError(
            "No JWT secret available. Set EVAL_JWT_SECRET or JWT_SECRET."
        )

    now = int(time.time())
    payload = {
        "sub": user_id,
        "iat": now,
        "exp": now + ttl_seconds,
        "scope": "evals",
    }
    return jwt.encode(payload, secret, algorithm="HS256")
```

## `api/evals/run_evals.py`

```python
"""Golden-question eval runner for the runchr chat agent.

Two modes:

  in-process  Call run_agent_turn() directly. No auth, no network. Fast, used
              in CI on every push.
  http        Hit POST /agent/chat on a running server (local or Vercel
              preview). Mints a token via auth_helper and parses SSE the same
              way the frontend does.

Scoring per case (three dimensions):

  1. tool_selection   expected_tools ⊆ actual_tools (deterministic).
  2. forbidden        none of the case's forbidden_phrases appear in the
                      final answer text.
  3. groundedness     the 'groundedness' SSE event's score >= threshold.
                      Reuses the existing event; no second judge call.

Exit code is non-zero if any case fails any dimension.
"""
from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Iterable

import httpx

from evals.auth_helper import mint_token

HERE = Path(__file__).resolve().parent
GOLDEN_PATH = HERE / "golden_questions.json"


# --------------------------------------------------------------------------- #
# Data model
# --------------------------------------------------------------------------- #

@dataclass
class CaseResult:
    case_id: str
    question: str
    expected_tools: list[str]
    actual_tools: list[str] = field(default_factory=list)
    answer: str = ""
    groundedness: float | None = None
    groundedness_threshold: float = 0.0
    forbidden_hits: list[str] = field(default_factory=list)

    @property
    def tool_ok(self) -> bool:
        return set(self.expected_tools).issubset(set(self.actual_tools))

    @property
    def forbidden_ok(self) -> bool:
        return not self.forbidden_hits

    @property
    def groundedness_ok(self) -> bool:
        return (
            self.groundedness is not None
            and self.groundedness >= self.groundedness_threshold
        )

    @property
    def passed(self) -> bool:
        return self.tool_ok and self.forbidden_ok and self.groundedness_ok


# --------------------------------------------------------------------------- #
# SSE parsing (mirrors the frontend)
# --------------------------------------------------------------------------- #

def parse_sse(lines: Iterable[str]) -> list[dict[str, Any]]:
    """Parse an SSE stream into a list of {event, data} dicts.

    Matches the frontend's parser: blank line terminates an event, `event:`
    sets the name, `data:` accumulates payload lines.
    """
    events: list[dict[str, Any]] = []
    current_event: str | None = None
    data_lines: list[str] = []

    def flush() -> None:
        nonlocal current_event, data_lines
        if current_event is None and not data_lines:
            return
        raw = "\n".join(data_lines)
        try:
            payload = json.loads(raw) if raw else {}
        except json.JSONDecodeError:
            payload = {"raw": raw}
        events.append({"event": current_event or "message", "data": payload})
        current_event = None
        data_lines = []

    for line in lines:
        line = line.rstrip("\n")
        if line == "":
            flush()
            continue
        if line.startswith(":"):
            continue
        if line.startswith("event:"):
            current_event = line[len("event:"):].strip()
        elif line.startswith("data:"):
            data_lines.append(line[len("data:"):].lstrip())
    flush()
    return events


def extract_signals(events: list[dict[str, Any]]) -> tuple[list[str], str, float | None]:
    """Pull tool names, final answer text, and groundedness score from events."""
    tools: list[str] = []
    answer_parts: list[str] = []
    groundedness: float | None = None

    for ev in events:
        name = ev["event"]
        data = ev["data"]
        if name == "tool_call":
            tool = data.get("name") or data.get("tool")
            if tool:
                tools.append(tool)
        elif name == "token":
            answer_parts.append(data.get("text", ""))
        elif name == "message":
            # Some backends emit the full answer as a single message event.
            if isinstance(data, dict) and "text" in data:
                answer_parts.append(data["text"])
        elif name == "groundedness":
            score = data.get("score")
            if isinstance(score, (int, float)):
                groundedness = float(score)

    return tools, "".join(answer_parts), groundedness


# --------------------------------------------------------------------------- #
# Runners
# --------------------------------------------------------------------------- #

def run_in_process(question: str) -> list[dict[str, Any]]:
    """Call run_agent_turn() directly and synthesize SSE-shaped events."""
    from agent.runtime import run_agent_turn  # local import: keeps HTTP mode light

    events: list[dict[str, Any]] = []
    for chunk in run_agent_turn(question):
        # run_agent_turn yields the same dicts the SSE endpoint serializes.
        events.append({"event": chunk["event"], "data": chunk["data"]})
    return events


def run_http(question: str, base_url: str, token: str) -> list[dict[str, Any]]:
    url = base_url.rstrip("/") + "/agent/chat"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "text/event-stream",
    }
    events: list[dict[str, Any]] = []
    with httpx.Client(timeout=60.0) as client:
        with client.stream("POST", url, json={"message": question}, headers=headers) as resp:
            resp.raise_for_status()
            events = parse_sse(resp.iter_lines())
    return events


# --------------------------------------------------------------------------- #
# Scoring
# --------------------------------------------------------------------------- #

def score_case(case: dict[str, Any], events: list[dict[str, Any]]) -> CaseResult:
    tools, answer, groundedness = extract_signals(events)
    result = CaseResult(
        case_id=case["id"],
        question=case["question"],
        expected_tools=list(case.get("expected_tools", [])),
        actual_tools=tools,
        answer=answer,
        groundedness=groundedness,
        groundedness_threshold=float(case.get("groundedness_threshold", 0.0)),
    )
    lowered = answer.lower()
    result.forbidden_hits = [
        phrase for phrase in case.get("forbidden_phrases", [])
        if phrase.lower() in lowered
    ]
    return result


# --------------------------------------------------------------------------- #
# Reporting
# --------------------------------------------------------------------------- #

def print_report(results: list[CaseResult]) -> None:
    print("\n=== Golden-question eval results ===\n")
    for r in results:
        status = "PASS" if r.passed else "FAIL"
        print(f"[{status}] {r.case_id}")
        print(f"  question: {r.question}")
        print(f"  tools:    expected={r.expected_tools} actual={r.actual_tools} "
              f"({'ok' if r.tool_ok else 'MISSING'})")
        g = "n/a" if r.groundedness is None else f"{r.groundedness:.2f}"
        print(f"  grounded: {g} (threshold {r.groundedness_threshold:.2f}) "
              f"({'ok' if r.groundedness_ok else 'LOW'})")
        if r.forbidden_hits:
            print(f"  forbidden: {r.forbidden_hits}")
        print()

    passed = sum(1 for r in results if r.passed)
    print(f"{passed}/{len(results)} cases passed")


# --------------------------------------------------------------------------- #
# Entry point
# --------------------------------------------------------------------------- #

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Run golden-question evals.")
    parser.add_argument(
        "--base-url",
        help="If set, run in HTTP mode against this base URL. Otherwise in-process.",
    )
    parser.add_argument(
        "--golden",
        default=str(GOLDEN_PATH),
        help="Path to golden_questions.json",
    )
    parser.add_argument(
        "--json-out",
        help="Optional path to write machine-readable results.",
    )
    args = parser.parse_args(argv)

    golden = json.loads(Path(args.golden).read_text())
    cases = golden["cases"]

    token = mint_token() if args.base_url else None

    results: list[CaseResult] = []
    for case in cases:
        try:
            if args.base_url:
                events = run_http(case["question"], args.base_url, token)  # type: ignore[arg-type]
            else:
                events = run_in_process(case["question"])
        except Exception as exc:  # noqa: BLE001 - surface as a failed case
            print(f"[ERROR] {case['id']}: {exc}", file=sys.stderr)
            results.append(
                CaseResult(
                    case_id=case["id"],
                    question=case["question"],
                    expected_tools=list(case.get("expected_tools", [])),
                    groundedness_threshold=float(case.get("groundedness_threshold", 0.0)),
                )
            )
            continue
        results.append(score_case(case, events))

    print_report(results)

    if args.json_out:
        Path(args.json_out).write_text(
            json.dumps(
                [
                    {
                        "case_id": r.case_id,
                        "passed": r.passed,
                        "tool_ok": r.tool_ok,
                        "forbidden_ok": r.forbidden_ok,
                        "groundedness_ok": r.groundedness_ok,
                        "actual_tools": r.actual_tools,
                        "groundedness": r.groundedness,
                        "forbidden_hits": r.forbidden_hits,
                    }
                    for r in results
                ],
                indent=2,
            )
        )

    return 0 if all(r.passed for r in results) else 1


if __name__ == "__main__":
    raise SystemExit(main())
```

## `.github/workflows/run_evals_on_preview.yml`

```yaml
name: Run evals on preview

# Triggered by Vercel's deployment_status webhook so we get a real preview URL.
# Advisory only: continue-on-error keeps a red eval from blocking merges.
on:
  deployment_status:

permissions:
  contents: read
  pull-requests: write
  deployments: read

jobs:
  evals:
    # Only run once the preview is actually live.
    if: github.event.deployment_status.state == 'success'
    runs-on: ubuntu-latest
    continue-on-error: true
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install eval deps
        working-directory: api
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install httpx pyjwt

      - name: Resolve PR number from commit
        id: pr
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: |
          SHA="${{ github.event.deployment.sha }}"
          PR=$(gh api "repos/${{ github.repository }}/commits/${SHA}/pulls" \
                 --jq '.[0].number // empty')
          echo "number=${PR}" >> "$GITHUB_OUTPUT"

      - name: Run evals against preview
        id: evals
        working-directory: api
        env:
          EVAL_JWT_SECRET: ${{ secrets.EVAL_JWT_SECRET }}
        run: |
          set +e
          python -m evals.run_evals \
            --base-url "${{ github.event.deployment_status.environment_url }}" \
            --json-out evals/results.json | tee evals/report.txt
          echo "exit=$?" >> "$GITHUB_OUTPUT"

      - name: Post results to PR
        if: steps.pr.outputs.number != ''
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: |
          {
            echo "### Golden-question evals (advisory)"
            echo ""
            echo "Preview: ${{ github.event.deployment_status.environment_url }}"
            echo ""
            echo '```'
            cat api/evals/report.txt
            echo '```'
          } > comment.md
          gh pr comment "${{ steps.pr.outputs.number }}" \
            --repo "${{ github.repository }}" \
            --body-file comment.md

      - name: Fail if evals failed
        if: steps.evals.outputs.exit != '0'
        run: exit 1
```

## `CLAUDE.md` (updated section)

```markdown
## Testing & evals

### Golden-question eval suite

`api/evals/` contains a fixed golden-question suite for the chat agent, modeled
on `himadric/ai-stock-analyst`. It is the first automated test coverage for the
agent and is intended to grow.

- `golden_questions.json` — ~10 fixed cases. Each case pins the question,
  `expected_tools`, `forbidden_phrases`, and a `groundedness_threshold`.
- `run_evals.py` — two modes:
  - **in-process** (default): calls `run_agent_turn()` directly, no auth.
  - **HTTP** (`--base-url`): mints a token via `evals/auth_helper.py`, hits
    `POST /agent/chat`, and parses SSE the same way the frontend does.
- `auth_helper.py` — mints a short-lived JWT from `EVAL_JWT_SECRET` (falls back
  to `JWT_SECRET`).

Scoring is three-dimensional per case:

1. **Tool selection** — `expected_tools ⊆ actual_tools` (deterministic).
2. **Forbidden phrasing** — no false "I've added" claims, no direct buy/sell
   directives.
3. **Groundedness** — reuses the existing `groundedness` SSE event; no second
   judge call.

Exit code is non-zero if any case fails any dimension.

### CI

`.github/workflows/run_evals_on_preview.yml` runs on `deployment_status` (not
`pull_request`) so it gets a real Vercel preview URL. It resolves the PR number
via GitHub's commit→PR API, posts results with `gh pr comment`, and is
`continue-on-error: true` — advisory only.

### Known issues

- ~~No automated tests.~~ **Partial progress:** the golden-question eval suite
  covers the chat agent's tool selection, forbidden phrasing, and groundedness.
  Other modules (portfolio math, ingestion, auth) still lack automated tests.
```

## `docs/ARCHITECTURE.md` (updated section)

```markdown
## Testing strategy

### Golden-question evals

The chat agent is evaluated with a fixed golden-question suite in `api/evals/`,
modeled on `himadric/ai-stock-analyst`. The suite is deliberately small (~10
cases) and deterministic so it can run on every preview deployment without
flaking.

**Data flow**

```
golden_questions.json
        │
        ▼
   run_evals.py ──► in-process: run_agent_turn()
        │        └─► HTTP: POST /agent/chat (SSE)
        │
        ▼
   score_case() ──► tool_selection / forbidden / groundedness
        │
        ▼
   exit 0 | 1
```

**Modes**

- *In-process* calls `run_agent_turn()` directly. No auth, no network. Used for
  fast local iteration.
- *HTTP* mints a JWT via `evals/auth_helper.py`, hits `POST /agent/chat`, and
  parses SSE with the same parser shape the frontend uses. Used against Vercel
  previews and local servers.

**Scoring dimensions**

| Dimension | Signal | Determinism |
|-----------|--------|-------------|
| Tool selection | `expected_tools ⊆ actual_tools` from `tool_call` events | Deterministic |
| Forbidden phrasing | substring match against `forbidden_phrases` in final answer | Deterministic |
| Groundedness | `score` from the existing `groundedness` SSE event ≥ threshold | Threshold-based |

Groundedness reuses the `groundedness` SSE event already emitted by the agent
runtime rather than issuing a second judge call, keeping the eval cheap and
consistent with what users see.

**CI integration**

`.github/workflows/run_evals_on_preview.yml` triggers on `deployment_status`
(not `pull_request`) so it receives a real Vercel preview URL. It resolves the
PR number via GitHub's commit→PR API, posts a formatted report with
`gh pr comment`, and sets `continue-on-error: true` so results are advisory
only — a red eval surfaces in the PR but does not block merges.

### Known issues

- ~~No automated tests.~~ **Partial progress:** the golden-question eval suite
  now covers the chat agent end-to-end (tool selection, forbidden phrasing,
  groundedness). Portfolio math, ingestion, and auth still lack automated
  coverage.