refactor: split pilot architecture

Remove the obsolete dashboard now that Langfuse is the analytics surface.\nIntroduce focused transport, model, and configuration modules while preserving the ai_review facade, and document the current runtime architecture.
test: trigger review after DB rule insert
2026-09-01 00:17:16 +00:00 · 2026-08-31 19:19:59 +00:00 · 2026-08-31 19:18:59 +00:00 · 2026-08-31 18:37:41 +00:00
30 changed files with 326 additions and 3171 deletions
@@ -0,0 +1 @@
 # judge trigger 1788203999
@@ -1,5 +0,0 @@
 {
  "enabled": true,
  "model": "headroom/MiniMax-M2.7",
  "static_message": "PR-Agent pilot on this repo. Comments are LLM-generated; treat as suggestions, not mandates."
 }
@@ -35,9 +35,9 @@ code never has to leave your network.
 ## Status
-A **pilot** is live and reviewing real PRs. The full framework (`pragent init`,
+A **pilot** is live and reviewing real PRs. The current runtime architecture is
-tiering as code, analyzer fan-out, `explain` / `replay`) is designed but not
+documented in [`docs/architecture.md`](docs/architecture.md); older framework
-built — see [`docs/plans/`](docs/plans/).
+plans remain in [`docs/plans/`](docs/plans/) as historical design material.
 What works today:
@@ -51,10 +51,12 @@ What works today:
 - `.pr-review.json` for per-repo focus and house rules (also the opt-in flag)
 - token-usage reporting on every review, measured from opencode `step_finish`
  events
 - Langfuse traces, equivalent-cost reporting, evaluation scores, and feedback
  harvesting
 - containment against hostile PR content (see [Security](#security))
 Not yet: status checks, fail-close, attention tiering enforced in code (it is
-currently a skill the agent follows), multi-model routing.
+currently a skill the agent follows), multi-model routing, and a CLI framework.
 ## How a review runs
@@ -94,9 +96,8 @@ path is in [`pilot/README.md`](pilot/README.md).
 The model endpoint is supplied at runtime via `PRAGENT_MODEL_BASE_URL`; the
 committed `opencode.json` carries a placeholder.
-Per-review token spend, latency and equivalent cost are shipped to a
+Per-review token spend, latency, equivalent cost, and evaluation scores are
-self-hosted Langfuse, split into `ollama` and `claude` environments so the two
+shipped to a self-hosted Langfuse: [`pilot/README-langfuse.md`](pilot/README-langfuse.md).
 spend stories stay separate: [`pilot/README-langfuse.md`](pilot/README-langfuse.md).
 Emission is a silent no-op unless `LANGFUSE_HOST` and the key pair are set.
 ## Extending it
@@ -175,7 +176,7 @@ python3 pilot/cost_model.py --help      # other mixes, volumes, models
 ## Development
 ```bash
-python3 -m pytest tests -q    # 137 tests, stdlib only, no network
+python3 -m pytest tests -q    # stdlib-only tests, no network
 ```
 The pilot is stdlib-only Python by design — it runs from a bare `python:slim`
@@ -185,3 +186,5 @@ review time.
 ## License
 Not yet chosen. Until one is added, no reuse rights are granted.
 _pilot eval judges test 1788201461_
@@ -0,0 +1,68 @@
 # pragent current architecture
 Status: pilot implementation, September 2026.
 ## System shape
 ```text
 Gitea pull_request webhook
          │ signed HTTP
          ▼
 webhook_server ── trusted base config ──► review_config
          │ bounded worker
          ▼
 review_pr facade/orchestrator
   ├── gitea_client       fetch diff, reviews, config; publish review
   ├── diff_compress      reduce prompt context
   ├── opencode_review     isolated checkout + agent execution
   │       └── model / repo factory (.opencode)
   ├── review parsing      normalize findings + validate anchors
   ├── feedback            persist reactions and derive scores
   └── langfuse_trace      usage, cost, evaluation telemetry
 ```
 ## Seams and responsibilities
 The external seam is `ai_review.review_pr(...)`: one call represents one review
 attempt and returns success/skip status. The module is retained as a facade for
 the CI and webhook callers that already import it.
 The internal seams are deliberately narrower:
 - `review_config.repo_enabled(get, ...)` owns the security-sensitive opt-in
  decision. It receives a transport function, so malformed configuration and
  failure behavior are deterministic in tests.
 - `gitea_client.request()` and `GiteaClient` own HTTP authentication, JSON
  request encoding, timeout, and Gitea URL construction.
 - `model_client.complete()` owns the legacy Anthropic-compatible request shape.
  `opencode_review` is the preferred agent adapter and keeps Gitea I/O out of
  the autonomous process.
 - `diff_compress`, finding parsing, config filtering, and rendering remain
  pure transformations. Their callers do not need to know how model or Gitea
  transport works.
 - `langfuse_trace` is an optional sink. It is fail-open and cannot change the
  review result.
 ## Trust model
 The review config is read from the PR base branch, never the PR head. The agent
 checkout is treated as hostile: instruction files are removed, credentials are
 not inherited, and the agent only returns text to the Python publisher. Python
 validates finding paths and post-change line anchors before sending comments.
 ## Observability
 Langfuse is the operational analytics surface. A trace groups runs by
 `owner/repo#PR`; generations carry usage and cost basis; evaluation scores and
 human-feedback scores are attached later. The former SQLite-backed dashboard
 was removed. SQLite remains only as the feedback/evaluation ingestion store.
 ## Removed surface
 The dashboard server, dashboard data module, dashboard tests, dashboard README,
 and dashboard Kubernetes manifest are intentionally gone. Operators use the
 Langfuse UI for review trends and cost analysis, and Gitea for review details
 and configuration changes.
 Historical design/implementation plans under `docs/plans/` describe the
 earlier TypeScript framework proposal and are not the runtime architecture.
@@ -1,82 +0,0 @@
 # Judge-side think-block patcher. Stands between Langfuse evaluators and the
 # headroom-ollama hub (port 8790). Local Ollama does not emit the `signature`
 # field that Langfuse's Anthropic adapter's Zod schema requires on every
 # `thinking` content block — without it, the evaluator preflight fails as
 # "Invalid JSON response". The proxy forwards /v1/* verbatim and adds a dummy
 # signature to each thinking block before returning.
 apiVersion: v1
 kind: ConfigMap
 metadata:
  name: judge-proxy
  namespace: pragent
 data:
  proxy.py: |
    #!/usr/bin/env python3
    """Judge proxy: forward to headroom-ollama, fix thinking blocks."""
    import json, sys, urllib.request, urllib.error
    from http.server import BaseHTTPRequestHandler, HTTPServer
    from socketserver import ThreadingMixIn
    UPSTREAM = "http://100.74.17.70:8790"
    DUMMY_SIG = "kimi-local-judge-no-signature"
    class H(BaseHTTPRequestHandler):
        def _proxy(self):
            n = int(self.headers.get("Content-Length", 0))
            body = self.rfile.read(n) if n else b""
            h = {k: v for k, v in self.headers.items() if k.lower() not in ("host", "content-length")}
            req = urllib.request.Request(UPSTREAM + self.path, data=body, headers=h, method=self.command)
            try:
                with urllib.request.urlopen(req, timeout=120) as r:
                    resp_body = r.read(); status = r.status; rh = dict(r.headers)
            except urllib.error.HTTPError as e:
                resp_body = e.read(); status = e.code; rh = dict(e.headers)
            ct = rh.get("content-type", "")
            if status == 200 and "application/json" in ct and self.path.startswith("/v1/messages"):
                try:
                    obj = json.loads(resp_body)
                    patched = 0
                    for blk in obj.get("content") or []:
                        if isinstance(blk, dict) and blk.get("type") == "thinking" and "signature" not in blk:
                            blk["signature"] = DUMMY_SIG; patched += 1
                    if patched:
                        resp_body = json.dumps(obj).encode("utf-8")
                        rh["content-length"] = str(len(resp_body))
                        print(f"judge-proxy: patched {patched} thinking block(s)", file=sys.stderr, flush=True)
                except Exception as e:
                    print(f"judge-proxy: patch failed: {e}", file=sys.stderr, flush=True)
            self.send_response(status)
            for k, v in rh.items():
                if k.lower() not in ("transfer-encoding", "content-length", "connection"):
                    self.send_header(k, v)
            self.send_header("Content-Length", str(len(resp_body)))
            self.end_headers(); self.wfile.write(resp_body)
        def do_POST(self): self._proxy()
        def do_GET(self): self._proxy()
        def log_message(self, *a, **k): pass
    class S(ThreadingMixIn, HTTPServer): daemon_threads = True
    S(("0.0.0.0", 8802), H).serve_forever()
 ---
 apiVersion: v1
 kind: Pod
 metadata:
  name: judge-proxy
  namespace: pragent
  labels:
    app: judge-proxy
 spec:
  nodeSelector:
    kubernetes.io/hostname: kubernets
  hostNetwork: true
  dnsPolicy: ClusterFirstWithHostNet
  restartPolicy: Always
  containers:
  - name: p
    image: python:3.12-alpine
    command: ["sh","-c","apk add --no-cache ca-certificates >/dev/null && python3 -u /etc/cfg/proxy.py"]
    volumeMounts:
    - {name: cfg, mountPath: /etc/cfg}
    ports:
    - {containerPort: 8802, hostPort: 8802}
  volumes:
  - name: cfg
    configMap:
      name: judge-proxy
@@ -1,118 +0,0 @@
 # pragent pilot — central dashboard service.
 #
 # Read-only overview + per-repo / per-PR drilldown over the same SQLite
 # feedback DB the webhook writes. Also mutates `.pr-review.json` on covered
 # repos via the Gitea contents API (Tasks C+D in pilot/dashboard.py). Same
 # image as the webhook (`pragent-webhook:optin`) — all pilot modules are
 # baked in at /app/pilot/.
 #
 # Routes: GET /  (overview), GET /r/<o>/<n>  (repo), GET /r/<o>/<n>/<i>  (PR),
 # GET /r/<o>/<n>/<i>/raw  (PR markdown raw), GET /login, GET /static/style.css,
 # POST /login, POST /r/<o>/<n>/edit.
 #
 # Auth: PRAGENT_DASHBOARD_TOKEN in the pragent-webhook Secret, cookie
 # `pragent_dash=<token>`, single-user. Empty / unset = no auth (tailnet-only).
 #
 # NodePort 30082 — only reachable on the Tailscale / LAN side of kubernets
 # (100.74.17.70 / 192.168.1.80) until/unconfigured. Mirrors pragent-webhook.yaml
 # in every other respect (uid 10001, nodeSelector, /data PVC).
 apiVersion: apps/v1
 kind: Deployment
 metadata:
  name: pragent-dashboard
  namespace: pragent
  labels:
    app: pragent-dashboard
 spec:
  replicas: 1
  selector:
    matchLabels:
      app: pragent-dashboard
  template:
    metadata:
      labels:
        app: pragent-dashboard
    spec:
      # Same node as the webhook — holds the headroom proxy + the /data PVC.
      nodeSelector:
        kubernetes.io/hostname: kubernets
      # Dashboard is read-only over /data and only mutates Gitea (not local
      # files), so unprivileged is fine. fsGroup matches the image's USER
      # directive (10001) so the RO mount is readable.
      securityContext:
        runAsNonRoot: true
        runAsUser: 10001
        runAsGroup: 10001
        fsGroup: 10001
      containers:
        - name: dashboard
          image: pragent-webhook:optin
          imagePullPolicy: Never
          workingDir: /app
          command: ["python3", "-m", "pilot.dashboard"]
          ports:
            - name: http
              containerPort: 8081
          env:
            - name: PRAGENT_FEEDBACK_DB
              value: /data/feedback.db
            - name: PRAGENT_GITEA_API
              value: http://gitea-http.gitea.svc.cluster.local:3000
            # Dashboard reads DASHBOARD_PORT (not PORT) — verified in
            # pilot/dashboard.py:51. Default 8081 if unset.
            - name: DASHBOARD_PORT
              value: "8081"
            # Used by /r/<o>/<n>/edit to PUT updated JSON to the repo's
            # contents API. Reuses the same bot token the webhook uses.
            - name: PRAGENT_BOT_TOKEN
              valueFrom:
                secretKeyRef:
                  name: pragent-webhook
                  key: PRAGENT_BOT_TOKEN
            # Auth cookie value. Add to the pragent-webhook Secret with:
            #   kubectl patch secret pragent-webhook -n pragent --type=json \
            #     -p='[{"op":"add","path":"/data/PRAGENT_DASHBOARD_TOKEN","value":"<base64>"}]'
            - name: PRAGENT_DASHBOARD_TOKEN
              valueFrom:
                secretKeyRef:
                  name: pragent-webhook
                  key: PRAGENT_DASHBOARD_TOKEN
          # /data is read-only — the dashboard doesn't write the SQLite file;
          # .pr-review.json mutations go through the Gitea contents API, not
          # local fs. RO avoids any chance of two pods racing the same RWO PVC.
          volumeMounts:
            - name: feedback-data
              mountPath: /data
              readOnly: true
          # No /health route in dashboard.py (returns 404 on unknown paths).
          # Probes omitted intentionally — see pilot/dashboard.py:687-721.
          # Resources: dashboard is read-heavy + tiny writes. /data RO + no
          # subprocess fan-out (no opencode) keeps footprint small.
          resources:
            requests:
              cpu: 100m
              memory: 256Mi
            limits:
              cpu: 500m
              memory: 512Mi
      volumes:
        - name: feedback-data
          persistentVolumeClaim:
            claimName: pragent-feedback-data
 ---
 apiVersion: v1
 kind: Service
 metadata:
  name: pragent-dashboard
  namespace: pragent
 spec:
  selector:
    app: pragent-dashboard
  ports:
    - name: http
      port: 80
      targetPort: http
      nodePort: 31540
  type: NodePort
  # 31540 — auto-allocated at first apply (30082 was already taken by
  # habitsnow/habitsnow-proxy). Tailscale / LAN only until a Caddy route is set.
@@ -1,269 +0,0 @@
 # pragent pilot — central dashboard service
 A read-only overview + per-repo / per-PR drilldown over the same SQLite
 feedback DB the webhook writes, plus a small form to mutate `.pr-review.json`
 on a covered repo via the Gitea contents API. Companion to the
 [webhook service](README-webhook.md); reuses the webhook image
 (`pragent-webhook:dashboard`) — the pilot modules are baked into `/app/pilot/`,
 and the dashboard is just `python3 -m pilot.dashboard`.
 ## Architecture
 ```
 Browser
   │
   ▼
 Caddy (TLS, wildcard cert via Cloudflare DNS-01)
   │   https://pragent-dashboard.marcospaulo.dev.br → 100.74.17.70:31541
   ▼
 Service oauth2-proxy-dashboard.pragent.svc.cluster.local  (NodePort 31541, ns pragent)
   │
   │   oauth2-proxy fronts the dashboard, enforces Logto SSO + email allowlist
   │   sets X-Forwarded-User / X-Forwarded-Email on accepted requests
   ▼
 Service pragent-dashboard.pragent.svc.cluster.local  (ClusterIP, ns pragent)
   │
   ▼
 pragent-dashboard pod  (uid 10001, /data RO, no subprocess fan-out)
   │
   ├── read /data/feedback.db           (PVC pragent-feedback-data, RO)
   ├── GET  .../repos/{o}/{r}/...        (Gitea contents API, bot token)
   └── PUT  .../repos/{o}/{r}/contents/.pr-review.json
            (edit form submit; Gitea commits a new sha)
 ```
 Fail-soft. Nothing is ever written to local disk by the dashboard — the
 SQLite file is read-only and `.pr-review.json` mutations go through Gitea's
 contents API so the commit history records who changed what.
 The dashboard `Service` is **ClusterIP** — only oauth2-proxy can reach it.
 Public access is gated by Caddy (TLS termination) → oauth2-proxy (Logto SSO
 + allowlist) → dashboard.
 ## What it does
 - **Overview** (`GET /`): summary stats across all onboarded repos — total
  reviews, distinct PRs, finding counts by severity, false-positive /
  accepted-pattern scores (see "Feedback loop" in README-webhook.md), and a
  sparkline of review activity.
 - **Repo drilldown** (`GET /r/<owner>/<name>`): per-repo PRs with their
  last-review status, finding counts, and links to PR-level drilldowns.
 - **PR drilldown** (`GET /r/<owner>/<name>/<index>`): the bot's review(s)
  on that PR, inline findings, and reaction / resolved status harvested
  by `feedback_harvest.py`.
 - **Raw review** (`GET /r/<owner>/<name>/<index>/raw`): the markdown body
  of the most recent review, for copy-paste / diff-with-prose workflows.
 - **Edit form** (`POST /r/<owner>/<name>/edit`): a small HTML page that
  loads the current `.pr-review.json` from the repo's default branch and
  lets the operator edit the JSON (validated, then PUT to Gitea contents
  API). This is how repo-local `focus` / `instructions` /
  `reviewers` / `severity_threshold` get tuned per-repo after seeing
  the feedback roll-up.
 All routes return HTML (or plain text for `/raw`) with the same stylesheet
 (`/static/style.css`).
 ## Routes
 | method | path                              | auth | description                                  |
 |--------|-----------------------------------|------|----------------------------------------------|
 | GET    | `/`                               | yes  | Overview                                     |
 | GET    | `/static/style.css`               | no   | Stylesheet                                   |
 | GET    | `/r/<owner>/<name>`               | yes  | Repo drilldown                               |
 | GET    | `/r/<owner>/<name>/<index>`       | yes  | PR drilldown                                 |
 | GET    | `/r/<owner>/<name>/<index>/raw`   | yes  | Most recent review body as markdown          |
 | POST   | `/r/<owner>/<name>/edit`          | yes  | Edit `.pr-review.json` on the default branch |
 Auth is enforced by oauth2-proxy upstream; the dashboard itself only
 checks the `X-Forwarded-User` header that oauth2-proxy sets after a
 successful Logto login + email allowlist match.
 There is no `/health` route — don't add one to the k8s probes without
 updating `pilot/dashboard.py` (the handler returns 404 on unknown paths,
 so a probe would loop forever).
 ## Mutations flow through Gitea, not local fs
 The edit endpoint reads the current `.pr-review.json` from
 `GET /repos/{o}/{r}/contents/.pr-review.json?ref=<default-branch>`, lets
 the operator edit it in a form (validated as JSON, length-capped per
 field, no schema migration), and PUTs the new content back via the
 contents API with a commit message like
 `pragent dashboard: update .pr-review.json`. Every edit is a real Gitea
 commit on the default branch, attributable to `pragent-bot`, and the
 next webhook fire picks up the new config — no Pod restart, no image
 rebuild, no pod-level state.
 The `/data` mount is **read-only** (see the `readOnly: true` on the
 volumeMount in `~/k8s/pragent-dashboard.yaml`): the dashboard never
 writes the SQLite file, only the webhook + the daily cronjob do, and
 keeping it RO means a buggy deploy can't corrupt the harvested feedback.
 ## Auth (Logto SSO via oauth2-proxy)
 Authentication is delegated to oauth2-proxy, which fronts the dashboard
 in-cluster. The dashboard never sees a cookie or a token — it only
 inspects `X-Forwarded-User` (set by oauth2-proxy after a successful
 Logto login + email allowlist match). Missing header → 401 with
 `WWW-Authenticate: Basic realm="pragent-dashboard"`, which lets
 oauth2-proxy intercept and bounce the browser to Logto.
 Email allowlist lives in the ConfigMap `oauth2-proxy-dashboard-emails`
 in namespace `pragent`:
 ```yaml
 data:
  authenticated-emails: |
    marcos.paulodasilva.mp@gmail.com
    thiago@marcospaulo.dev.br
 ```
 Edit the ConfigMap to add/remove users; oauth2-proxy hot-reloads the
 file (it logs `watching ... for updates`), no restart needed. This is
 the same isolation pattern as the minecraft-sso / code-server
 allowlists — see `~/.claude/memory/minecraft-sso.md`.
 The Logto app is `pragent-dashboard` (tenant `default`, type
 `Traditional`), created by direct INSERT into Logto Postgres mirroring
 the proven `minecraft-sso` pattern. Credentials live in
 `~/k8s/oauth2-proxy-dashboard-secret.yaml` (mode 600, NOT in git).
 Public URL: **https://pragent-dashboard.marcospaulo.dev.br** (Caddy
 TLS termination via wildcard cert → Tailscale → NodePort 31541 →
 oauth2-proxy → dashboard ClusterIP).
 ### Emergency bypass (cookie)
 If Logto goes down and you need to access the dashboard before the
 oauth2-proxy restart dance (see `~/.claude/memory/logto-fix.md`),
 `pilot/dashboard.py` can be patched to accept a fallback cookie by
 re-adding the `PRAGENT_DASHBOARD_TOKEN` env path — the route gate is
 isolated in `_is_authed` and the logic is straightforward. The current
 commit intentionally has no bypass because Logto SSO is the single
 source of truth for "who can touch `.pr-review.json`".
 ## Deploy
 The dashboard shares the webhook image, so there's nothing to rebuild
 beyond what the webhook already does. After editing `pilot/dashboard.py`
 or `pilot/dashboard_data.py`, redo the webhook image rebuild + containerd
 import (see `README-webhook.md` § "K8s deployment") and roll both
 deployments.
 ```bash
 K="microk8s kubectl"
 # 1. (one-time) create the Logto app + cookie secret + oauth2-proxy
 #    See ~/.claude/memory/minecraft-sso.md for the SQL INSERT recipe
 #    and ~/k8s/oauth2-proxy-dashboard*.yaml for the manifests.
 # 2. apply all pragent-dashboard manifests (dashboard + oauth2-proxy)
 $K apply -f ~/k8s/oauth2-proxy-dashboard.yaml
 $K apply -f ~/k8s/pragent-dashboard.yaml
 # 3. roll on image / code changes
 $K -n pragent rollout restart deploy/pragent-dashboard
 $K -n pragent rollout status  deploy/pragent-dashboard --timeout=120s
 $K -n pragent logs -f deploy/pragent-dashboard
 ```
 K8s manifests:
 - `~/k8s/pragent-dashboard.yaml` — Deployment + ClusterIP Service.
  - `image: pragent-webhook:dashboard` + `imagePullPolicy: Never` —
    local containerd only, same image as the webhook.
  - `nodeSelector: kubernetes.io/hostname: kubernets` — pinned to the
    node holding the `/data` PVC.
  - `securityContext: runAsNonRoot: true, runAsUser: 10001, runAsGroup:
    10001, fsGroup: 10001` — matches the image's USER directive;
    fsGroup makes the RO hostpath volume readable.
  - `volumeMounts.feedback-data.readOnly: true` — dashboard is
    read-only over `/data`; mutations go through Gitea, not local fs.
  - No `readinessProbe` / `livenessProbe` — the dashboard has no
    `/health` route. If you add one to `pilot/dashboard.py`, add a
    probe here too.
  - `resources.requests: {cpu: 100m, memory: 256Mi}` /
    `limits: {cpu: 500m, memory: 512Mi}` — read-heavy + tiny writes,
    no opencode subprocess fan-out, much smaller than the webhook.
  - `Service.type: ClusterIP` — only oauth2-proxy can reach it.
 - `~/k8s/oauth2-proxy-dashboard.yaml` — Deployment + ConfigMap +
  NodePort Service (`oauth2-proxy-dashboard`, NodePort 31541,
  namespace `pragent`). Same shape as the code-server /
  minecraft-sso oauth2-proxy. NodePort 31541 was chosen because
  31540 was the old dashboard NodePort and the 30096..30969 media
  range + 30350-30351 (other oauth2-proxy NodePorts) were taken.
 - `~/k8s/oauth2-proxy-dashboard-secret.yaml` — client-id /
  client-secret / cookie-secret (mode 600, NOT in git).
 ## Smoke test
 ```bash
 # 1. anonymous request → 302 redirect to Logto
 curl -I https://pragent-dashboard.marcospaulo.dev.br/
 # 2. pod logs
 microk8s kubectl logs -n pragent -l app=oauth2-proxy-dashboard --tail=50
 microk8s kubectl logs -n pragent -l app=pragent-dashboard --tail=50
 # 3. in-cluster direct probe (should 401 without X-Forwarded-User)
 microk8s kubectl port-forward -n pragent svc/pragent-dashboard 8181:80 &
 sleep 2
 curl -I http://localhost:8181/                       # expect 401 + WWW-Authenticate: Basic
 curl -I -H "X-Forwarded-User: marcos@example.com" http://localhost:8181/  # expect 200
 kill %1
 ```
 The HTML returned with a valid `X-Forwarded-User` should contain a
 `<title>` (whatever the dashboard renders) and **never** `Traceback` or
 any Python exception output. A 401 on the unauthenticated GET is the
 expected behaviour — oauth2-proxy catches it and redirects to Logto.
 ## Threat model / security notes
 - **Behind Logto SSO.** Anonymous traffic gets 302 → Logto. Allowed
  emails (marcos, thiago) reach the dashboard after Logto login; all
  others see oauth2-proxy's "not authorized" page. Adding a user is a
  one-line ConfigMap edit; oauth2-proxy hot-reloads the allowlist.
 - **`PRAGENT_BOT_TOKEN` is Gitea Write scoped** to onboarded repos, so
  a successful auth bypass on the dashboard is Gitea repo write access,
  not just read. oauth2-proxy's email allowlist is the only
  authentication factor — there is no second factor. If this becomes a
  concern, swap oauth2-proxy for an IdP that supports TOTP/WebAuthn
  and the dashboard needs no further changes (it just reads the
  forwarded headers).
 - **CSRF on the edit form.** Per-process random secret embedded as a
  hidden input + double-submit via the `X-Forwarded-User` context. An
  attacker would need to (a) steal the user's Logto session cookie
  from oauth2-proxy and (b) read the rendered HTML to harvest the
  CSRF token. Both have to happen in the same browser.
 - **Read-only `/data` mount.** The dashboard can't corrupt the
  harvested SQLite file even if it's compromised. The webhook and the
  daily cronjob are the only writers.
 - **ClusterIP dashboard Service.** Even if a malicious actor discovered
  the dashboard's container port, they cannot reach it from outside the
  cluster — only oauth2-proxy can. NetworkPolicy is the cluster
  default deny.
 - **`uid 10001` + `runAsNonRoot: true`.** No host-level escalation if
  the dashboard is popped — it has no caps, no `/proc` mounts.
 - **No author-controlled input is `eval`-ed.** The edit form parses the
  JSON, validates types / lengths, and re-serialises before the Gitea
  PUT. The review-side hostile-input concerns from `README-webhook.md`
  § "Threat model" do **not** apply to the dashboard — the dashboard
  is a read-mostly viewer over already-harvested, already-posted data.
 ## Known limitations (pilot)
 - Logto SSO is the only auth factor — no per-user sessions, no CSRF
  token tied to a per-user identity (the per-process CSRF secret is
  global). Adequate for a single-operator dashboard; not adequate for
  multi-tenant.
 - No `/health` route — if the dashboard process wedges on a Gitea hang,
  k8s won't restart it. Add a `/health` route to `pilot/dashboard.py`
  + a probe here before relying on this in production.
 - The overview is a single-process render over a SQLite file that the
  daily cronjob also writes. A long Gitea hang during a page render can
  stall the dashboard until the client request times out (30 s). The
  underlying SQLite reader is read-only and concurrent-safe, so no
  data corruption — just a slow page.
@@ -75,103 +75,6 @@ regression baseline: re-run a candidate model over these PRs and the diff
 against this column is the behaviour change. Promoting an item to real ground
 truth means a human editing it in the dataset view after re-reading the PR.
 ### Item ids
 `{owner}__{repo}__pr{n}`. The obvious `{repo}#{pr}` cannot be used: items are
 routed as `/datasets/{id}/items/{item_id}`, so the `/` in `owner/repo` splits
 into extra path segments and everything after `#` is a fragment the browser
 never sends — the item is created fine by the API and then 404s when opened.
 Session ids elsewhere keep `{repo}#{pr}`; those are never path segments.
 ### Filterable metadata
 The filter bar matches on `metadata` only — not on `input`, and not on the item
 id — so every facet worth slicing on is a flat, primitive key in `metadata`
 even where it duplicates `input`:
 | key | why it is there |
 | --- | --- |
 | `repo`, `owner`, `repo_name` | `owner` exists because a filter on the joined `repo` matches one repo, never a whole org |
 | `pr`, `head_sha` | jump from a filtered row back to the actual PR |
 | `finding_count`, `has_findings` | isolate the silent reviews, which are the interesting negatives |
 | `max_severity` | `"none"` rather than absent — an absent key matches no filter |
 | `reviews_run` | how churny the PR was; high values skew per-item averages |
 | `last_reviewed_at` / `_iso` | epoch sorts, ISO reads |
 | `labelled_by_human` | `false` everywhere today; the flag to filter on before trusting any of it |
 Nested objects and lists are deliberately absent: the filter bar cannot reach
 into them.
 `max_severity` is derived from `feedback.db`, whose `severity` column is
 re-parsed out of the rendered comment by `feedback_harvest._parse_severity` and
 defaults to `INFO` when its regex misses the badge. Trust the `severity_max`
 **score** (read from the model's structured output) over this facet.
 ## Experiments
 `eval_experiment.py` links reviews that already ran into a dataset run, so the
 Experiments tab is populated without re-running anything. Runs are grouped by
 model — the comparison the pilot actually needs is the same PRs under a
 candidate model with `finding_rate` and `cost_per_finding` side by side. A new
 model produces a new run automatically on the next invocation.
 One trace per (run, item), the most recent: a PR re-reviewed on every push has
 many traces, and a run is one output per input.
 It uses `POST /api/public/dataset-run-items`, which is deprecated in favour of
 the SDK experiment runner and disappears in Langfuse v4. The deprecation notice
 exempts self-hosted v3 from the cutoff date, and this pilot is stdlib-only by
 design. Revisit when this deployment moves to v4.
 Coverage is bounded by the dataset, not by the traces: items only exist for PRs
 with a row in `feedback.db`, and a review that posted no comment leaves a trace
 but no row. That is why a run links fewer items than there are traces.
 ## Evaluators: `eval_judges.py`
 Behaviour scores answer "how many, how severe, how much" — computable from data
 already in hand. Two things they cannot answer:
 - **Was the finding any good?** Specificity vs. hedge, generic advice vs.
  fix-it-now advice — the difference between a useful review and one a
  developer scrolls past.
 - **Did the summary match the findings?** Claiming "no issues" above two
  criticals, or describing a problem in prose that never became a finding.
 These need a judge. `eval_judges.py` registers two `llm_as_judge` evaluators
 against the trace names this project emits (`pr-review`, `opencode-review`)
 and wires a sampling=1 rule per evaluator. Both run on every observation in a
 matching trace; the only observations in those traces are the review itself.
 | evaluator | output | what it answers |
 |---|---|---|
 | `finding_actionability` | NUMERIC 0–1 | How specific and fixable is each finding? |
 | `review_self_consistency` | BOOLEAN | Does the summary agree with the findings? |
 The judge is a different model from the reviewer (`kimi-k2.7-code` through the
 headroom hub). A model grading its own output agrees with itself for reasons
 that have nothing to do with quality. The judges are also asked only what they
 can answer from the review itself — never whether a finding is correct, since
 that needs the diff the trace does not carry.
 ### Why the judge goes through `judge-proxy` (port 8802)
 The headroom hub in front of local Ollama returns Anthropic-format responses,
 but every `thinking` content block is missing the `signature` field real
 Claude emits. Langfuse's Zod schema requires it; the omission fails the
 evaluator preflight as `Invalid JSON response`. The `judge-proxy` pod sits in
 front of the hub on `100.74.17.70:8802` and patches every thinking block with
 a synthetic signature before forwarding the response. The model is unchanged;
 only the wire shape is fixed.
 ```bash
 python3 pilot/eval_judges.py --dry-run   # show what would be created
 python3 pilot/eval_judges.py             # create the LLM connection, evaluators, rules
 ```
 Idempotent: existing evaluators and rules are skipped, not duplicated. The
 connection is upserted on `provider` so re-runs return the same record.
 ## Running it
 ```bash
@@ -180,10 +83,6 @@ python3 pilot/eval_bootstrap.py --db /data/feedback.db --backfill-traces
 # ship feedback verdicts (runs daily from the feedback CronJob)
 python3 pilot/feedback_scores.py --db /data/feedback.db
 # link already-traced reviews into a dataset run per model
 python3 pilot/eval_experiment.py --dry-run
 python3 pilot/eval_experiment.py
 ```
 Both need `LANGFUSE_HOST`, `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY`. In
@@ -21,7 +21,7 @@ and lands on the trace's `environment`:
 | `headroom/MiniMax-M2.7`     | `ollama`    |
 | `vllm-qwen38/qwen3.8-27b`   | `ollama`    |
-Langfuse takes an environment selector on every dashboard, filter and cost
+Langfuse takes an environment selector on every view, filter and cost
 breakdown, so the two spend stories stay separate inside one project — one key
 pair to rotate instead of two. Tags carry the finer cut:
 `provider:headroom`, `model:<bare>`, `engine:opencode`, `repo:<owner/name>`,
@@ -56,11 +56,11 @@ A model that costs nothing through the headroom proxy is priced against a
 **comparison target** instead: basis `equivalent:<target>`. That covers the
 models absent from `PRICES` (`MiniMax-M2.7` — which is what the webhook
 actually runs — and `glm-5.2:cloud`) as well as entries priced at all zeros
-(the self-hosted vLLM `qwen3.8-27b`). Without this the dashboard would be a
+(the self-hosted vLLM `qwen3.8-27b`). Without this Langfuse would show a
 flat $0.00 line, since the pilot's own path is free.
-The target follows the same precedence as the review body, so the PR and the
+The target follows the same precedence as the review body, so the PR and
-dashboard never disagree:
+Langfuse never disagree:
    .pr-review.json:cost_target  >  PRAGENT_PRICE_TARGET  >  claude-sonnet-5
@@ -1,100 +1,76 @@
-# pragent pilot — AI Review bot
+# pragent pilot
-A minimal AI code-review bot for Gitea, running as a CI step on the existing
+The pilot is a central, stdlib-only Gitea webhook service. It reviews opted-in
-`act-runner`. This is the **pilot** — a small, self-contained reviewer that
+pull requests with an on-network model, posts inline findings, and emits review
-predates the full `pragent` framework (whose design lives in
+telemetry to Langfuse. The service is fail-open: a review failure is reported
-`docs/plans/2026-08-04-pragent-design.md`). The framework will later absorb
+as a PR comment and does not block CI.
 this; until then, this is what runs.
-## How it works
+## Runtime flow
-1. You add `pragent-bot` to a repo and commit `.gitea/workflows/ai-review.yml`.
+1. Gitea sends a signed `pull_request` webhook.
-2. On a PR, you add the **`AI-REVIEW`** label.
+2. `webhook_server.py` validates the request, checks the base branch's
-3. Gitea Actions runs the workflow on the `act-runner`; it fetches the PR diff,
+   `.pr-review.json` for `"enabled": true`, and claims `(repo, PR, SHA)`.
-   asks `glm-5.2:cloud` (on-network via the headroom proxy) to review it, and
+3. `ai_review.review_pr()` fetches the diff, trusted config, and prior reviews.
-   posts the findings back as a PR review authored by `pragent-bot`.
+4. `opencode_review.py` checks out the PR head in a sanitized temporary
-4. Remove the label to stop re-reviews on further pushes.
+   directory and runs the review agent. The legacy Ollama-compatible path is
   still available through `PRAGENT_ENGINE`.
 5. The review output is parsed and normalized, valid post-change line anchors
   are separated from summary-only findings, and Gitea receives the result.
 6. `langfuse_trace.py` records usage, cost basis, findings, and evaluation
   scores when Langfuse credentials are configured.
-Fail-open: the job always exits 0 and never blocks CI. Errors become a short
+## Module map
 "review failed" comment.
-## Onboard a repo (3 steps)
+| Module | Responsibility |
 ### 1. Add `pragent-bot` as collaborator
 Repo → Settings → Collaborators → Add → `pragent-bot` → permission **Write**.
 (Write is required to post reviews/comments.)
 Or via API (with an admin/owner token):
 ```bash
 curl -X PUT -H "Authorization: token $OWNER_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"permission":"write"}' \
  "http://<gitea-host>:3000/api/v1/repos/OWNER/REPO/collaborators/pragent-bot"
 ```
 ### 2. Add the `PRAGENT_BOT_TOKEN` secret
 Repo → Settings → Actions → Secrets → New secret → name `PRAGENT_BOT_TOKEN`,
 value = the bot's access token (ask the platform admin; stored mode-600 at
 `~/.claude/.pragent-bot-token` on the admin host).
 ### 3. Commit the workflow
 Copy `pilot/workflow-template.yml` into the target repo as
 `.gitea/workflows/ai-review.yml` and commit it. That's it.
 ## Use it
 Open a PR (or push to an open one), add the **`AI-REVIEW`** label. The review
 appears within ~30–90s depending on diff size and model latency.
 ## What's intentionally NOT in the pilot
 Deferred to the full framework (by design, see the design doc):
 - Attention tiering (trivial/lite/full/oversized) and per-tier cost control.
 - Multiple analyzer fan-out over a shared cached prompt prefix.
 - Prior-comment synthesis (so each push re-posts; the latest review is tagged
  with the head SHA so it's easy to spot).
 - Inline line comments and status checks.
 - `pragent explain` / `replay` / analytics JSONL.
 - A second forge (GitLab) and the provider matrix.
 ## Pieces
 | File | Role |
 |---|---|
-| `pilot/ai_review.py` | The reviewer script (stdlib only). Single source of truth — fetched at runtime by each repo's workflow. |
+| `webhook_server.py` | HTTP ingress, signature verification, opt-in gate, concurrency |
-| `pilot/workflow-template.yml` | The Gitea Action consumers copy into `.gitea/workflows/ai-review.yml`. |
+| `review_config.py` | Trusted base-branch opt-in policy; transport injected for tests |
-| `tests/pilot/test_ai_review.py` | Unit tests for the pure helpers (no network). |
+| `gitea_client.py` | HTTP transport adapter and repository-scoped client |
 | `ai_review.py` | Compatibility facade and review orchestration |
 | `model_client.py` | Anthropic-compatible model adapter and response text extraction |
 | `opencode_review.py` | Hostile-checkout containment and agent execution |
 | `diff_compress.py` | Diff compression and prior-review extraction |
 | `feedback*.py` | Feedback persistence, harvesting, analysis, and Langfuse scores |
 | `langfuse_trace.py` | Fail-open Langfuse ingestion and cost metadata |
 | `cost_model.py` | Provider price catalog and equivalent-cost calculations |
 | `eval_*.py` | Dataset bootstrap, evaluators, and behavioral scoring |
-## Run the tests
+`ai_review.py` remains the stable import surface for existing workflow and
 webhook deployments. New code should put policy, adapters, and pure transforms
 in the focused modules above rather than adding unrelated functions there.
-```bash
+## Onboard a repository
-cd ~/Projects/pragent
+
-PYTHONPATH=pilot python3 -m pytest tests/pilot/    # if pytest available
+1. Add `pragent-bot` as a Write collaborator.
-# or, without pytest:
+2. Commit this file to the default branch:
-python3 - <<'PY'
+
-import os, sys, importlib.util
+   ```json
-sys.path.insert(0, os.path.abspath("pilot"))
+   {"enabled": true}
 import ai_review  # noqa: F401
 spec = importlib.util.spec_from_file_location("t", "tests/pilot/test_ai_review.py")
 m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
 fails = 0
 for n in sorted(x for x in dir(m) if x.startswith("test_")):
    try: getattr(m, n)(); print("PASS", n)
    except Exception as e: fails += 1; print("FAIL", n, e)
 print("failed:", fails)
 PY
   ```
-## Configuration knobs (env in the workflow)
+3. Open or update a pull request.
-| Env | Default | Purpose |
+No per-repository workflow, secret, or label is required for the central
-|---|---|---|
+webhook path. See [`README-webhook.md`](README-webhook.md) for deployment,
-| `OLLAMA_MODEL` | `glm-5.2:cloud` | Model id passed to the headroom proxy. |
+security, and webhook registration details.
-| `OLLAMA_MAX_TOKENS` | `6000` | Output token cap. |
+
-| `DIFF_MAX_CHARS` | `150000` | Diff truncation cap (with a noted truncation marker). |
+## Configuration
-| `OLLAMA_URL` | `http://<model-proxy-host>:8789` | headroom proxy (tailnet). If the act-runner can't reach the tailnet IP, expose 8789 as an in-cluster Service+Endpoints and set this to the cluster DNS name. |
+
 | Variable | Default | Purpose |
 |---|---:|---|
 | `GITEA_API` | in-cluster URL | Gitea API base URL |
 | `PRAGENT_BOT_TOKEN` | — | Bot credential |
 | `OLLAMA_URL` / `OLLAMA_MODEL` | headroom / `glm-5.2:cloud` | Legacy model path |
 | `PRAGENT_ENGINE` | `opencode` | `opencode` or legacy model path |
 | `DIFF_MAX_CHARS` | `150000` | Diff input cap |
 | `PRAGENT_MAX_CONCURRENT_REVIEWS` | `2` | Process concurrency bound |
 | `LANGFUSE_HOST` + keys | unset | Enables telemetry; unset is a no-op |
 ## Tests
 ```bash
 python3 -m pytest tests -q
 ```
 Tests use mocked transports and local fixtures. They do not require Gitea,
 Langfuse, a model endpoint, or network access.
@@ -185,15 +185,8 @@ def parse_text_blocks(content: list) -> str:
    Drops `thinking` blocks (glm-5.2:cloud is a reasoning model and emits them).
    Tolerates missing/malformed blocks by skipping them.
    """
-    if not isinstance(content, list):
+    from model_client import parse_text_blocks as _parse_text_blocks
-        return ""
+    return _parse_text_blocks(content)
    out = []
    for block in content:
        if not isinstance(block, dict):
            continue
        if block.get("type") == "text" and isinstance(block.get("text"), str):
            out.append(block["text"])
    return "\n".join(out).strip()
 def _int_env(name: str, default: int) -> int:
@@ -1803,19 +1796,8 @@ def compact_prior_reviews(prior_bodies: list[str]) -> list[str]:
 def _http(method: str, url: str, token: str, body: dict | None = None, accept: str = "application/json") -> tuple[int, bytes]:
-    headers = {"Authorization": f"token {token}", "Accept": accept}
+    from gitea_client import request
-    data = None
+    return request(method, url, token, body, accept)
    if body is not None:
        data = json.dumps(body).encode()
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=180) as r:
            return r.status, r.read()
    except urllib.error.HTTPError as e:
        return e.code, e.read()
    except urllib.error.URLError as e:
        raise RuntimeError(f"network error: {e.reason}") from e
 def gitea_get(api: str, repo: str, path: str, token: str, accept: str = "application/json") -> tuple[int, bytes]:
@@ -2008,22 +1990,8 @@ def fetch_repo_config(api: str, repo: str, token: str, ref: str = "") -> dict:
 def call_model(ollama_url: str, model: str, system: str, user: str, max_tokens: int) -> str:
-    payload = {
+    from model_client import complete
-        "model": model,
+    return complete(ollama_url, model, system, user, max_tokens)
        "max_tokens": max_tokens,
        "system": system,
        "messages": [{"role": "user", "content": user}],
    }
    status, raw = _http(
        "POST",
        f"{ollama_url.rstrip('/')}/v1/messages",
        "ollama",  # headroom ollama hub uses x-api-key: ollama
        payload,
    )
    if status != 200:
        raise RuntimeError(f"model call failed: HTTP {status}: {raw[:500].decode('utf-8', errors='replace')}")
    data = json.loads(raw)
    return parse_text_blocks(data.get("content", []))
 def post_review(api: str, repo: str, index: str, token: str, body: str) -> None:
@@ -1,754 +0,0 @@
 #!/usr/bin/env python3
 """pragent pilot — read-mostly dashboard.
 Stdlib HTTP server (mirrors `webhook_server.py`'s BaseHTTPRequestHandler +
 ThreadingHTTPServer shape) that renders three views off the feedback SQLite:
  GET /                                    overview
  GET /r/<owner>/<name>                    repo summary + edit form
  GET /r/<owner>/<name>/<index>            one PR's findings
  GET /r/<owner>/<name>/<index>/raw        raw Markdown body (via Gitea)
  GET /static/style.css                    CSS
  POST /r/<owner>/<name>/edit              mutate .pr-review.json (Tasks C+D)
 Auth: oauth2-proxy fronts this service in-cluster. Every route except
 `/static/*` requires the `X-Forwarded-User` header (set by oauth2-proxy
 once the user has logged in via Logto). Missing header → 401 +
 `WWW-Authenticate: Basic realm="pragent-dashboard"` so oauth2-proxy
 intercepts the response.
 DB: `PRAGENT_FEEDBACK_DB` points at the SQLite file the webhook server
 also writes. Per-request open (SQLite is cheap, no concurrency hazard,
 no stale-conn surprise after the file rotates).
 All HTML is rendered via `string.Template` and every dynamic value is
 escaped with `html.escape(..., quote=True)`. No `.format`, no f-string
 templates — see `_render_*` for the discipline.
 """
 from __future__ import annotations
 import base64
 import datetime
 import html
 import json
 import os
 import secrets
 import string
 import urllib.error
 import urllib.parse
 import urllib.request
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pilot import dashboard_data
 # ---------------------------------------------------------------------------
 # Config
 # ---------------------------------------------------------------------------
 FEEDBACK_DB = ""  # legacy; readers should call _feedback_db()
 PORT = int(os.environ.get("DASHBOARD_PORT", "8081"))
 GITEA_API = ""  # legacy; readers should call _gitea_api()
 BOT_TOKEN = ""  # legacy; readers should call _bot_token()
 # CSRF secret for the edit form. Regenerated per process (each Python
 # interpreter launch). Behind oauth2-proxy this is enough — only an
 # already-authenticated same-tab request can read this and echo it back.
 _CSRF_SECRET: str = secrets.token_urlsafe(24)
 # ---------------------------------------------------------------------------
 # Lazy config readers — tests set env after import, so each request re-reads.
 # Production: env is fixed for the process lifetime; the per-request lookup is
 # a dict access, not a syscall.
 # ---------------------------------------------------------------------------
 def _feedback_db() -> str:
    return os.environ.get("PRAGENT_FEEDBACK_DB", "")
 def _bot_token() -> str:
    return os.environ.get("PRAGENT_BOT_TOKEN", "")
 def _gitea_api() -> str:
    return os.environ.get("GITEA_API", "http://gitea-http.gitea.svc.cluster.local:3000")
 # ---------------------------------------------------------------------------
 # Stylesheet — small, dark-mode-friendly, deliberately under 100 lines
 # ---------------------------------------------------------------------------
 STYLE_CSS = """
 :root { color-scheme: light dark; }
 * { box-sizing: border-box; }
 body {
  font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", system-ui, sans-serif;
  margin: 0; padding: 0;
  background: #0f1115; color: #e6e6e6;
  line-height: 1.5;
 }
 header {
  background: #1a1d23; padding: 12px 20px;
  border-bottom: 1px solid #2a2f38;
  display: flex; align-items: center; gap: 18px;
 }
 header h1 { font-size: 18px; margin: 0; }
 header nav a {
  color: #8ab4f8; text-decoration: none; margin-right: 12px;
 }
 header nav a:hover { text-decoration: underline; }
 main { padding: 20px; max-width: 1100px; margin: 0 auto; }
 h2 { margin-top: 24px; font-size: 16px; color: #c9d1d9; }
 .metric-row { display: flex; gap: 16px; flex-wrap: wrap; margin-bottom: 16px; }
 .metric {
  background: #1a1d23; padding: 14px 18px; border-radius: 8px;
  min-width: 140px; border: 1px solid #2a2f38;
 }
 .metric .v { font-size: 28px; font-weight: 600; }
 .metric .l { font-size: 12px; color: #8b949e; text-transform: uppercase; letter-spacing: 0.04em; }
 table { width: 100%; border-collapse: collapse; margin: 8px 0 16px; font-size: 14px; }
 th, td { text-align: left; padding: 6px 10px; border-bottom: 1px solid #2a2f38; }
 th { color: #8b949e; font-weight: 500; text-transform: uppercase; font-size: 11px; letter-spacing: 0.04em; }
 tr:hover td { background: #161922; }
 .sev-critical { color: #ff7b72; font-weight: 600; }
 .sev-high { color: #f0883e; }
 .sev-medium { color: #d29922; }
 .sev-low { color: #8b949e; }
 .muted { color: #8b949e; font-size: 12px; }
 .sparkline { font-family: ui-monospace, "SF Mono", monospace; letter-spacing: 1px; }
 form { background: #1a1d23; padding: 14px 18px; border-radius: 8px; border: 1px solid #2a2f38; margin: 12px 0; }
 form label { display: block; margin: 8px 0 4px; color: #c9d1d9; font-size: 13px; }
 form input[type=text], form textarea, form select {
  background: #0f1115; color: #e6e6e6; border: 1px solid #2a2f38;
  border-radius: 4px; padding: 6px 8px; font-family: inherit; font-size: 14px;
  width: 100%;
 }
 form textarea { min-height: 80px; }
 form .row { display: flex; gap: 8px; align-items: center; margin-top: 12px; }
 form button {
  background: #2ea043; color: white; border: none; border-radius: 4px;
  padding: 6px 14px; font-size: 14px; cursor: pointer;
 }
 form button:hover { background: #3fb950; }
 .flash { background: #3d1e1e; color: #ff7b72; padding: 8px 12px; border-radius: 4px; margin-bottom: 12px; }
 code { background: #161922; padding: 1px 4px; border-radius: 3px; font-size: 13px; }
 pre { background: #161922; padding: 12px; border-radius: 6px; overflow-x: auto; }
 """
 # ---------------------------------------------------------------------------
 # Templates — string.Template so dynamic values are always escaped explicitly
 # ---------------------------------------------------------------------------
 _BASE = string.Template("""<!doctype html>
 <html lang="en">
 <head>
 <meta charset="utf-8">
 <title>${title}</title>
 <link rel="stylesheet" href="/static/style.css">
 </head>
 <body>
 <header>
  <h1>pragent dashboard</h1>
  <nav>
    <a href="/">Home</a>
    <a href="/r/${repos_first}">repos</a>
  </nav>
  <span class="muted" style="margin-left:auto">${db_status}</span>
 </header>
 <main>
 ${body}
 </main>
 </body>
 </html>""")
 _OVERVIEW = string.Template("""<h2>Overview</h2>
 <div class="metric-row">
  <div class="metric"><div class="v">${total_reviews}</div><div class="l">reviews</div></div>
  <div class="metric"><div class="v">${total_findings}</div><div class="l">findings</div></div>
  <div class="metric"><div class="v">${total_repos}</div><div class="l">repos</div></div>
  <div class="metric"><div class="v">${last_30d_reviews}</div><div class="l">last 30d</div></div>
 </div>
 <h2>Last 7 days</h2>
 <div class="sparkline">${sparkline}</div>
 <div class="muted">total cost: $${total_cost_usd} — no per-review cost logged</div>
 <h2>Top repos</h2>
 ${top_repos_table}
 """)
 _REPO = string.Template("""<h2>Repo: <code>${repo}</code></h2>
 <div class="metric-row">
  <div class="metric"><div class="v">${total_runs}</div><div class="l">runs</div></div>
  <div class="metric"><div class="v">${sev_critical}</div><div class="l sev-critical">critical</div></div>
  <div class="metric"><div class="v">${sev_high}</div><div class="l sev-high">high</div></div>
  <div class="metric"><div class="v">${sev_medium}</div><div class="l sev-medium">medium</div></div>
  <div class="metric"><div class="v">${sev_low}</div><div class="l sev-low">low</div></div>
 </div>
 <h2>Edit .pr-review.json</h2>
 ${flash}
 <form method="post" action="/r/${repo_url}/edit">
  <input type="hidden" name="_csrf" value="${csrf}">
  <label for="static_message">Static banner message (max 400 chars)</label>
  <textarea id="static_message" name="static_message" maxlength="400">${current_static_message}</textarea>
  <label for="model">Model (PRICES keys)</label>
  <select id="model" name="model">${model_options}</select>
  <div class="row">
    <button type="submit">Save</button>
    <span class="muted">posted via the bot identity; one commit on the base branch</span>
  </div>
 </form>
 <h2>Top findings (by occurrence)</h2>
 ${top_findings_table}
 <h2>Runs by day (last 30d)</h2>
 ${runs_by_day_table}
 <h2>Reviews</h2>
 ${reviews_table}
 """)
 _PR = string.Template("""<h2>PR <code>${repo}</code> #${pr}</h2>
 <div class="muted">head sha: <code>${head_sha}</code></div>
 <div class="muted">posted_at: ${posted_at_iso}</div>
 <div class="muted">review_id_gitea: ${review_id_gitea} · body_comment_id: ${body_comment_id}</div>
 <h2>Findings</h2>
 ${findings_table}
 <p><a href="/r/${repo_url}/${pr}/raw">raw review body (Markdown)</a></p>
 """)
 # ---------------------------------------------------------------------------
 # Small helpers
 # ---------------------------------------------------------------------------
 def _esc(s) -> str:
    """HTML-escape any value to a string."""
    return html.escape(str(s), quote=True)
 def _ts_iso(ts: int) -> str:
    if not ts:
        return "—"
    return datetime.datetime.fromtimestamp(int(ts), tz=datetime.timezone.utc).isoformat()
 def _sparkline(buckets: list[dict]) -> str:
    """7-bucket sparkline as unicode bars."""
    bars = "▁▂▃▄▅▆▇█"
    if not buckets:
        return ""
    mx = max((b.get("count", 0) for b in buckets), default=0) or 1
    out = []
    for b in buckets:
        n = b.get("count", 0)
        idx = min(len(bars) - 1, int(round(n / mx * (len(bars) - 1))))
        out.append(bars[idx])
    return "".join(out)
 # ---------------------------------------------------------------------------
 # Renderers — one per page
 # ---------------------------------------------------------------------------
 def _overview_body(data: dict) -> str:
    top_rows = "".join(
        f"<tr><td><a href=\"/r/{_esc(r['repo'])}\">{_esc(r['repo'])}</a></td>"
        f"<td>{int(r['run_count'])}</td>"
        f"<td class=\"muted\">{_ts_iso(int(r['last_seen']))}</td></tr>"
        for r in data.get("top_repos", [])
    ) or "<tr><td class=\"muted\">no reviews yet</td></tr>"
    top_table = f"<table><thead><tr><th>repo</th><th>runs</th><th>last seen</th></tr></thead><tbody>{top_rows}</tbody></table>"
    return _OVERVIEW.substitute(
        total_reviews=_esc(data.get("total_reviews", 0)),
        total_findings=_esc(data.get("total_findings", 0)),
        total_repos=_esc(data.get("total_repos", 0)),
        last_30d_reviews=_esc(data.get("last_30d_reviews", 0)),
        sparkline=_esc(_sparkline(data.get("daily", []))),
        total_cost_usd=f"{float(data.get('total_cost_usd', 0.0)):.2f}",
        top_repos_table=top_table,
    )
 def _repo_body(data: dict, *, repo_url: str, csrf: str, current_model: str,
               current_static_message: str, flash: str = "") -> str:
    fbs = data.get("findings_by_severity", {})
    tf = data.get("top_findings", [])
    # Top findings table.
    if tf:
        rows = "".join(
            f"<tr><td><code>{_esc(f['path'])}:{_esc(f['line'])}</code></td>"
            f"<td class=\"sev-{_esc(f.get('severity', 'low').lower())}\">{_esc(f.get('severity', ''))}</td>"
            f"<td>{_esc(f.get('problem', ''))}</td>"
            f"<td>{int(f.get('occurrences', 0))}</td>"
            f"<td>+{int(f.get('upvotes', 0))} / -{int(f.get('downvotes', 0))}</td>"
            f"<td>{'resolved' if int(f.get('resolved', 0)) else 'open'}</td>"
            f"<td>{int(f.get('reply_count', 0))}</td></tr>"
            for f in tf
        )
        top_findings_table = (
            "<table><thead><tr><th>location</th><th>severity</th>"
            "<th>problem</th><th>occurrences</th><th>votes</th>"
            "<th>state</th><th>replies</th></tr></thead><tbody>"
            f"{rows}</tbody></table>"
        )
    else:
        top_findings_table = "<p class=\"muted\">no findings yet</p>"
    # Runs by day.
    runs = data.get("runs_by_day", [])
    if runs:
        rows = "".join(
            f"<tr><td>{_esc(r['date'])}</td><td>{int(r.get('count', 0))}</td></tr>"
            for r in runs
        )
        runs_by_day_table = (
            "<table><thead><tr><th>date</th><th>runs</th></tr></thead>"
            f"<tbody>{rows}</tbody></table>"
        )
    else:
        runs_by_day_table = "<p class=\"muted\">no runs in the last 30 days</p>"
    # Reviews list — derived from finding timestamps; cheap because we
    # just enumerate the repo's review rows.
    reviews_table = _repo_reviews_table(repo_url, data.get("recent_reviews", []))
    # Model select (Task D) — sorted PRICES keys + "keep current".
    from cost_model import PRICES  # local: pilot-only dep
    model_options = (
        f"<option value=\"\">— keep current ({_esc(current_model or 'unset')}) —</option>"
        + "".join(
            f"<option value=\"{_esc(k)}\" {'selected' if k == current_model else ''}>{_esc(k)}</option>"
            for k in sorted(PRICES)
        )
    )
    return _REPO.substitute(
        repo=_esc(data.get("repo", "")),
        repo_url=_esc(repo_url),
        total_runs=_esc(data.get("total_runs", 0)),
        sev_critical=_esc(fbs.get("critical", 0)),
        sev_high=_esc(fbs.get("high", 0)),
        sev_medium=_esc(fbs.get("medium", 0)),
        sev_low=_esc(fbs.get("low", 0)),
        csrf=_esc(csrf),
        current_static_message=_esc(current_static_message),
        model_options=model_options,
        flash=_esc(flash),
        top_findings_table=top_findings_table,
        runs_by_day_table=runs_by_day_table,
        reviews_table=reviews_table,
    )
 def _repo_reviews_table(repo_url: str, rows: list[dict]) -> str:
    if not rows:
        return "<p class=\"muted\">no reviews yet</p>"
    out = "<table><thead><tr><th>PR</th><th>head sha</th><th>posted</th></tr></thead><tbody>"
    for r in rows:
        out += (
            f"<tr><td><a href=\"/r/{_esc(repo_url)}/{int(r['pr'])}\">#{int(r['pr'])}</a></td>"
            f"<td><code>{_esc(r['head_sha'][:10])}</code></td>"
            f"<td class=\"muted\">{_ts_iso(int(r.get('posted_at', 0)))}</td></tr>"
        )
    out += "</tbody></table>"
    return out
 def _pr_body(data: dict, *, repo_url: str) -> str:
    findings = data.get("findings", [])
    if findings:
        rows = "".join(
            f"<tr><td><code>{_esc(f['path'])}:{_esc(f['line'])}</code></td>"
            f"<td class=\"sev-{_esc(f.get('severity', 'low').lower())}\">{_esc(f.get('severity', ''))}</td>"
            f"<td>{_esc(f.get('problem', ''))}</td>"
            f"<td>{_esc(f.get('fix', ''))}</td>"
            f"<td>{_esc(f.get('suggestion', ''))}</td>"
            f"<td>+{int(f.get('upvotes', 0))} / -{int(f.get('downvotes', 0))}</td>"
            f"<td>{'resolved' if int(f.get('resolved', 0)) else 'open'}</td>"
            f"<td>{int(f.get('reply_count', 0))}</td></tr>"
            for f in findings
        )
        findings_table = (
            "<table><thead><tr><th>location</th><th>severity</th>"
            "<th>problem</th><th>fix</th><th>suggestion</th>"
            "<th>votes</th><th>state</th><th>replies</th></tr></thead>"
            f"<tbody>{rows}</tbody></table>"
        )
    else:
        findings_table = "<p class=\"muted\">no findings</p>"
    return _PR.substitute(
        repo=_esc(data.get("repo", "")),
        repo_url=_esc(repo_url),
        pr=_esc(data.get("pr", 0)),
        head_sha=_esc(data.get("head_sha", "")),
        posted_at_iso=_ts_iso(int(data.get("posted_at", 0))),
        review_id_gitea=_esc(data.get("review_id_gitea", "") or "—"),
        body_comment_id=_esc(data.get("body_comment_id", "") or "—"),
        findings_table=findings_table,
    )
 def _page(title: str, body: str, *, repos_first: str = "") -> str:
    db_status = _feedback_db() or "(no DB configured)"
    return _BASE.substitute(
        title=_esc(title),
        body=body,
        repos_first=_esc(repos_first),
        db_status=_esc(db_status),
    )
 # ---------------------------------------------------------------------------
 # Gitea HTTP helper — minimal, used by the raw body fetch and the edit endpoint
 # ---------------------------------------------------------------------------
 def _http(method: str, url: str, *, token: str = "", body: dict | None = None,
          raw_body: bytes | None = None) -> tuple[int, bytes]:
    """Like ai_review._http but local: this module is stdlib-only and doesn't
    depend on the ai_review import (which pulls in a 1700-line reviewer)."""
    headers = {"Accept": "application/json"}
    data: bytes | None = None
    if raw_body is not None:
        data = raw_body
        headers["Content-Type"] = "application/json"
    elif body is not None:
        data = json.dumps(body).encode()
        headers["Content-Type"] = "application/json"
    if token:
        headers["Authorization"] = f"token {token}"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return r.status, r.read()
    except urllib.error.HTTPError as e:
        return e.code, e.read()
    except urllib.error.URLError as e:
        raise RuntimeError(f"network error: {e.reason}") from e
 # ---------------------------------------------------------------------------
 # Auth
 # ---------------------------------------------------------------------------
 def _is_authed(headers) -> bool:
    """True when oauth2-proxy forwarded a verified user.
    oauth2-proxy sets `X-Forwarded-User` (and friends) only after a
    successful Logto login + email allowlist check. Unauthenticated
    requests never see the header, so the dashboard never has to know
    about cookies, secrets, or Logto's token shape.
    """
    return bool((headers.get("X-Forwarded-User") or "").strip())
 # ---------------------------------------------------------------------------
 # Routes
 # ---------------------------------------------------------------------------
 def _route_overview() -> bytes:
    data = dashboard_data.overview(_feedback_db())
    body = _overview_body(data)
    # nav: first repo if any
    repos_first = ""
    if data.get("top_repos"):
        repos_first = data["top_repos"][0]["repo"]
    return _page("Overview", body, repos_first=repos_first).encode()
 def _route_repo(owner: str, name: str) -> bytes:
    repo_url = f"{owner}/{name}"
    data = dashboard_data.repo_summary(_feedback_db(), repo_url)
    # Pull current .pr-review.json (best-effort) so the form fields prefill.
    current_static_message, current_model, flash = "", "", ""
    cfg, err = _fetch_pr_review_json(repo_url)
    if cfg:
        current_static_message = cfg.get("static_message", "")
        current_model = cfg.get("model", "")
    elif err and err != "404":
        flash = f"could not read .pr-review.json: {err}"
    body = _repo_body(
        data,
        repo_url=repo_url,
        csrf=_CSRF_SECRET,
        current_model=current_model,
        current_static_message=current_static_message,
        flash=flash,
    )
    return _page(f"repo {repo_url}", body, repos_first=repo_url).encode()
 def _route_pr(owner: str, name: str, index: int) -> bytes:
    repo_url = f"{owner}/{name}"
    data = dashboard_data.pr_summary(_feedback_db(), repo_url, int(index))
    body = _pr_body(data, repo_url=repo_url)
    return _page(f"PR {repo_url}#{index}", body, repos_first=repo_url).encode()
 def _route_pr_raw(owner: str, name: str, index: int) -> tuple[int, bytes]:
    repo_url = f"{owner}/{name}"
    data = dashboard_data.pr_summary(_feedback_db(), repo_url, int(index))
    body_comment_id = data.get("body_comment_id")
    if not body_comment_id:
        return 404, b"no body_comment_id"
    status, raw = _http(
        "GET",
        f"{_gitea_api()}/api/v1/repos/{repo_url}/issues/{index}/comments/{body_comment_id}",
        token=_bot_token(),
    )
    if status != 200:
        return 404, f"Gitea returned {status}".encode()
    try:
        parsed = json.loads(raw)
        md = parsed.get("body", "")
    except (json.JSONDecodeError, ValueError):
        return 404, b"could not parse Gitea response"
    return 200, md.encode()
 def _route_static_css() -> bytes:
    return STYLE_CSS.encode()
 def _route_edit(owner: str, name: str, form: dict) -> tuple[int, dict, bytes]:
    """Mutate .pr-review.json via the Gitea contents API (Tasks C+D)."""
    repo_url = f"{owner}/{name}"
    csrf = form.get("_csrf", "")
    if csrf != _CSRF_SECRET:
        return 302, {"Location": f"/r/{repo_url}"}, b""
    static_message = (form.get("static_message") or "").strip()[:400]
    model = (form.get("model") or "").strip()
    # Validate model against PRICES.
    from cost_model import PRICES
    if model and model not in PRICES:
        flash = urllib.parse.quote(f"unknown model {model!r}; not saved")
        return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b""
    cfg, err = _fetch_pr_review_json(repo_url)
    if err and err != "404":
        flash = urllib.parse.quote(f"could not read .pr-review.json: {err}")
        return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b""
    if cfg is None:
        cfg = {}
    if static_message:
        cfg["static_message"] = static_message
    elif "static_message" in cfg and not static_message:
        # Empty submission clears the banner.
        del cfg["static_message"]
    if model:
        cfg["model"] = model
    elif "model" in cfg and not model:
        del cfg["model"]
    payload = json.dumps(cfg, indent=2, sort_keys=True).encode()
    b64 = base64.b64encode(payload).decode()
    body = {"content": b64, "message": "pragent dashboard: update .pr-review.json"}
    if err == "404":
        # File didn't exist — Gitea contents PUT still creates the file when
        # `sha` is omitted, but only on certain versions; passing sha=None is
        # safer.
        pass
    else:
        # GET returned a sha — include it so Gitea enforces optimistic lock.
        # The sha lives in cfg's wrapper: re-fetch once to capture it.
        _, raw = _http(
            "GET",
            f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json",
            token=_bot_token(),
        )
        try:
            existing = json.loads(raw)
            sha = existing.get("sha")
            if sha:
                body["sha"] = sha
        except (json.JSONDecodeError, ValueError):
            pass
    status, _ = _http(
        "PUT",
        f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json",
        token=_bot_token(),
        body=body,
    )
    if status not in (200, 201):
        flash = urllib.parse.quote(f"Gitea PUT failed: status {status}")
        return 302, {"Location": f"/r/{repo_url}?flash={flash}"}, b""
    return 302, {"Location": f"/r/{repo_url}"}, b""
 def _fetch_pr_review_json(repo_url: str) -> tuple[dict | None, str | None]:
    """Return (cfg, None) on success, (None, None) when the file doesn't exist,
    (None, 'reason') on error."""
    if not _bot_token():
        return None, "PRAGENT_BOT_TOKEN not set"
    status, raw = _http(
        "GET",
        f"{_gitea_api()}/api/v1/repos/{repo_url}/contents/.pr-review.json",
        token=_bot_token(),
    )
    if status == 404:
        return None, "404"
    if status != 200:
        return None, f"status {status}"
    try:
        wrapper = json.loads(raw)
        content_b64 = wrapper.get("content", "").replace("\n", "")
        decoded = base64.b64decode(content_b64).decode("utf-8", errors="replace")
        cfg = json.loads(decoded)
    except (json.JSONDecodeError, ValueError) as e:
        return None, f"parse error: {e}"
    if not isinstance(cfg, dict):
        return None, "not a JSON object"
    return cfg, None
 # ---------------------------------------------------------------------------
 # Handler
 # ---------------------------------------------------------------------------
 class Handler(BaseHTTPRequestHandler):
    def _send(self, status: int, body: bytes, *, content_type: str = "text/html; charset=utf-8",
              extra_headers: dict | None = None) -> None:
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        if extra_headers:
            for k, v in extra_headers.items():
                self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)
    def _redirect(self, location: str) -> None:
        body = b""
        self.send_response(302)
        self.send_header("Location", location)
        self.send_header("Content-Length", "0")
        self.end_headers()
        self.wfile.write(body)
    def _unauthorized(self) -> None:
        """401 + Basic challenge so oauth2-proxy intercepts and redirects to Logto."""
        body = b"unauthorized\n"
        self.send_response(401)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("WWW-Authenticate", 'Basic realm="pragent-dashboard"')
        self.end_headers()
        self.wfile.write(body)
    # --- GET -----------------------------------------------------------------
    def do_GET(self):
        path = self.path
        # Static is exempt from auth (also unauthenticated browser fingerprinting
        # noise, but it's the same CSS regardless of viewer).
        if path == "/static/style.css":
            self._send(200, _route_static_css(), content_type="text/css; charset=utf-8")
            return
        if not _is_authed(self.headers):
            self._unauthorized()
            return
        if path == "/" or path == "":
            self._send(200, _route_overview())
            return
        # /r/<owner>/<name>            → repo
        # /r/<owner>/<name>/<index>    → PR
        # /r/<owner>/<name>/<index>/raw → raw Markdown
        m = _REPO_PR_RAW_RE.match(path)
        if m:
            owner, name, idx, raw = m.group(1), m.group(2), m.group(3), m.group(4)
            if raw:
                status, body = _route_pr_raw(owner, name, int(idx))
                self._send(status, body,
                           content_type="text/plain; charset=utf-8" if status == 200 else "text/plain")
                return
            if idx:
                self._send(200, _route_pr(owner, name, int(idx)))
                return
            self._send(200, _route_repo(owner, name))
            return
        self._send(404, b"not found", content_type="text/plain")
    # --- POST ----------------------------------------------------------------
    def do_POST(self):
        path = self.path
        if not _is_authed(self.headers):
            self._unauthorized()
            return
        # /r/<owner>/<name>/edit
        m = _EDIT_RE.match(path)
        if m:
            owner, name = m.group(1), m.group(2)
            length = int(self.headers.get("Content-Length", "0") or "0")
            raw = self.rfile.read(length) if length else b""
            form = urllib.parse.parse_qs(raw.decode("utf-8", errors="replace"))
            # Collapse lists to single values.
            form_single = {k: v[0] for k, v in form.items()}
            status, extra, body = _route_edit(owner, name, form_single)
            self._send(status, body, content_type="text/plain", extra_headers=extra)
            return
        self._send(404, b"not found", content_type="text/plain")
    def log_message(self, fmt, *args):
        print(f"pragent-dashboard: {self.address_string()} {fmt % args}", flush=True)
 # ---------------------------------------------------------------------------
 # Routing regexes (compiled at import time)
 # ---------------------------------------------------------------------------
 import re  # noqa: E402
 _REPO_PR_RAW_RE = re.compile(
    r"^/r/([^/]+)/([^/]+)(?:/(\d+)(?:/(raw))?)?/?$"
 )
 _EDIT_RE = re.compile(r"^/r/([^/]+)/([^/]+)/edit/?$")
 # ---------------------------------------------------------------------------
 # Main
 # ---------------------------------------------------------------------------
 def main() -> int:
    if not _feedback_db():
        print("pragent-dashboard: WARNING: PRAGENT_FEEDBACK_DB not set; dashboard will be empty",
              flush=True)
    print("pragent-dashboard: auth via oauth2-proxy (X-Forwarded-User required)", flush=True)
    server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
    print(f"pragent-dashboard: listening on :{PORT}", flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    return 0
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -1,302 +0,0 @@
 """pragent pilot — dashboard read-only query layer.
 Three functions: overview / repo_summary / pr_summary. Each opens the SQLite
 feedback DB via `feedback.init`, runs the queries it needs, and returns plain
 dicts/lists. NEVER writes — that's the dashboard_server's job (via the Gitea
 contents API). This module is what the dashboard_server's templates render.
 All three functions are tolerant of a missing or empty DB: they return the
 shaped dict with zeros/empty lists rather than crashing. The dashboard is a
 read-only view; the pilot can boot with no feedback DB and the dashboard
 should still load.
 Cost note: `total_cost_usd` is hardcoded to 0.0. Per-review `usage:cost` is
 not in the feedback SQLite — only the raw `review` / `inline_finding` rows
 are stored there. The equivalent-cost calc lives in `ai_review._render_collapsible_usage`
 and only knows about the latest review's tokens. Surfacing a rolled-up dollar
 figure without per-row telemetry would be guessing, so we don't.
 """
 from __future__ import annotations
 import datetime
 import os
 import sqlite3
 from pilot import feedback
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
 def _empty_overview() -> dict:
    return {
        "total_reviews": 0,
        "total_findings": 0,
        "total_repos": 0,
        "last_30d_reviews": 0,
        "daily": [{"date": _iso_date(i), "count": 0} for i in range(7)],
        "top_repos": [],
        "total_cost_usd": 0.0,
    }
 def _empty_repo_summary(repo: str) -> dict:
    return {
        "repo": repo,
        "total_runs": 0,
        "last_run_ts": 0,
        "runs_by_day": [],
        "findings_by_severity": {"critical": 0, "high": 0, "medium": 0, "low": 0},
        "top_findings": [],
        # NOTE: review rows don't carry a `model` column in the schema today,
        # so we have nothing to aggregate. When that lands, replace this
        # empty list with a `SELECT model, COUNT(*) …` over `review`.
        "models_used": [],
    }
 def _empty_pr_summary(repo: str, pr: int) -> dict:
    return {
        "repo": repo,
        "pr": pr,
        "head_sha": "",
        "posted_at": 0,
        "review_id_gitea": None,
        "body_comment_id": None,
        "findings": [],
        # usage isn't on the review row today; ai_review.py renders it
        # in-memory at review time. Leave empty.
        "usage": {},
    }
 def _iso_date(days_ago: int) -> str:
    """Return YYYY-MM-DD for `days_ago` days before today (UTC)."""
    d = datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days=days_ago)
    return d.isoformat()
 def _open_or_none(db_path: str) -> sqlite3.Connection | None:
    """Open the DB if it exists and looks like a feedback DB. Else None.
    Tolerates missing files (fresh container) and a schema-less file (the
    operator dropped a stray DB at the path). Returns a connection with
    Row factory set so callers can use `row["col"]`.
    """
    if not db_path or not os.path.exists(db_path):
        return None
    try:
        conn = feedback.init(db_path)
    except sqlite3.DatabaseError:
        return None
    return conn
 # ---------------------------------------------------------------------------
 # Public API
 # ---------------------------------------------------------------------------
 def overview(db_path: str) -> dict:
    """Top-of-page summary: totals + 7-bucket daily sparkline + top 5 repos."""
    conn = _open_or_none(db_path)
    if conn is None:
        return _empty_overview()
    try:
        cur = conn.execute("SELECT COUNT(*) FROM review")
        total_reviews = cur.fetchone()[0]
        cur = conn.execute("SELECT COUNT(*) FROM inline_finding")
        total_findings = cur.fetchone()[0]
        cur = conn.execute("SELECT COUNT(DISTINCT repo) FROM review")
        total_repos = cur.fetchone()[0]
        # Last 30d window — reviews AND findings posted within the window.
        ts_30d_ago = int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 30 * 86400
        cur = conn.execute("SELECT COUNT(*) FROM review WHERE posted_at >= ?", (ts_30d_ago,))
        last_30d_reviews = cur.fetchone()[0]
        # 7-bucket daily sparkline, oldest first. Bucket key is UTC date.
        cur = conn.execute(
            "SELECT posted_at FROM review WHERE posted_at >= ?",
            (int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 7 * 86400,),
        )
        buckets: dict[str, int] = {_iso_date(i): 0 for i in range(7)}
        for (ts,) in cur.fetchall():
            d = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).date().isoformat()
            if d in buckets:
                buckets[d] += 1
        daily = [{"date": _iso_date(i), "count": buckets[_iso_date(i)]} for i in range(7)]
        # Top 5 repos by run count, descending. last_seen is the most recent
        # review timestamp on that repo.
        cur = conn.execute(
            "SELECT repo, COUNT(*) AS runs, MAX(posted_at) AS last_seen "
            "FROM review GROUP BY repo ORDER BY runs DESC, last_seen DESC LIMIT 5"
        )
        top_repos = [
            {"repo": row[0], "run_count": row[1], "last_seen": int(row[2])}
            for row in cur.fetchall()
        ]
        return {
            "total_reviews": total_reviews,
            "total_findings": total_findings,
            "total_repos": total_repos,
            "last_30d_reviews": last_30d_reviews,
            "daily": daily,
            "top_repos": top_repos,
            "total_cost_usd": 0.0,
        }
    finally:
        conn.close()
 def repo_summary(db_path: str, repo: str) -> dict:
    """Per-repo drill-down: runs by day, severity histogram, top findings."""
    conn = _open_or_none(db_path)
    if conn is None:
        return _empty_repo_summary(repo)
    try:
        cur = conn.execute(
            "SELECT COUNT(*), MAX(posted_at) FROM review WHERE repo = ?", (repo,)
        )
        row = cur.fetchone()
        total_runs = row[0] or 0
        last_run_ts = int(row[1]) if row[1] else 0
        # runs_by_day for the last 30 days, oldest first; zero-buckets included.
        cur = conn.execute(
            "SELECT posted_at FROM review WHERE repo = ? AND posted_at >= ?",
            (repo, int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 30 * 86400),
        )
        buckets: dict[str, int] = {}
        for d in range(30):
            buckets[_iso_date(d)] = 0  # newest-day mapped to 0; we'll iterate
        # Re-key: build oldest-first, days_ago goes 29..0
        oldest_first = {}
        for d in range(30):
            oldest_first[_iso_date(29 - d)] = 0
        for (ts,) in cur.fetchall():
            d = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).date().isoformat()
            if d in oldest_first:
                oldest_first[d] += 1
        runs_by_day = [{"date": k, "count": v} for k, v in oldest_first.items()]
        # findings_by_severity — case-insensitive match; bucket unknown as 'low'.
        cur = conn.execute(
            "SELECT severity, COUNT(*) FROM inline_finding WHERE repo = ? GROUP BY severity",
            (repo,),
        )
        fbs = {"critical": 0, "high": 0, "medium": 0, "low": 0}
        for sev, n in cur.fetchall():
            k = (sev or "").strip().lower()
            if k not in fbs:
                k = "low"
            fbs[k] += n
        # top_findings — top 5 posthashes by occurrence count, joined with
        # vote rollups via feedback.findings_with_votes.
        cur = conn.execute(
            "SELECT f.path, f.line, MAX(f.severity) AS severity, MAX(f.problem) AS problem, "
            "COUNT(*) AS occurrences, "
            "COALESCE(SUM(CASE WHEN rct.content = '+1' THEN 1 ELSE 0 END), 0) AS upvotes, "
            "COALESCE(SUM(CASE WHEN rct.content = '-1' THEN 1 ELSE 0 END), 0) AS downvotes, "
            "MAX(ts.resolved) AS resolved, "
            "COALESCE((SELECT COUNT(*) FROM reply WHERE finding_id IN "
            "  (SELECT id FROM inline_finding WHERE posthash = f.posthash AND repo = f.repo AND path = f.path AND line = f.line)), 0) AS reply_count "
            "FROM inline_finding f "
            "LEFT JOIN reaction rct ON rct.comment_id = f.comment_id "
            "LEFT JOIN thread_state ts ON ts.finding_id = f.id "
            "WHERE f.repo = ? "
            "GROUP BY f.posthash, f.repo, f.path, f.line "
            "ORDER BY occurrences DESC, upvotes DESC LIMIT 5",
            (repo,),
        )
        top_findings = [
            {
                "path": r[0],
                "line": r[1],
                "severity": r[2],
                "problem": r[3],
                "occurrences": r[4],
                "upvotes": int(r[5] or 0),
                "downvotes": int(r[6] or 0),
                "resolved": int(r[7] or 0),
                "reply_count": int(r[8] or 0),
            }
            for r in cur.fetchall()
        ]
        return {
            "repo": repo,
            "total_runs": total_runs,
            "last_run_ts": last_run_ts,
            "runs_by_day": runs_by_day,
            "findings_by_severity": fbs,
            "top_findings": top_findings,
            "models_used": [],  # see _empty_repo_summary NOTE
        }
    finally:
        conn.close()
 def pr_summary(db_path: str, repo: str, pr: int) -> dict:
    """Per-PR view: meta + every finding the bot ever posted on that PR."""
    conn = _open_or_none(db_path)
    if conn is None:
        return _empty_pr_summary(repo, pr)
    try:
        cur = conn.execute(
            "SELECT head_sha, posted_at, review_id_gitea, body_comment_id "
            "FROM review WHERE repo = ? AND pr = ? ORDER BY posted_at DESC LIMIT 1",
            (repo, pr),
        )
        row = cur.fetchone()
        if row is None:
            return _empty_pr_summary(repo, pr)
        head_sha, posted_at, review_id_gitea, body_comment_id = row
        cur = conn.execute(
            "SELECT f.path, f.line, f.severity, f.problem, f.fix, f.suggestion, "
            "COALESCE(SUM(CASE WHEN rct.content = '+1' THEN 1 ELSE 0 END), 0) AS upvotes, "
            "COALESCE(SUM(CASE WHEN rct.content = '-1' THEN 1 ELSE 0 END), 0) AS downvotes, "
            "MAX(ts.resolved) AS resolved, "
            "COALESCE((SELECT COUNT(*) FROM reply WHERE finding_id = f.id), 0) AS reply_count "
            "FROM inline_finding f "
            "LEFT JOIN reaction rct ON rct.comment_id = f.comment_id "
            "LEFT JOIN thread_state ts ON ts.finding_id = f.id "
            "WHERE f.repo = ? AND f.pr = ? "
            "GROUP BY f.id "
            "ORDER BY f.path, f.line",
            (repo, pr),
        )
        findings = [
            {
                "path": r[0],
                "line": r[1],
                "severity": r[2],
                "problem": r[3],
                "fix": r[4],
                "suggestion": r[5],
                "upvotes": int(r[6] or 0),
                "downvotes": int(r[7] or 0),
                "resolved": int(r[8] or 0),
                "reply_count": int(r[9] or 0),
            }
            for r in cur.fetchall()
        ]
        return {
            "repo": repo,
            "pr": pr,
            "head_sha": head_sha,
            "posted_at": int(posted_at),
            "review_id_gitea": review_id_gitea,
            "body_comment_id": body_comment_id,
            "findings": findings,
            "usage": {},
        }
    finally:
        conn.close()
@@ -44,7 +44,6 @@ import sqlite3
 import sys
 import urllib.error
 import urllib.request
 from datetime import datetime, timezone
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@@ -112,51 +111,6 @@ def ensure_score_configs() -> dict:
 # 2. Dataset from recorded reviews
 # ---------------------------------------------------------------------------
 def item_id(repo: str, pr) -> str:
    """A dataset-item id that survives being put in a URL path.
    The obvious `{repo}#{pr}` is unusable: the UI routes items as
    `/datasets/{id}/items/{item_id}`, so the `/` in `owner/repo` splits into
    extra path segments and everything after the `#` is a fragment the browser
    never sends. The item is created fine and then 404s when opened.
    Session ids elsewhere keep the `{repo}#{pr}` form — those are never path
    segments, and `feedback_scores` depends on that shape.
    """
    return f"{repo.replace('/', '__')}__pr{pr}"
 def _item_metadata(*, repo, pr, head_sha, reviews_run, last_seen, findings) -> dict:
    """Filterable facets for one dataset item.
    Kept flat and primitive: the filter bar matches a metadata key against a
    literal, so a nested object or a list is not reachable from the UI.
    """
    owner, _, repo_name = str(repo).partition("/")
    sevs = [str(f["severity"] or "").lower() for f in findings]
    ranked = [s for s in sevs if s in eval_scores.SEVERITY_RANK]
    return {
        "repo": repo,
        "owner": owner or repo,
        "repo_name": repo_name or repo,
        "pr": int(pr),
        "head_sha": head_sha,
        "reviews_run": reviews_run,
        "last_reviewed_at": last_seen,
        "last_reviewed_iso": datetime.fromtimestamp(last_seen, timezone.utc).isoformat(),
        "finding_count": len(findings),
        "has_findings": bool(findings),
        # "none" rather than omitting the key: a filter for silent reviews needs
        # something to match, and an absent key matches nothing.
        "max_severity": (
            max(ranked, key=lambda s: eval_scores.SEVERITY_RANK[s]) if ranked else "none"
        ),
        # Flags that this row is the reviewer's own past output, not a human
        # judgement. Filter on it before anyone treats the dataset as truth.
        "labelled_by_human": False,
    }
 def read_review_items(db_path: str) -> list[dict]:
    """One dataset item per (repo, pr) the reviewer has run on.
@@ -186,7 +140,7 @@ def read_review_items(db_path: str) -> list[dict]:
            ).fetchall()
            items.append(
                {
-                    "id": item_id(row["repo"], row["pr"]),
+                    "id": f'{row["repo"]}#{row["pr"]}',
                    "input": {
                        "repo": row["repo"],
                        "pr": int(row["pr"]),
@@ -196,19 +150,14 @@ def read_review_items(db_path: str) -> list[dict]:
                        "findings": [dict(f) for f in findings],
                        "finding_count": len(findings),
                    },
-                    # The UI's filter bar reads metadata and nothing else, so
+                    "metadata": {
-                    # anything worth slicing on is a top-level key here even
+                        "reviews_run": int(row["reviews"]),
-                    # where it duplicates `input`. `owner` and `repo_name` are
+                        "last_reviewed_at": int(row["last_seen"]),
-                    # split out because a filter on the joined `repo` can only
+                        # Flags that this row is the reviewer's own past output,
-                    # match one repo at a time, never a whole org.
+                        # not a human judgement. Filter on it before anyone
-                    "metadata": _item_metadata(
+                        # treats the dataset as ground truth.
-                        repo=row["repo"],
+                        "labelled_by_human": False,
-                        pr=row["pr"],
+                    },
                        head_sha=row["head_sha"],
                        reviews_run=int(row["reviews"]),
                        last_seen=int(row["last_seen"]),
                        findings=findings,
                    ),
                }
            )
        return items
@@ -1,212 +0,0 @@
 #!/usr/bin/env python3
 """pragent pilot — populate the Experiments tab from reviews already traced.
 An "experiment" in Langfuse is a dataset run: a set of (dataset item, trace)
 links under one run name. The Experiments tab then shows one row per item with
 its scores, and lets two runs be diffed side by side.
 Nothing here re-runs the reviewer. Every PR in `pragent-reviews` has already
 been reviewed, and each of those reviews left a trace carrying its findings,
 cost and scores. This links what exists, which is what makes the tab useful on
 day one instead of after the next N pushes.
 Runs are grouped by **model** by default, because that is the comparison the
 pilot actually needs to make: the same PRs reviewed by MiniMax vs whatever
 replaces it, with `finding_rate` and `cost_per_finding` side by side. Group by
 `none` for a single "all traces" run.
 One trace per (run, item) — the most recent. A PR re-reviewed on every push has
 many traces, and a dataset run is defined as one output per input; feeding it
 the other five would make the per-run averages meaningless.
 Note on the endpoint: `POST /api/public/dataset-run-items` is deprecated in
 favour of the SDK experiment runner / OTel ingestion, and disappears in
 Langfuse v4. This instance is self-hosted v3, which the deprecation notice
 explicitly exempts from the cutoff date, and the pilot is stdlib-only by
 design. Revisit when this deployment moves to v4.
 Usage:
    LANGFUSE_HOST=... LANGFUSE_PUBLIC_KEY=... LANGFUSE_SECRET_KEY=... \\
        python3 eval_experiment.py --dry-run
 """
 from __future__ import annotations
 import argparse
 import json
 import os
 import sys
 import urllib.parse
 from collections import defaultdict
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 import eval_bootstrap as eb  # noqa: E402
 TRACE_NAME = "pr-review"
 # ---------------------------------------------------------------------------
 # Reading what already exists
 # ---------------------------------------------------------------------------
 def fetch_traces(name: str = TRACE_NAME, limit: int = 100, max_pages: int = 50) -> list[dict]:
    """Every review trace, newest first."""
    out: list[dict] = []
    for page in range(1, max_pages + 1):
        q = urllib.parse.urlencode({"name": name, "limit": limit, "page": page})
        st, body = eb._call("GET", f"/api/public/traces?{q}")
        if st != 200 or not isinstance(body, dict):
            raise SystemExit(f"listing traces failed: {st} {body}")
        data = body.get("data") or []
        out.extend(data)
        meta = body.get("meta") or {}
        if page * meta.get("limit", limit) >= meta.get("totalItems", 0):
            break
    return out
 def fetch_item_ids(dataset: str) -> set[str]:
    """Ids present in the dataset, so runs never reference a missing item."""
    ids: set[str] = set()
    for page in range(1, 51):
        q = urllib.parse.urlencode({"datasetName": dataset, "limit": 100, "page": page})
        st, body = eb._call("GET", f"/api/public/dataset-items?{q}")
        if st != 200 or not isinstance(body, dict):
            raise SystemExit(f"listing dataset items failed: {st} {body}")
        ids.update(i["id"] for i in body.get("data") or [])
        meta = body.get("meta") or {}
        if page * meta.get("limit", 100) >= meta.get("totalItems", 0):
            break
    return ids
 # ---------------------------------------------------------------------------
 # Grouping traces into runs
 # ---------------------------------------------------------------------------
 def trace_model(trace: dict) -> str:
    """The model that produced a review, from its `model:` tag."""
    for tag in trace.get("tags") or []:
        if tag.startswith("model:"):
            return tag[len("model:"):] or "unknown"
    return "unknown"
 def trace_item_id(trace: dict) -> str | None:
    """The dataset item a trace belongs to, or None if it is not a PR review."""
    md = trace.get("metadata") or {}
    repo, pr = md.get("repo"), md.get("pr")
    if not repo or pr in (None, ""):
        return None
    return eb.item_id(str(repo), pr)
 def _sort_key(trace: dict):
    return (trace.get("timestamp") or "", trace.get("id") or "")
 def plan_runs(traces: list[dict], known_items: set[str], group_by: str = "model") -> dict:
    """Map run name -> {item id: trace}, keeping only the newest trace per item.
    Traces whose PR is not in the dataset are dropped: `feedback.db` is the
    source for both, but a review can be traced without its row landing (the
    posting step can fail after the model ran), and a run item pointing at a
    non-existent dataset item is rejected.
    """
    runs: dict[str, dict[str, dict]] = defaultdict(dict)
    skipped_no_item, skipped_unknown = 0, 0
    for tr in traces:
        iid = trace_item_id(tr)
        if iid is None:
            skipped_unknown += 1
            continue
        if iid not in known_items:
            skipped_no_item += 1
            continue
        run = "all-traces" if group_by == "none" else trace_model(tr)
        prev = runs[run].get(iid)
        if prev is None or _sort_key(tr) > _sort_key(prev):
            runs[run][iid] = tr
    return {
        "runs": dict(runs),
        "skipped_not_in_dataset": skipped_no_item,
        "skipped_not_a_review": skipped_unknown,
    }
 def run_name(prefix: str, key: str) -> str:
    return f"{prefix}-{key}" if prefix else key
 # ---------------------------------------------------------------------------
 # Writing the runs
 # ---------------------------------------------------------------------------
 def create_run(name: str, items: dict[str, dict], description: str = "") -> dict:
    """Link each (item, trace) pair into the named run. Idempotent per pair."""
    created, failed = 0, []
    for iid, tr in sorted(items.items()):
        md = tr.get("metadata") or {}
        body = {
            "runName": name,
            "runDescription": description,
            "datasetItemId": iid,
            "traceId": tr["id"],
            "metadata": {
                "model": trace_model(tr),
                "engine": md.get("engine"),
                "findings": md.get("findings"),
                "duration_s": md.get("duration_s"),
                "cost_basis": md.get("cost_basis"),
                "linked_by": "eval_experiment.py",
            },
        }
        st, resp = eb._call("POST", "/api/public/dataset-run-items", body)
        if st in (200, 201):
            created += 1
        else:
            failed.append({"item": iid, "status": st, "error": resp})
    return {"run": name, "items_linked": created, "failed": failed}
 def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--dataset", default=eb.DATASET_NAME)
    ap.add_argument("--group-by", choices=("model", "none"), default="model")
    ap.add_argument("--prefix", default="baseline",
                    help="run name prefix; '' for the bare group key")
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args(argv)
    traces = fetch_traces()
    items = fetch_item_ids(args.dataset)
    plan = plan_runs(traces, items, group_by=args.group_by)
    report = {
        "traces_read": len(traces),
        "dataset_items": len(items),
        "skipped_not_in_dataset": plan["skipped_not_in_dataset"],
        "skipped_not_a_review": plan["skipped_not_a_review"],
        "runs": {},
    }
    for key, mapping in sorted(plan["runs"].items()):
        name = run_name(args.prefix, key)
        if args.dry_run:
            report["runs"][name] = {"items_would_link": len(mapping)}
            continue
        report["runs"][name] = create_run(
            name,
            mapping,
            description=(
                "Reviews already run by the pilot, linked after the fact. "
                "Scores come from the traces; expectedOutput is the reviewer's "
                "own prior output, not human-verified ground truth."
            ),
        )
    report["dry_run"] = args.dry_run
    print(json.dumps(report, indent=2))
    return 0
 if __name__ == "__main__":
    raise SystemExit(main())
@@ -208,18 +208,24 @@ def ensure_evaluators() -> dict:
 # ---------------------------------------------------------------------------
 def rule_body(name: str, judge_name: str, sampling: float) -> dict:
-    """POST /evaluation-rules shape for an LLM-as-judge observation rule.
+    """POST /evaluation-rules shape for an LLM-as-judge trace rule.
-    The judge is referenced by `name`+`scope`, not by id — ids name specific
+    Target is `trace` rather than `observation` on purpose: the standard
-    versions, names name the evaluator across versions. Mapping is required at
+    `/api/public/ingestion` path that ships review traces here feeds only
-    both the rule root (the server validates it there) and inside `evaluator`
+    the trace-upsert queue, and `evalService.createEvalJobs` only creates
-    (the API echoes it back). Filter is on `traceName` because that is the only
+    jobs for `targetObject ∈ {TRACE, DATASET}`. Observation rules are
-    stringOptions column the observation-rule schema exposes.
+    triggered exclusively from the OTel ingestion pipeline, which this
    pilot does not use. A trace rule reads the trace's own input/output —
    `langfuse_trace` already writes `_review_input`/`_review_output` onto
    the trace body for exactly this reason.
    Mapping is required at both the rule root (server validates it there)
    and inside `evaluator` (the API echoes it back).
    """
    return {
        "name": name,
        "enabled": True,
-        "target": "observation",
+        "target": "trace",
        "sampling": sampling,
        "filter": [
            {"column": "traceName", "operator": "any of",
@@ -163,7 +163,7 @@ def _md_escape(s: str) -> str:
 def analyze(db_path: str, *, since_ts: Optional[int] = None,
            as_json: bool = False) -> str:
    """Build the daily report. Returns a markdown string by default;
-    `as_json=True` returns a structured dict (for tests + dashboards)."""
+    `as_json=True` returns a structured dict (for tests + automation)."""
    conn = feedback.init(db_path)
    try:
        findings = list(feedback.findings_with_votes(conn, since_ts=since_ts))
@@ -0,0 +1,46 @@
 """Gitea transport adapter.
 This module owns HTTP mechanics only. Review policy, parsing, and publishing
 decisions stay in the review layer so they can be tested without a network.
 """
 from __future__ import annotations
 import json
 import urllib.error
 import urllib.request
 def request(
    method: str,
    url: str,
    token: str,
    body: dict | None = None,
    accept: str = "application/json",
 ) -> tuple[int, bytes]:
    headers = {"Authorization": f"token {token}", "Accept": accept}
    data = None
    if body is not None:
        data = json.dumps(body).encode()
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=180) as response:
            return response.status, response.read()
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read()
    except urllib.error.URLError as exc:
        raise RuntimeError(f"network error: {exc.reason}") from exc
 class GiteaClient:
    """Small adapter for repository-scoped Gitea calls."""
    def __init__(self, api: str, token: str):
        self.api = api.rstrip("/")
        self.token = token
    def get(self, path: str, accept: str = "application/json") -> tuple[int, bytes]:
        return request("GET", f"{self.api}/api/v1/repos/{path}", self.token, accept=accept)
    def post(self, path: str, body: dict) -> tuple[int, bytes]:
        return request("POST", f"{self.api}/api/v1/repos/{path}", self.token, body)
@@ -16,7 +16,7 @@ Provider split
 --------------
 `environment` on every trace is either `ollama` or `claude`, derived from the
 resolved display model (`resolve_environment`). That is what keeps the two
-spend stories separate in Langfuse: every dashboard, filter and cost breakdown
+spend stories separate in Langfuse: every view, filter and cost breakdown
 takes an environment selector, so "what did the local/self-hosted path cost"
 and "what did the Claude path cost" are two views of one project rather than
 two projects with two key pairs to rotate. Tags carry the finer split
@@ -280,8 +280,8 @@ def build_batch(
        "timestamp": ts,
        "environment": env,
        "sessionId": f"{repo}#{index}",
-        "input": _review_input(repo, index, sha, title),
+        "input": {"repo": repo, "pr": index, "sha": sha, "title": title},
-        "output": _review_output(summary, findings),
+        "output": {"summary": summary[:2000], "findings": len(findings or [])},
        "metadata": metadata,
        "tags": tags,
    }
@@ -310,11 +310,6 @@ def build_batch(
            "usageDetails": _usage_details(usage),
            "metadata": metadata,
            "level": "DEFAULT",
            # Repeated from the trace on purpose: an evaluator's variable
            # mapping reads the *observation's* input/output, so a generation
            # left blank cannot be judged at all.
            "input": _review_input(repo, index, sha, title),
            "output": _review_output(summary, findings),
        }
        if costs:
            gen_body["costDetails"] = costs
@@ -342,46 +337,6 @@ def build_batch(
    return events
 MAX_JUDGED_FINDINGS = 25
 _FIELD_CAP = 600
 def _review_input(repo: str, index, sha: str, title: str) -> dict:
    return {"repo": repo, "pr": index, "sha": sha, "title": title}
 def _review_output(summary: str, findings) -> dict:
    """What the reviewer actually said, in a shape an evaluator can read.
    The findings themselves are included, not just their count. A judge given
    only `{"summary": ..., "findings": 3}` can say nothing about whether those
    three findings are specific, actionable, or consistent with the summary —
    which is the whole question worth asking of a reviewer that has no ground
    truth to check against.
    Capped rather than complete: this rides in every ingestion batch, and a
    review with 80 findings would push the payload past what is reasonable to
    store per trace. `finding_count` stays exact so nothing reading the count
    is misled by the cap.
    """
    items = list(findings or [])
    return {
        "summary": summary[:2000],
        "finding_count": len(items),
        "findings_truncated": len(items) > MAX_JUDGED_FINDINGS,
        "findings": [
            {
                "path": f.get("path"),
                "line": f.get("line"),
                "severity": f.get("severity"),
                "problem": str(f.get("problem") or "")[:_FIELD_CAP],
                "fix": str(f.get("fix") or "")[:_FIELD_CAP],
            }
            for f in items[:MAX_JUDGED_FINDINGS]
        ],
    }
 def _score_events(*, cost_basis: str, **kwargs) -> list[dict]:
    """Deterministic scores for this review, or [] if the scorer is missing.
@@ -0,0 +1,36 @@
 """Model-provider adapter for the legacy Anthropic-compatible endpoint."""
 from __future__ import annotations
 import json
 try:  # Works both as `python pilot/ai_review.py` and `import pilot.model_client`.
    from .gitea_client import request
 except ImportError:  # pragma: no cover - script-style runtime
    from gitea_client import request
 def parse_text_blocks(content: object) -> str:
    """Return only text blocks from an Anthropic-style response."""
    if not isinstance(content, list):
        return ""
    return "\n".join(
        block["text"]
        for block in content
        if isinstance(block, dict)
        and block.get("type") == "text"
        and isinstance(block.get("text"), str)
    ).strip()
 def complete(base_url: str, model: str, system: str, user: str, max_tokens: int) -> str:
    payload = {
        "model": model,
        "max_tokens": max_tokens,
        "system": system,
        "messages": [{"role": "user", "content": user}],
    }
    status, raw = request("POST", f"{base_url.rstrip('/')}/v1/messages", "ollama", payload)
    if status != 200:
        detail = raw[:500].decode("utf-8", errors="replace")
        raise RuntimeError(f"model call failed: HTTP {status}: {detail}")
    return parse_text_blocks(json.loads(raw).get("content", []))
@@ -0,0 +1,32 @@
 """Trusted repository configuration and opt-in policy."""
 from __future__ import annotations
 import base64
 import json
 import urllib.parse
 from collections.abc import Callable
 def repo_enabled(
    get: Callable[..., tuple[int, bytes]],
    api: str,
    repo: str,
    ref: str,
    token: str,
 ) -> bool:
    """Read the opt-in flag from the trusted base branch.
    The transport is injected so the policy is testable without a live Gitea.
    Any missing, malformed, or non-boolean value disables review.
    """
    path = "contents/.pr-review.json?ref=" + urllib.parse.quote(ref, safe="")
    status, raw = get(api, repo, path, token)
    if status != 200:
        return False
    try:
        envelope = json.loads(raw)
        encoded = envelope.get("content", "").replace("\n", "")
        config = json.loads(base64.b64decode(encoded).decode("utf-8", errors="replace"))
    except (AttributeError, TypeError, ValueError, json.JSONDecodeError):
        return False
    return isinstance(config, dict) and config.get("enabled") is True
@@ -0,0 +1,17 @@
 """Stable interfaces shared by the review pipeline and its adapters."""
 from __future__ import annotations
 from typing import Protocol
 class Forge(Protocol):
    def get(self, path: str, accept: str = "application/json") -> tuple[int, bytes]: ...
    def post(self, path: str, body: dict) -> tuple[int, bytes]: ...
 class Reviewer(Protocol):
    def review(self, system: str, user: str, max_tokens: int) -> str: ...
 class Telemetry(Protocol):
    def emit(self, **event: object) -> None: ...
@@ -46,6 +46,7 @@ import urllib.parse
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from ai_review import gitea_get, review_pr
 from review_config import repo_enabled
 try:
    import feedback_harvest  # optional — absent in CI-step pod, present in
@@ -102,21 +103,7 @@ def is_repo_enabled(api: str, repo: str, ref: str, token: str) -> bool:
    The bool-coerce of `.get("enabled") is True` rejects the common
    gotchas (`null`, `1`, `"yes"`, missing field all yield False).
    """
-    code, raw = gitea_get(
+    return repo_enabled(gitea_get, api, repo, ref, token)
        api, repo,
        "contents/.pr-review.json?ref=" + urllib.parse.quote(ref, safe=""),
        token,
    )
    if code != 200:
        return False
    try:
        data = json.loads(raw)
        content_b64 = data.get("content", "").replace("\n", "")
        decoded = base64.b64decode(content_b64).decode("utf-8", errors="replace")
        cfg = json.loads(decoded)
    except (json.JSONDecodeError, ValueError):
        return False
    return isinstance(cfg, dict) and cfg.get("enabled") is True
 def _verify_signature(raw_body: bytes, headers) -> bool:
@@ -1,203 +0,0 @@
 """Tests for pilot/dashboard.py — stdlib HTTP server rendering dashboard HTML.
 We spin up the server on an ephemeral port in setUp, drive it with
 http.client, and tear it down in tearDown. Auth is now performed by
 oauth2-proxy: the dashboard trusts `X-Forwarded-User` set by the proxy
 and returns 401 (with a Basic challenge) when the header is missing.
 The dashboard reads `PRAGENT_FEEDBACK_DB` and renders views via
 `dashboard_data`. We seed an in-memory SQLite at `tmp_path` for each
 scenario that needs rows.
 """
 import http.client
 import os
 import socket
 import sys
 import tempfile
 import threading
 import time
 import unittest
 HERE = os.path.dirname(os.path.abspath(__file__))
 ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
 sys.path.insert(0, os.path.join(ROOT, "pilot"))
 import dashboard as dash  # noqa: E402
 from pilot import feedback  # noqa: E402
 def _free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port
 class _ServerThread:
    def __init__(self, port: int, handler):
        self.server = handler((host := "127.0.0.1", port), None)
        self.port = port
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
    def stop(self):
        self.server.shutdown()
        self.server.server_close()
        self.thread.join(timeout=2)
 def _get(port: int, path: str, headers: dict | None = None) -> tuple[int, dict, bytes]:
    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
    conn.request("GET", path, headers=headers or {})
    r = conn.getresponse()
    body = r.read()
    h = dict(r.getheaders())
    conn.close()
    return r.status, h, body
 def _post(port: int, path: str, body: bytes, headers: dict | None = None) -> tuple[int, dict, bytes]:
    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
    hdrs = {"Content-Type": "application/x-www-form-urlencoded"}
    if headers:
        hdrs.update(headers)
    conn.request("POST", path, body=body, headers=hdrs)
    r = conn.getresponse()
    body_b = r.read()
    h = dict(r.getheaders())
    conn.close()
    return r.status, h, body_b
 class TestDashboardAuth(unittest.TestCase):
    """Auth gate: require X-Forwarded-User (set by oauth2-proxy).
    When the header is missing every non-static route returns 401 with a
    Basic challenge, which lets oauth2-proxy redirect the browser to
    Logto. Static is exempt so the unauthenticated probe traffic doesn't
    loop the proxy through the auth flow.
    """
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = f"{self.tmp.name}/f.db"
        os.environ["PRAGENT_FEEDBACK_DB"] = self.db
        os.environ["DASHBOARD_PORT"] = str(0)  # we override below
        self.port = _free_port()
        from http.server import ThreadingHTTPServer
        self.srv = ThreadingHTTPServer(("127.0.0.1", self.port), dash.Handler)
        self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True)
        self.thread.start()
    def tearDown(self):
        self.srv.shutdown()
        self.srv.server_close()
        self.thread.join(timeout=2)
        for k in ("PRAGENT_FEEDBACK_DB", "DASHBOARD_PORT"):
            os.environ.pop(k, None)
        self.tmp.cleanup()
    def test_anonymous_overview_returns_401_with_basic_challenge(self):
        status, h, body = _get(self.port, "/")
        self.assertEqual(status, 401)
        self.assertEqual(h.get("WWW-Authenticate"), 'Basic realm="pragent-dashboard"')
        self.assertEqual(body, b"unauthorized\n")
    def test_anonymous_repo_returns_401(self):
        status, _h, _body = _get(self.port, "/r/alpha/one")
        self.assertEqual(status, 401)
    def test_anonymous_post_returns_401(self):
        status, _h, _body = _post(self.port, "/r/alpha/one/edit", b"x=1")
        self.assertEqual(status, 401)
    def test_authenticated_overview_succeeds(self):
        status, h, body = _get(self.port, "/", headers={"X-Forwarded-User": "marcos@example.com"})
        self.assertEqual(status, 200)
        self.assertIn(b"Overview", body)
    def test_static_does_not_require_auth(self):
        status, h, body = _get(self.port, "/static/style.css")
        self.assertEqual(status, 200)
        self.assertIn("text/css", h.get("Content-Type", ""))
        self.assertGreater(len(body), 50)
    def test_empty_x_forwarded_user_treated_as_anonymous(self):
        status, _h, _body = _get(self.port, "/", headers={"X-Forwarded-User": "   "})
        self.assertEqual(status, 401)
 class TestDashboardRender(unittest.TestCase):
    """Render-only tests — X-Forwarded-User set, real seeded data."""
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = f"{self.tmp.name}/f.db"
        os.environ["PRAGENT_FEEDBACK_DB"] = self.db
        # Seed: 2 repos, a couple of reviews + findings each.
        conn = feedback.init(self.db)
        for repo, n_prs in (("alpha/one", 2), ("beta/two", 1)):
            for n in range(n_prs):
                rid = feedback.record_review(
                    conn, repo=repo, pr=n + 1, head_sha=f"sha{repo}-{n}",
                    review_id_gitea=1000 + n, body_comment_id=2000 + n,
                    posted_at=int(time.time()) - n * 60,
                )
                for k in range(3):
                    feedback.record_inline_finding(
                        conn, review_id=rid, repo=repo, pr=n + 1,
                        path=f"src/file_{k}.py", line=k + 1,
                        severity=["critical", "high", "medium"][k],
                        problem=f"problem {k}",
                        fix=f"fix {k}", suggestion=f"suggestion {k}",
                        comment_id=3000 + n * 10 + k,
                    )
        conn.close()
        self.port = _free_port()
        from http.server import ThreadingHTTPServer
        self.srv = ThreadingHTTPServer(("127.0.0.1", self.port), dash.Handler)
        self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True)
        self.thread.start()
        self.auth_hdr = {"X-Forwarded-User": "marcos@example.com"}
    def tearDown(self):
        self.srv.shutdown()
        self.srv.server_close()
        self.thread.join(timeout=2)
        os.environ.pop("PRAGENT_FEEDBACK_DB", None)
        self.tmp.cleanup()
    def test_overview_200_contains_repo_names(self):
        status, _h, body = _get(self.port, "/", headers=self.auth_hdr)
        self.assertEqual(status, 200)
        text = body.decode()
        self.assertIn("Overview", text)
        self.assertIn("alpha/one", text)
        self.assertIn("beta/two", text)
    def test_repo_page_200(self):
        status, _h, body = _get(self.port, "/r/alpha/one", headers=self.auth_hdr)
        self.assertEqual(status, 200)
        text = body.decode()
        self.assertIn("alpha/one", text)
        # The findings table should appear.
        self.assertIn("src/file_0.py", text)
    def test_pr_page_200(self):
        status, _h, body = _get(self.port, "/r/alpha/one/1", headers=self.auth_hdr)
        self.assertEqual(status, 200)
        text = body.decode()
        self.assertIn("alpha/one", text)
        self.assertIn("#1", text)
        self.assertIn("src/file_0.py", text)
    def test_unknown_route_404(self):
        status, _h, _body = _get(self.port, "/no/such/route", headers=self.auth_hdr)
        self.assertEqual(status, 404)
 if __name__ == "__main__":
    unittest.main()
@@ -1,249 +0,0 @@
 """Tests for pilot/dashboard_data.py — read-only query layer over the feedback SQLite.
 Covers: empty-DB fallbacks (no crash on missing/empty DB), overview rollups,
 per-repo drill-down (findings by severity, top findings, runs by day), and
 the per-PR view. The dashboard never writes — only reads.
 """
 import os
 import sys
 import tempfile
 import time
 import unittest
 HERE = os.path.dirname(os.path.abspath(__file__))
 ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
 sys.path.insert(0, os.path.join(HERE, "..", ".."))  # so `from pilot import …` works
 from pilot import dashboard_data, feedback
 def _seed_repo(conn, *, repo: str, prs: int, findings_per_pr: int, day_offset: int = 0):
    """Seed one repo with `prs` PRs each with `findings_per_pr` findings.
    All timestamps cluster on (now - day_offset days). Returns list of review ids.
    """
    base = int(time.time()) - day_offset * 86400
    rids = []
    for n in range(prs):
        rid = feedback.record_review(
            conn, repo=repo, pr=n + 1, head_sha=f"sha{n}",
            review_id_gitea=1000 + n, body_comment_id=2000 + n,
            posted_at=base + n * 60,
        )
        rids.append(rid)
        for k in range(findings_per_pr):
            feedback.record_inline_finding(
                conn, review_id=rid, repo=repo, pr=n + 1,
                path=f"src/file_{k}.py", line=k + 1,
                severity=["critical", "high", "medium", "low"][k % 4],
                problem=f"problem {k}",
                fix=f"fix {k}", suggestion=f"suggestion {k}",
                comment_id=3000 + n * 10 + k,
                posted_at=base + n * 60,
            )
    return rids
 class TestEmptyDB(unittest.TestCase):
    def test_missing_file_returns_zero_dict(self):
        with tempfile.TemporaryDirectory() as d:
            missing = f"{d}/nope.db"
            data = dashboard_data.overview(missing)
            self.assertEqual(data["total_reviews"], 0)
            self.assertEqual(data["total_findings"], 0)
            self.assertEqual(data["total_repos"], 0)
            self.assertEqual(data["last_30d_reviews"], 0)
            self.assertEqual(len(data["daily"]), 7)
            self.assertEqual(data["top_repos"], [])
            self.assertEqual(data["total_cost_usd"], 0.0)
    def test_missing_file_repo_summary_safe(self):
        with tempfile.TemporaryDirectory() as d:
            data = dashboard_data.repo_summary(f"{d}/nope.db", "o/r")
            self.assertEqual(data["repo"], "o/r")
            self.assertEqual(data["total_runs"], 0)
            self.assertEqual(data["runs_by_day"], [])
            for sev in ("critical", "high", "medium", "low"):
                self.assertEqual(data["findings_by_severity"][sev], 0)
            self.assertEqual(data["top_findings"], [])
            self.assertEqual(data["models_used"], [])
    def test_missing_file_pr_summary_safe(self):
        with tempfile.TemporaryDirectory() as d:
            data = dashboard_data.pr_summary(f"{d}/nope.db", "o/r", 1)
            self.assertEqual(data["repo"], "o/r")
            self.assertEqual(data["pr"], 1)
            self.assertEqual(data["findings"], [])
            self.assertEqual(data["usage"], {})
 class TestEmptyButExistingDB(unittest.TestCase):
    """`init` creates the schema — DB exists but has no rows."""
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = f"{self.tmp.name}/f.db"
        feedback.init(self.db)
    def tearDown(self):
        self.tmp.cleanup()
    def test_overview_is_zero(self):
        data = dashboard_data.overview(self.db)
        self.assertEqual(data["total_reviews"], 0)
        self.assertEqual(data["total_findings"], 0)
        self.assertEqual(data["total_repos"], 0)
    def test_repo_summary_is_zero(self):
        data = dashboard_data.repo_summary(self.db, "o/r")
        self.assertEqual(data["total_runs"], 0)
        self.assertEqual(data["findings_by_severity"], {"critical": 0, "high": 0, "medium": 0, "low": 0})
    def test_pr_summary_is_zero(self):
        data = dashboard_data.pr_summary(self.db, "o/r", 1)
        self.assertEqual(data["findings"], [])
 class TestOverview(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = f"{self.tmp.name}/f.db"
        self.conn = feedback.init(self.db)
        _seed_repo(self.conn, repo="alpha/one", prs=3, findings_per_pr=2)
        _seed_repo(self.conn, repo="beta/two", prs=1, findings_per_pr=4)
        self.conn.close()
    def tearDown(self):
        self.tmp.cleanup()
    def test_totals(self):
        data = dashboard_data.overview(self.db)
        self.assertEqual(data["total_reviews"], 4)
        self.assertEqual(data["total_findings"], 6 + 4)  # 3*2 + 1*4 = 10
        self.assertEqual(data["total_repos"], 2)
        self.assertEqual(data["total_cost_usd"], 0.0)
    def test_top_repos_sorted_by_run_count(self):
        data = dashboard_data.overview(self.db)
        repos = [r["repo"] for r in data["top_repos"]]
        # alpha/one has 3 runs, beta/two has 1.
        self.assertEqual(repos[0], "alpha/one")
        self.assertEqual(data["top_repos"][0]["run_count"], 3)
        self.assertEqual(data["top_repos"][1]["run_count"], 1)
        # last_seen is a unix timestamp int.
        for r in data["top_repos"]:
            self.assertIsInstance(r["last_seen"], int)
    def test_daily_buckets_are_7(self):
        data = dashboard_data.overview(self.db)
        self.assertEqual(len(data["daily"]), 7)
        for b in data["daily"]:
            self.assertIn("date", b)
            self.assertIn("count", b)
    def test_last_30d_reviews(self):
        data = dashboard_data.overview(self.db)
        self.assertEqual(data["last_30d_reviews"], 4)
 class TestRepoSummary(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = f"{self.tmp.name}/f.db"
        self.conn = feedback.init(self.db)
        # 4 PRs with 2 findings each → 8 findings, severity cycle [c,h,m,l,c,h,m,l]
        _seed_repo(self.conn, repo="o/r", prs=4, findings_per_pr=2)
        # Add some reactions so top_findings has signal.
        rows = self.conn.execute(
            "SELECT id, comment_id FROM inline_finding WHERE repo=? ORDER BY id LIMIT 3",
            ("o/r",),
        ).fetchall()
        for r in rows:
            feedback.record_reaction(self.conn, comment_id=r["comment_id"], user="u", content="+1")
        self.conn.close()
    def tearDown(self):
        self.tmp.cleanup()
    def test_basic_shape(self):
        data = dashboard_data.repo_summary(self.db, "o/r")
        self.assertEqual(data["repo"], "o/r")
        self.assertEqual(data["total_runs"], 4)
        self.assertIsInstance(data["last_run_ts"], int)
    def test_findings_by_severity(self):
        data = dashboard_data.repo_summary(self.db, "o/r")
        fbs = data["findings_by_severity"]
        # 4 PRs × 2 findings; per-PR severities are [critical, high].
        # (k in range(2) → k=0 critical, k=1 high for every PR.)
        self.assertEqual(fbs["critical"], 4)
        self.assertEqual(fbs["high"], 4)
        self.assertEqual(fbs["medium"], 0)
        self.assertEqual(fbs["low"], 0)
    def test_runs_by_day_is_list(self):
        data = dashboard_data.repo_summary(self.db, "o/r")
        self.assertIsInstance(data["runs_by_day"], list)
        for r in data["runs_by_day"]:
            self.assertIn("date", r)
            self.assertIn("count", r)
    def test_top_findings_structure(self):
        data = dashboard_data.repo_summary(self.db, "o/r")
        self.assertGreater(len(data["top_findings"]), 0)
        first = data["top_findings"][0]
        for k in ("path", "line", "severity", "problem", "occurrences", "upvotes", "downvotes", "resolved", "reply_count"):
            self.assertIn(k, first)
    def test_models_used_is_empty_list_with_note(self):
        # The schema has no `model` column on review — the dashboard can't show
        # model usage from this DB today. We document that via an empty list.
        data = dashboard_data.repo_summary(self.db, "o/r")
        self.assertEqual(data["models_used"], [])
 class TestPRSummary(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = f"{self.tmp.name}/f.db"
        self.conn = feedback.init(self.db)
        rid = feedback.record_review(
            self.conn, repo="o/r", pr=42, head_sha="abc",
            review_id_gitea=9001, body_comment_id=8001,
            posted_at=1700000000,
        )
        for k in range(3):
            feedback.record_inline_finding(
                self.conn, review_id=rid, repo="o/r", pr=42,
                path=f"src/x_{k}.py", line=k + 10,
                severity=["critical", "high", "low"][k],
                problem=f"p{k}", fix=f"f{k}", suggestion=f"s{k}",
                comment_id=7000 + k,
            )
        self.conn.close()
    def tearDown(self):
        self.tmp.cleanup()
    def test_meta(self):
        data = dashboard_data.pr_summary(self.db, "o/r", 42)
        self.assertEqual(data["repo"], "o/r")
        self.assertEqual(data["pr"], 42)
        self.assertEqual(data["head_sha"], "abc")
        self.assertEqual(data["review_id_gitea"], 9001)
        self.assertEqual(data["body_comment_id"], 8001)
        self.assertEqual(data["posted_at"], 1700000000)
        # usage is empty because the schema has no usage column.
        self.assertEqual(data["usage"], {})
    def test_findings(self):
        data = dashboard_data.pr_summary(self.db, "o/r", 42)
        self.assertEqual(len(data["findings"]), 3)
        for f in data["findings"]:
            for k in ("path", "line", "severity", "problem", "fix", "suggestion",
                      "upvotes", "downvotes", "resolved", "reply_count"):
                self.assertIn(k, f)
 if __name__ == "__main__":
    unittest.main()
@@ -1,209 +0,0 @@
 """Tests for the edit endpoint — POST /r/<owner>/<name>/edit (Task C).
 We mock the Gitea HTTP layer (urllib.request.urlopen) so the test never
 touches the network. The dashboard handler is responsible for:
  * auth (X-Forwarded-User set by oauth2-proxy) + CSRF
  * read .pr-review.json via GET (404 → start from {})
  * validate model against cost_model.PRICES
  * PUT the updated file back, with sha + base64 content
  * redirect to /r/<owner>/<name> on success
 """
 import base64
 import http.client
 import json
 import os
 import socket
 import sys
 import tempfile
 import threading
 import unittest
 from unittest.mock import patch
 HERE = os.path.dirname(os.path.abspath(__file__))
 ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
 sys.path.insert(0, os.path.join(ROOT, "pilot"))
 import dashboard as dash  # noqa: E402
 from pilot import feedback  # noqa: E402
 def _free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port
 def _post(port: int, path: str, body: bytes, *, headers: dict | None = None) -> tuple[int, dict, bytes]:
    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
    hdrs = {"Content-Type": "application/x-www-form-urlencoded"}
    if headers:
        hdrs.update(headers)
    conn.request("POST", path, body=body, headers=hdrs)
    r = conn.getresponse()
    body_b = r.read()
    h = dict(r.getheaders())
    conn.close()
    return r.status, h, body_b
 class _FakeResp:
    def __init__(self, status: int, body: bytes):
        self.status = status
        self._body = body
    def read(self):
        return self._body
    def __enter__(self):
        return self
    def __exit__(self, *a):
        return False
 class TestDashboardEdit(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = f"{self.tmp.name}/f.db"
        os.environ["PRAGENT_FEEDBACK_DB"] = self.db
        os.environ["PRAGENT_BOT_TOKEN"] = "bot-token"
        # Seed a row so the repo page is meaningful.
        conn = feedback.init(self.db)
        feedback.record_review(
            conn, repo="o/r", pr=1, head_sha="x",
        )
        conn.close()
        self.port = _free_port()
        from http.server import ThreadingHTTPServer
        self.srv = ThreadingHTTPServer(("127.0.0.1", self.port), dash.Handler)
        self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True)
        self.thread.start()
        # Pull the per-process CSRF secret from the rendered repo page — the
        # edit form embeds the same token as a hidden input.
        self.csrf = dash._CSRF_SECRET
        self.auth_hdr = {"X-Forwarded-User": "marcos@example.com"}
        # Records of HTTP calls made by the handler.
        self.calls: list[tuple[str, str, dict | None, bytes | None]] = []
    def tearDown(self):
        self.srv.shutdown()
        self.srv.server_close()
        self.thread.join(timeout=2)
        for k in ("PRAGENT_FEEDBACK_DB", "PRAGENT_BOT_TOKEN"):
            os.environ.pop(k, None)
        self.tmp.cleanup()
    def _urlopen(self, req, timeout=30):
        """Replacement for urllib.request.urlopen that the handler uses."""
        url = req.full_url if hasattr(req, "full_url") else req
        method = getattr(req, "method", None) or "GET"
        body = getattr(req, "data", None)
        headers = dict(getattr(req, "headers", {}) or {})
        self.calls.append((method, url, headers, body))
        # Route based on URL: GET contents/.../raw vs PUT contents/.pr-review.json
        if method == "GET" and ".pr-review.json" in url:
            return _FakeResp(200, json.dumps({
                "content": base64.b64encode(b'{"focus":["x"],"model":"claude-haiku-4-5"}').decode(),
                "sha": "deadbeef",
            }).encode())
        if method == "PUT" and ".pr-review.json" in url:
            return _FakeResp(200, b'{}')
        return _FakeResp(404, b'{"message":"not found"}')
    def test_edit_updates_static_message_and_model(self):
        form = (
            f"_csrf={self.csrf}"
            f"&static_message=Hello%20world"
            f"&model=claude-sonnet-5"
        ).encode()
        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
            status, h, _b = _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
        self.assertEqual(status, 302)
        self.assertEqual(h.get("Location"), "/r/o/r")
        # Find the PUT call.
        put_calls = [c for c in self.calls if c[0] == "PUT"]
        self.assertEqual(len(put_calls), 1, self.calls)
        method, url, _hdrs, body = put_calls[0]
        self.assertIn(".pr-review.json", url)
        payload = json.loads(body)
        self.assertIn("content", payload)
        self.assertEqual(payload["sha"], "deadbeef")
        decoded = base64.b64decode(payload["content"]).decode()
        cfg = json.loads(decoded)
        self.assertEqual(cfg.get("static_message"), "Hello world")
        self.assertEqual(cfg.get("model"), "claude-sonnet-5")
    def test_edit_strips_static_message_to_400(self):
        long_msg = "x" * 600
        form = (
            f"_csrf={self.csrf}"
            f"&static_message={long_msg}"
            f"&model=claude-haiku-4-5"
        ).encode()
        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
            _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
        put = next(c for c in self.calls if c[0] == "PUT")
        cfg = json.loads(base64.b64decode(json.loads(put[3])["content"]))
        self.assertEqual(len(cfg["static_message"]), 400)
    def test_edit_rejects_unknown_model_with_flash(self):
        form = (
            f"_csrf={self.csrf}"
            f"&static_message=hi"
            f"&model=does-not-exist"
        ).encode()
        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
            status, h, _b = _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
        self.assertEqual(status, 302)
        self.assertIn("flash=", h.get("Location", ""))
        # No PUT should have been issued.
        put_calls = [c for c in self.calls if c[0] == "PUT"]
        self.assertEqual(put_calls, [])
    def test_edit_requires_auth(self):
        form = f"_csrf={self.csrf}&static_message=x&model=claude-haiku-4-5".encode()
        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
            status, h, _b = _post(self.port, "/r/o/r/edit", form)
        self.assertEqual(status, 401)
        self.assertEqual(h.get("WWW-Authenticate"), 'Basic realm="pragent-dashboard"')
        # No Gitea calls at all — auth gate fires first.
        self.assertEqual(self.calls, [])
    def test_edit_csrf_mismatch_redirects_without_save(self):
        form = f"_csrf=wrong&static_message=x&model=claude-haiku-4-5".encode()
        with patch.object(dash.urllib.request, "urlopen", side_effect=self._urlopen):
            status, h, _b = _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
        self.assertEqual(status, 302)
        self.assertEqual(h.get("Location"), "/r/o/r")
        put_calls = [c for c in self.calls if c[0] == "PUT"]
        self.assertEqual(put_calls, [])
    def test_edit_creates_file_when_missing(self):
        """When GET returns 404, the PUT must still happen (no sha)."""
        def _route(req, timeout=30):
            url = req.full_url
            method = getattr(req, "method", None) or "GET"
            body = getattr(req, "data", None)
            self.calls.append((method, url, {}, body))
            if method == "GET" and ".pr-review.json" in url:
                return _FakeResp(404, b'{"message":"not found"}')
            if method == "PUT" and ".pr-review.json" in url:
                return _FakeResp(201, b"{}")
            return _FakeResp(404, b"")
        form = f"_csrf={self.csrf}&static_message=hi&model=claude-haiku-4-5".encode()
        with patch.object(dash.urllib.request, "urlopen", side_effect=_route):
            status, h, _b = _post(self.port, "/r/o/r/edit", form, headers=self.auth_hdr)
        self.assertEqual(status, 302)
        put = next(c for c in self.calls if c[0] == "PUT")
        payload = json.loads(put[3])
        self.assertNotIn("sha", payload, "missing-file PUT should omit sha")
 if __name__ == "__main__":
    unittest.main()
@@ -1,74 +0,0 @@
 """Tests for the model <select> in the repo edit form (Task D)."""
 import http.client
 import os
 import socket
 import sys
 import tempfile
 import threading
 import unittest
 HERE = os.path.dirname(os.path.abspath(__file__))
 ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
 sys.path.insert(0, os.path.join(ROOT, "pilot"))
 import dashboard as dash  # noqa: E402
 from pilot import cost_model, feedback  # noqa: E402
 def _free_port() -> int:
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return port
 def _get(port: int, path: str, headers: dict | None = None) -> tuple[int, dict, bytes]:
    conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5)
    conn.request("GET", path, headers=headers or {})
    r = conn.getresponse()
    body = r.read()
    h = dict(r.getheaders())
    conn.close()
    return r.status, h, body
 class TestRepoEditSelect(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db = f"{self.tmp.name}/f.db"
        os.environ["PRAGENT_FEEDBACK_DB"] = self.db
        os.environ["PRAGENT_BOT_TOKEN"] = ""
        conn = feedback.init(self.db)
        feedback.record_review(conn, repo="o/r", pr=1, head_sha="x")
        conn.close()
        self.port = _free_port()
        from http.server import ThreadingHTTPServer
        self.srv = ThreadingHTTPServer(("127.0.0.1", self.port), dash.Handler)
        self.thread = threading.Thread(target=self.srv.serve_forever, daemon=True)
        self.thread.start()
        self.auth_hdr = {"X-Forwarded-User": "marcos@example.com"}
    def tearDown(self):
        self.srv.shutdown()
        self.srv.server_close()
        self.thread.join(timeout=2)
        for k in ("PRAGENT_FEEDBACK_DB", "PRAGENT_BOT_TOKEN"):
            os.environ.pop(k, None)
        self.tmp.cleanup()
    def test_repo_page_renders_select_with_one_option_per_price(self):
        status, _h, body = _get(self.port, "/r/o/r", headers=self.auth_hdr)
        self.assertEqual(status, 200)
        text = body.decode()
        self.assertIn('<select id="model" name="model">', text)
        # Every PRICES key should appear as an <option value="…">.
        for k in sorted(cost_model.PRICES):
            self.assertIn(f'<option value="{k}"', text, f"missing {k} in select")
        # Plus the "keep current" placeholder.
        self.assertIn("— keep current", text)
 if __name__ == "__main__":
    unittest.main()
@@ -1,158 +0,0 @@
 """Tests for the eval bootstrap's dataset-item construction."""
 import os
 import sqlite3
 import sys
 import urllib.parse
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot"))
 import eval_bootstrap as eb  # noqa: E402
 # --- item_id --------------------------------------------------------------
 def test_item_id_has_no_path_separator():
    """A `/` would split the UI's item route into extra path segments."""
    assert "/" not in eb.item_id("netcracker/interview", 29)
 def test_item_id_has_no_fragment_marker():
    """Everything after a `#` is a fragment the browser never sends."""
    assert "#" not in eb.item_id("netcracker/interview", 29)
 def test_item_id_survives_a_url_round_trip():
    """The id must appear verbatim in a path, needing no percent-encoding."""
    ident = eb.item_id("netcracker/interview", 29)
    assert urllib.parse.quote(ident, safe="") == ident
 def test_item_id_keeps_repo_and_pr_readable():
    assert eb.item_id("netcracker/interview", 29) == "netcracker__interview__pr29"
 def test_item_id_is_unique_per_pr():
    assert eb.item_id("o/r", 1) != eb.item_id("o/r", 2)
 def test_item_id_is_unique_per_repo():
    assert eb.item_id("o/one", 1) != eb.item_id("o/two", 1)
 def test_item_id_accepts_a_string_pr():
    assert eb.item_id("o/r", "29") == eb.item_id("o/r", 29)
 # --- read_review_items ----------------------------------------------------
 def _db(tmp_path, rows, findings=()):
    path = str(tmp_path / "feedback.db")
    conn = sqlite3.connect(path)
    conn.execute(
        "CREATE TABLE review (repo TEXT, pr INTEGER, posted_at INTEGER, head_sha TEXT)"
    )
    conn.execute(
        "CREATE TABLE inline_finding (repo TEXT, pr INTEGER, path TEXT, line INTEGER,"
        " severity TEXT, problem TEXT, fix TEXT)"
    )
    conn.executemany("INSERT INTO review VALUES (?,?,?,?)", rows)
    conn.executemany("INSERT INTO inline_finding VALUES (?,?,?,?,?,?,?)", findings)
    conn.commit()
    conn.close()
    return path
 def test_items_use_url_safe_ids(tmp_path):
    path = _db(tmp_path, [("netcracker/interview", 29, 100, "abc")])
    items = eb.read_review_items(path)
    assert [i["id"] for i in items] == ["netcracker__interview__pr29"]
 def test_item_input_keeps_the_real_repo_name(tmp_path):
    """The id is mangled for the URL; the payload must stay faithful."""
    path = _db(tmp_path, [("netcracker/interview", 29, 100, "abc")])
    item = eb.read_review_items(path)[0]
    assert item["input"]["repo"] == "netcracker/interview"
    assert item["input"]["pr"] == 29
 def test_one_item_per_pr_not_per_review(tmp_path):
    path = _db(
        tmp_path,
        [
            ("o/r", 1, 100, "a"),
            ("o/r", 1, 200, "b"),
            ("o/r", 2, 300, "c"),
        ],
    )
    items = eb.read_review_items(path)
    assert [i["id"] for i in items] == ["o__r__pr1", "o__r__pr2"]
    assert items[0]["metadata"]["reviews_run"] == 2
 def test_items_are_not_flagged_as_human_labelled(tmp_path):
    path = _db(tmp_path, [("o/r", 1, 100, "a")])
    assert eb.read_review_items(path)[0]["metadata"]["labelled_by_human"] is False
 # --- metadata facets ------------------------------------------------------
 def _md(findings=(), repo="netcracker/interview", pr=29):
    return eb._item_metadata(
        repo=repo, pr=pr, head_sha="abc", reviews_run=2, last_seen=1788189422,
        findings=[{"severity": s} for s in findings],
    )
 def test_metadata_carries_the_repo_for_filtering():
    assert _md()["repo"] == "netcracker/interview"
 def test_metadata_splits_owner_from_repo_name():
    """A filter on the joined repo can match one repo; owner matches an org."""
    md = _md()
    assert md["owner"] == "netcracker"
    assert md["repo_name"] == "interview"
 def test_owner_falls_back_when_the_repo_is_unqualified():
    md = _md(repo="standalone")
    assert md["owner"] == "standalone"
    assert md["repo_name"] == "standalone"
 def test_metadata_values_are_filterable_primitives():
    """Nested objects and lists are not reachable from the filter bar."""
    for key, value in _md(["high"]).items():
        assert isinstance(value, (str, int, float, bool)), key
 def test_max_severity_is_the_worst_finding():
    assert _md(["low", "critical", "medium"])["max_severity"] == "critical"
 def test_max_severity_is_none_not_absent_for_a_silent_review():
    md = _md([])
    assert md["max_severity"] == "none"
    assert md["has_findings"] is False
 def test_unknown_severity_does_not_win_the_max():
    assert _md(["banana", "low"])["max_severity"] == "low"
 def test_severity_comparison_ignores_case():
    assert _md(["HIGH"])["max_severity"] == "high"
 def test_finding_count_matches_the_findings():
    md = _md(["low", "low"])
    assert md["finding_count"] == 2
    assert md["has_findings"] is True
 def test_last_reviewed_is_exposed_both_ways():
    """The epoch sorts; the ISO string is what a human reads in a filter."""
    md = _md()
    assert md["last_reviewed_at"] == 1788189422
    assert md["last_reviewed_iso"].startswith("2026-08-31T")
@@ -1,159 +0,0 @@
 """Tests for linking existing review traces into dataset runs."""
 import os
 import sys
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "pilot"))
 import eval_experiment as ex  # noqa: E402
 def trace(tid, repo="o/r", pr=1, model="M2", ts="2026-08-01T00:00:00Z", **md):
    meta = {"repo": repo, "pr": pr}
    meta.update(md)
    return {
        "id": tid,
        "timestamp": ts,
        "tags": [f"model:{model}", "engine:opencode"],
        "metadata": meta,
    }
 # --- trace_model ----------------------------------------------------------
 def test_model_read_from_tag():
    assert ex.trace_model(trace("t1", model="MiniMax-M2.7")) == "MiniMax-M2.7"
 def test_model_falls_back_when_untagged():
    assert ex.trace_model({"tags": ["engine:opencode"]}) == "unknown"
 def test_model_falls_back_when_tags_absent():
    assert ex.trace_model({}) == "unknown"
 # --- trace_item_id --------------------------------------------------------
 def test_item_id_matches_the_bootstrap_scheme():
    assert ex.trace_item_id(trace("t1", repo="netcracker/interview", pr=29)) == \
        "netcracker__interview__pr29"
 def test_trace_without_repo_is_not_an_item():
    assert ex.trace_item_id({"metadata": {"pr": 1}}) is None
 def test_trace_without_pr_is_not_an_item():
    assert ex.trace_item_id({"metadata": {"repo": "o/r"}}) is None
 def test_trace_without_metadata_is_not_an_item():
    assert ex.trace_item_id({}) is None
 # --- plan_runs ------------------------------------------------------------
 ITEMS = {"o__r__pr1", "o__r__pr2"}
 def test_traces_group_by_model():
    plan = ex.plan_runs(
        [trace("a", pr=1, model="x"), trace("b", pr=2, model="y")], ITEMS
    )
    assert set(plan["runs"]) == {"x", "y"}
 def test_group_by_none_collapses_to_one_run():
    plan = ex.plan_runs(
        [trace("a", pr=1, model="x"), trace("b", pr=2, model="y")],
        ITEMS,
        group_by="none",
    )
    assert list(plan["runs"]) == ["all-traces"]
 def test_only_the_newest_trace_per_item_is_kept():
    """A re-reviewed PR has many traces; a run takes one output per input."""
    plan = ex.plan_runs(
        [
            trace("old", pr=1, ts="2026-08-01T00:00:00Z"),
            trace("new", pr=1, ts="2026-08-09T00:00:00Z"),
        ],
        ITEMS,
    )
    assert plan["runs"]["M2"]["o__r__pr1"]["id"] == "new"
 def test_newest_wins_regardless_of_input_order():
    older = trace("old", pr=1, ts="2026-08-01T00:00:00Z")
    newer = trace("new", pr=1, ts="2026-08-09T00:00:00Z")
    for order in ([older, newer], [newer, older]):
        plan = ex.plan_runs(order, ITEMS)
        assert plan["runs"]["M2"]["o__r__pr1"]["id"] == "new"
 def test_trace_for_a_pr_outside_the_dataset_is_skipped():
    plan = ex.plan_runs([trace("a", pr=99)], ITEMS)
    assert plan["runs"] == {}
    assert plan["skipped_not_in_dataset"] == 1
 def test_non_review_trace_is_counted_separately():
    plan = ex.plan_runs([{"id": "x", "metadata": {}}], ITEMS)
    assert plan["skipped_not_a_review"] == 1
    assert plan["skipped_not_in_dataset"] == 0
 def test_same_pr_different_models_lands_in_both_runs():
    plan = ex.plan_runs([trace("a", pr=1, model="x"), trace("b", pr=1, model="y")], ITEMS)
    assert plan["runs"]["x"]["o__r__pr1"]["id"] == "a"
    assert plan["runs"]["y"]["o__r__pr1"]["id"] == "b"
 # --- run_name -------------------------------------------------------------
 def test_run_name_prefixed():
    assert ex.run_name("baseline", "MiniMax-M2.7") == "baseline-MiniMax-M2.7"
 def test_empty_prefix_leaves_the_key_bare():
    assert ex.run_name("", "MiniMax-M2.7") == "MiniMax-M2.7"
 # --- create_run -----------------------------------------------------------
 def test_create_run_posts_one_item_per_pair(monkeypatch):
    calls = []
    def fake_call(method, path, body=None, timeout=20.0):
        calls.append((method, path, body))
        return 201, {}
    monkeypatch.setattr(ex.eb, "_call", fake_call)
    res = ex.create_run("run-1", {"o__r__pr1": trace("t1"), "o__r__pr2": trace("t2", pr=2)})
    assert res["items_linked"] == 2
    assert res["failed"] == []
    assert {c[1] for c in calls} == {"/api/public/dataset-run-items"}
    assert {c[2]["runName"] for c in calls} == {"run-1"}
 def test_create_run_links_the_trace_to_the_item(monkeypatch):
    seen = {}
    def fake_call(method, path, body=None, timeout=20.0):
        seen.update(body)
        return 201, {}
    monkeypatch.setattr(ex.eb, "_call", fake_call)
    ex.create_run("run-1", {"o__r__pr1": trace("t1")})
    assert seen["datasetItemId"] == "o__r__pr1"
    assert seen["traceId"] == "t1"
    assert seen["metadata"]["model"] == "M2"
 def test_create_run_reports_rejected_items(monkeypatch):
    monkeypatch.setattr(ex.eb, "_call", lambda *a, **k: (400, "nope"))
    res = ex.create_run("run-1", {"o__r__pr1": trace("t1")})
    assert res["items_linked"] == 0
    assert res["failed"][0]["item"] == "o__r__pr1"
    assert res["failed"][0]["status"] == 400
@@ -11,10 +11,16 @@ import eval_judges as ej  # noqa: E402
 # --- rule_body ------------------------------------------------------------
-def test_rule_body_targets_observations():
+def test_rule_body_targets_traces():
-    """Trace-level rules wouldn't see observation input/output."""
+    """Trace target matches the path `/api/public/ingestion` triggers.
    Observation rules only fire from the OTel ingestion pipeline; this
    pilot uses standard ingestion, so its jobs only come from
    `evalService.createEvalJobs` and that dispatcher handles
    `targetObject ∈ {TRACE, DATASET}`.
    """
    body = ej.rule_body("rule-x", "finding_actionability", 1.0)
-    assert body["target"] == "observation"
+    assert body["target"] == "trace"
    assert body["enabled"] is True
Author	SHA1	Message	Date
Claude	b4041f6892	refactor: split pilot architecture Remove the obsolete dashboard now that Langfuse is the analytics surface.\nIntroduce focused transport, model, and configuration modules while preserving the ai_review facade, and document the current runtime architecture.	2026-09-01 00:17:16 +00:00
Claude	5b8be61e2f	test: trigger review after DB rule insert	2026-08-31 19:19:59 +00:00
Claude	3109bd7c2d	pilot(eval): switch rule target from observation to trace The standard /api/public/ingestion path feeds only the trace-upsert queue; evalService.createEvalJobs only dispatches targetObject in {TRACE, DATASET}. Observation rules fire exclusively from the OTel pipeline, which this pilot does not use. The trace body already carries review input/output via langfuse_trace, so a trace rule sees the same material an observation rule would.	2026-08-31 19:18:59 +00:00
gitea_admin	d746b1fdc2	test: trigger review for eval judges	2026-08-31 18:37:41 +00:00