diff --git a/apps/api/app/auth_sessions.py b/apps/api/app/auth_sessions.py
index 277b589..3f79bbc 100644
--- a/apps/api/app/auth_sessions.py
+++ b/apps/api/app/auth_sessions.py
@@ -343,15 +343,46 @@ async def _runtime_tables_ready(conn) -> bool:
             WHERE stage_code IN ('라포','탐색','개입','정리')
           ) AS has_stage_defs,
           to_regclass('app.admin_health_event') IS NOT NULL AS has_admin_health_event,
+          to_regclass('app.admin_health_daily_rollup') IS NOT NULL AS has_admin_health_daily_rollup,
+          EXISTS (
+            SELECT 1 FROM information_schema.columns
+            WHERE table_schema = 'app'
+              AND table_name = 'admin_health_daily_rollup'
+              AND column_name = 'last_down_at'
+          ) AS has_admin_health_daily_rollup_columns,
           to_regclass('app.support_ticket') IS NOT NULL AS has_support_ticket,
+          EXISTS (
+            SELECT 1 FROM information_schema.columns
+            WHERE table_schema = 'app'
+              AND table_name = 'support_ticket'
+              AND column_name IN ('fingerprint', 'parent_ticket_id')
+            GROUP BY table_schema, table_name
+            HAVING count(*) = 2
+          ) AS has_support_ticket_duplicate_columns,
           EXISTS (
             SELECT 1 FROM pg_policies
             WHERE schemaname = 'app'
               AND tablename = 'admin_health_event'
-              AND policyname IN ('p_admin_health_event_select','p_admin_health_event_insert')
+              AND policyname IN (
+                'p_admin_health_event_select',
+                'p_admin_health_event_insert',
+                'p_admin_health_event_delete'
+              )
             GROUP BY schemaname, tablename
-            HAVING count(*) = 2
+            HAVING count(*) = 3
           ) AS has_admin_health_event_policies,
+          EXISTS (
+            SELECT 1 FROM pg_policies
+            WHERE schemaname = 'app'
+              AND tablename = 'admin_health_daily_rollup'
+              AND policyname IN (
+                'p_admin_health_daily_rollup_select',
+                'p_admin_health_daily_rollup_insert',
+                'p_admin_health_daily_rollup_update'
+              )
+            GROUP BY schemaname, tablename
+            HAVING count(*) = 3
+          ) AS has_admin_health_daily_rollup_policies,
           EXISTS (
             SELECT 1 FROM pg_policies
             WHERE schemaname = 'app'
@@ -365,6 +396,19 @@ async def _runtime_tables_ready(conn) -> bool:
             GROUP BY schemaname, tablename
             HAVING count(*) = 4
           ) AS has_support_ticket_policies,
+          to_regclass('app.learner_prepost_measure') IS NOT NULL AS has_learner_prepost_measure,
+          EXISTS (
+            SELECT 1 FROM pg_policies
+            WHERE schemaname = 'app'
+              AND tablename = 'learner_prepost_measure'
+              AND policyname IN (
+                'p_learner_prepost_measure_select',
+                'p_learner_prepost_measure_insert',
+                'p_learner_prepost_measure_update'
+              )
+            GROUP BY schemaname, tablename
+            HAVING count(*) = 3
+          ) AS has_learner_prepost_measure_policies,
           EXISTS (
             SELECT 1 FROM pg_policies
             WHERE schemaname = 'app'
@@ -407,9 +451,15 @@ async def _runtime_tables_ready(conn) -> bool:
         and row["has_state_columns"]
         and row["has_stage_defs"]
         and row["has_admin_health_event"]
+        and row["has_admin_health_daily_rollup"]
+        and row["has_admin_health_daily_rollup_columns"]
         and row["has_support_ticket"]
+        and row["has_support_ticket_duplicate_columns"]
         and row["has_admin_health_event_policies"]
+        and row["has_admin_health_daily_rollup_policies"]
         and row["has_support_ticket_policies"]
+        and row["has_learner_prepost_measure"]
+        and row["has_learner_prepost_measure_policies"]
         and row["has_session_write_policies"]
         and row["removed_old_session_policy"]
         and row["has_turn_write_policies"]
@@ -574,6 +624,34 @@ async def ensure_runtime_tables() -> None:
               ON app.admin_health_event(observed_at DESC, service_key)
             """
         )
+        await conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS app.admin_health_daily_rollup (
+              rollup_date DATE NOT NULL,
+              environment TEXT NOT NULL,
+              engine_mode TEXT NOT NULL,
+              service_key TEXT NOT NULL,
+              service_name TEXT NOT NULL,
+              sample_count INTEGER NOT NULL DEFAULT 0 CHECK (sample_count >= 0),
+              ok_samples INTEGER NOT NULL DEFAULT 0 CHECK (ok_samples >= 0),
+              degraded_samples INTEGER NOT NULL DEFAULT 0 CHECK (degraded_samples >= 0),
+              down_samples INTEGER NOT NULL DEFAULT 0 CHECK (down_samples >= 0),
+              first_observed_at TIMESTAMPTZ NOT NULL,
+              last_observed_at TIMESTAMPTZ NOT NULL,
+              latest_status TEXT NOT NULL CHECK (latest_status IN ('ok','degraded','down')),
+              last_down_at TIMESTAMPTZ,
+              max_load REAL NOT NULL DEFAULT 0.0 CHECK (max_load >= 0.0 AND max_load <= 1.0),
+              generated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+              PRIMARY KEY (rollup_date, environment, engine_mode, service_key)
+            )
+            """
+        )
+        await conn.execute(
+            """
+            CREATE INDEX IF NOT EXISTS idx_admin_health_daily_rollup_latest
+              ON app.admin_health_daily_rollup(last_observed_at DESC, service_key)
+            """
+        )
         await conn.execute(
             """
             CREATE TABLE IF NOT EXISTS app.support_ticket (
@@ -599,6 +677,8 @@ async def ensure_runtime_tables() -> None:
               subject TEXT NOT NULL,
               body TEXT NOT NULL,
               source_path TEXT NOT NULL DEFAULT '',
+              fingerprint TEXT NOT NULL DEFAULT '',
+              parent_ticket_id UUID REFERENCES app.support_ticket(id) ON DELETE SET NULL,
               assigned_group TEXT NOT NULL DEFAULT '',
               resolution_note TEXT NOT NULL DEFAULT '',
               created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
@@ -620,17 +700,95 @@ async def ensure_runtime_tables() -> None:
               ON app.support_ticket(reporter_id, created_at DESC)
             """
         )
+        await conn.execute(
+            """
+            ALTER TABLE app.support_ticket
+              ADD COLUMN IF NOT EXISTS fingerprint TEXT NOT NULL DEFAULT '',
+              ADD COLUMN IF NOT EXISTS parent_ticket_id UUID REFERENCES app.support_ticket(id) ON DELETE SET NULL
+            """
+        )
+        await conn.execute(
+            """
+            CREATE INDEX IF NOT EXISTS idx_support_ticket_fingerprint
+              ON app.support_ticket(fingerprint, created_at DESC)
+              WHERE fingerprint <> ''
+            """
+        )
+        await conn.execute(
+            """
+            CREATE INDEX IF NOT EXISTS idx_support_ticket_parent
+              ON app.support_ticket(parent_ticket_id)
+              WHERE parent_ticket_id IS NOT NULL
+            """
+        )
+        await conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS app.learner_prepost_measure (
+              id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+              learner_id UUID NOT NULL REFERENCES app.app_user(user_id) ON DELETE CASCADE,
+              pilot_id TEXT NOT NULL DEFAULT 'phase3-pilot-draft',
+              measure_name TEXT NOT NULL CHECK (
+                measure_name IN ('self_efficacy','skill_proficiency','training_satisfaction')
+              ),
+              timepoint TEXT NOT NULL CHECK (timepoint IN ('pre','post')),
+              raw_score NUMERIC(8,3) NOT NULL,
+              min_score NUMERIC(8,3) NOT NULL DEFAULT 1.0,
+              max_score NUMERIC(8,3) NOT NULL DEFAULT 5.0,
+              instrument_version TEXT NOT NULL DEFAULT 'pilot-prepost-scaffold-2026-06-28',
+              item_count INTEGER NOT NULL DEFAULT 1 CHECK (item_count >= 1 AND item_count <= 80),
+              collected_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+              updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+              CHECK (max_score > min_score),
+              CHECK (raw_score >= min_score AND raw_score <= max_score),
+              UNIQUE (learner_id, pilot_id, measure_name, timepoint, instrument_version)
+            )
+            """
+        )
+        await conn.execute(
+            """
+            CREATE INDEX IF NOT EXISTS idx_learner_prepost_measure_pilot
+              ON app.learner_prepost_measure(pilot_id, measure_name, timepoint)
+            """
+        )
+        await conn.execute(
+            """
+            CREATE INDEX IF NOT EXISTS idx_learner_prepost_measure_learner
+              ON app.learner_prepost_measure(learner_id, pilot_id)
+            """
+        )
         await conn.execute(
             """
             ALTER TABLE app.admin_health_event ENABLE ROW LEVEL SECURITY;
             DROP POLICY IF EXISTS p_admin_health_event_select ON app.admin_health_event;
             DROP POLICY IF EXISTS p_admin_health_event_insert ON app.admin_health_event;
+            DROP POLICY IF EXISTS p_admin_health_event_delete ON app.admin_health_event;
             CREATE POLICY p_admin_health_event_select
               ON app.admin_health_event FOR SELECT
               USING (app.current_role_name() = 'admin');
             CREATE POLICY p_admin_health_event_insert
               ON app.admin_health_event FOR INSERT
               WITH CHECK (app.current_role_name() = 'admin');
+            CREATE POLICY p_admin_health_event_delete
+              ON app.admin_health_event FOR DELETE
+              USING (app.current_role_name() = 'admin');
+
+            ALTER TABLE app.admin_health_daily_rollup ENABLE ROW LEVEL SECURITY;
+            DROP POLICY IF EXISTS p_admin_health_daily_rollup_select ON app.admin_health_daily_rollup;
+            DROP POLICY IF EXISTS p_admin_health_daily_rollup_insert ON app.admin_health_daily_rollup;
+            DROP POLICY IF EXISTS p_admin_health_daily_rollup_update ON app.admin_health_daily_rollup;
+            DROP POLICY IF EXISTS p_admin_health_daily_rollup_delete ON app.admin_health_daily_rollup;
+            ALTER TABLE app.admin_health_daily_rollup
+              ADD COLUMN IF NOT EXISTS last_down_at TIMESTAMPTZ;
+            CREATE POLICY p_admin_health_daily_rollup_select
+              ON app.admin_health_daily_rollup FOR SELECT
+              USING (app.current_role_name() = 'admin');
+            CREATE POLICY p_admin_health_daily_rollup_insert
+              ON app.admin_health_daily_rollup FOR INSERT
+              WITH CHECK (app.current_role_name() = 'admin');
+            CREATE POLICY p_admin_health_daily_rollup_update
+              ON app.admin_health_daily_rollup FOR UPDATE
+              USING (app.current_role_name() = 'admin')
+              WITH CHECK (app.current_role_name() = 'admin');
 
             ALTER TABLE app.support_ticket ENABLE ROW LEVEL SECURITY;
             DROP POLICY IF EXISTS p_support_ticket_select ON app.support_ticket;
@@ -656,6 +814,41 @@ async def ensure_runtime_tables() -> None:
             CREATE POLICY p_support_ticket_delete
               ON app.support_ticket FOR DELETE
               USING (app.current_role_name() = 'admin');
+
+            ALTER TABLE app.learner_prepost_measure ENABLE ROW LEVEL SECURITY;
+            DROP POLICY IF EXISTS p_learner_prepost_measure_select ON app.learner_prepost_measure;
+            DROP POLICY IF EXISTS p_learner_prepost_measure_insert ON app.learner_prepost_measure;
+            DROP POLICY IF EXISTS p_learner_prepost_measure_update ON app.learner_prepost_measure;
+            CREATE POLICY p_learner_prepost_measure_select
+              ON app.learner_prepost_measure FOR SELECT
+              USING (
+                app.current_role_name() = 'admin'
+                OR learner_id = app.current_uid()
+                OR (
+                  app.current_role_name() = 'instructor'
+                  AND EXISTS (
+                    SELECT 1 FROM app.app_user u
+                    WHERE u.user_id = app.learner_prepost_measure.learner_id
+                      AND u.cohort = current_setting('app.current_cohort', true)
+                  )
+                )
+              );
+            CREATE POLICY p_learner_prepost_measure_insert
+              ON app.learner_prepost_measure FOR INSERT
+              WITH CHECK (
+                app.current_role_name() = 'admin'
+                OR learner_id = app.current_uid()
+              );
+            CREATE POLICY p_learner_prepost_measure_update
+              ON app.learner_prepost_measure FOR UPDATE
+              USING (
+                app.current_role_name() = 'admin'
+                OR learner_id = app.current_uid()
+              )
+              WITH CHECK (
+                app.current_role_name() = 'admin'
+                OR learner_id = app.current_uid()
+              );
             """
         )
         await conn.execute(
@@ -674,6 +867,12 @@ async def ensure_runtime_tables() -> None:
               ADD COLUMN IF NOT EXISTS turns_in_stage INT NOT NULL DEFAULT 0
             """
         )
+        await conn.execute(
+            """
+            ALTER TABLE app.turns
+              ADD COLUMN IF NOT EXISTS provider_events JSONB NOT NULL DEFAULT '[]'::jsonb
+            """
+        )
         await conn.execute(
             """
             INSERT INTO app.stage_def (stage_code, display_name, seq, base_openness)
@@ -730,6 +929,37 @@ async def ensure_runtime_tables() -> None:
             )
             """
         )
+        await conn.execute(
+            """
+            ALTER TABLE app.pinned_fact ENABLE ROW LEVEL SECURITY;
+            DROP POLICY IF EXISTS p_pinned_insert ON app.pinned_fact;
+            DROP POLICY IF EXISTS p_pinned_update ON app.pinned_fact;
+
+            CREATE POLICY p_pinned_insert ON app.pinned_fact FOR INSERT WITH CHECK (
+              app.current_role_name() IN ('admin','instructor')
+              OR EXISTS (
+                SELECT 1 FROM app.case_profile cp
+                WHERE cp.case_id = app.pinned_fact.case_id
+                  AND cp.learner_id = app.current_uid()
+              )
+            );
+            CREATE POLICY p_pinned_update ON app.pinned_fact FOR UPDATE USING (
+              app.current_role_name() IN ('admin','instructor')
+              OR EXISTS (
+                SELECT 1 FROM app.case_profile cp
+                WHERE cp.case_id = app.pinned_fact.case_id
+                  AND cp.learner_id = app.current_uid()
+              )
+            ) WITH CHECK (
+              app.current_role_name() IN ('admin','instructor')
+              OR EXISTS (
+                SELECT 1 FROM app.case_profile cp
+                WHERE cp.case_id = app.pinned_fact.case_id
+                  AND cp.learner_id = app.current_uid()
+              )
+            )
+            """
+        )
         await conn.execute(
             """
             DROP POLICY IF EXISTS p_turns_modify ON app.turns;
diff --git a/apps/api/app/db.py b/apps/api/app/db.py
index b69dc1c..32e2860 100644
--- a/apps/api/app/db.py
+++ b/apps/api/app/db.py
@@ -150,6 +150,7 @@ async def healthcheck() -> bool:
                   to_regclass('app.user_preferences') IS NOT NULL AS has_preferences,
                   to_regclass('app.admin_engine_config') IS NOT NULL AS has_engine_config,
                   to_regclass('app.admin_health_event') IS NOT NULL AS has_admin_health_event,
+                  to_regclass('app.admin_health_daily_rollup') IS NOT NULL AS has_admin_health_daily_rollup,
                   to_regclass('app.support_ticket') IS NOT NULL AS has_support_ticket
                 """
             )
@@ -160,6 +161,7 @@ async def healthcheck() -> bool:
                 and row["has_preferences"]
                 and row["has_engine_config"]
                 and row["has_admin_health_event"]
+                and row["has_admin_health_daily_rollup"]
                 and row["has_support_ticket"]
             )
     except Exception:
diff --git a/apps/api/app/routes/admin.py b/apps/api/app/routes/admin.py
index 918be6b..8b456f2 100644
--- a/apps/api/app/routes/admin.py
+++ b/apps/api/app/routes/admin.py
@@ -6,6 +6,7 @@ import time
 from datetime import datetime, timezone
 from decimal import Decimal
 from typing import Annotated, Literal
+from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from pydantic import BaseModel, Field
@@ -28,7 +29,7 @@ from ..deps import Principal, require_admin_access
 from ..engine_client import engine_client
 from ..runtime_policy import require_runtime_fallback_allowed
 from ..services.voice import voice_service
-from ..services import rag
+from ..services import evaluator, rag
 from ..store import store
 
 router = APIRouter(prefix="/admin", tags=["admin"])
@@ -73,6 +74,14 @@ class AdminUsageBreakdown(BaseModel):
     cost_usd: float
 
 
+class AdminUsageDailyCost(BaseModel):
+    day: str
+    turns: int
+    tokens_in: int
+    tokens_out: int
+    cost_usd: float
+
+
 class AdminUsageBudget(BaseModel):
     limit_usd: float
     used_ratio: float
@@ -80,6 +89,17 @@ class AdminUsageBudget(BaseModel):
     status: UsageBudgetStatus
 
 
+class AdminUsageEvaluatorCache(BaseModel):
+    enabled: bool
+    entries: int
+    hits: int
+    misses: int
+    stores: int
+    evictions: int
+    requests: int
+    hit_rate: float
+
+
 class AdminUsageResponse(BaseModel):
     source: Literal["database", "server_session_registry"]
     durable: bool
@@ -91,7 +111,9 @@ class AdminUsageResponse(BaseModel):
     tokens_out: int
     cost_usd: float
     budget: AdminUsageBudget
+    evaluator_cache: AdminUsageEvaluatorCache
     by_provider: list[AdminUsageBreakdown]
+    daily_cost: list[AdminUsageDailyCost] = Field(default_factory=list)
 
 
 class AdminHealthEvent(BaseModel):
@@ -147,6 +169,11 @@ class AdminSupportTicketResponse(BaseModel):
     subject: str
     body: str
     source_path: str
+    fingerprint: str
+    parent_ticket_id: str | None = None
+    duplicate_count: int = 0
+    duplicate_parent_candidate_id: str | None = None
+    child_ticket_count: int = 0
     assigned_group: str
     resolution_note: str
     created_at: float
@@ -195,6 +222,7 @@ class AdminTicketPatch(BaseModel):
     priority: TicketPriority | None = None
     assigned_group: str | None = Field(default=None, max_length=120)
     resolution_note: str | None = Field(default=None, max_length=2000)
+    parent_ticket_id: str | None = Field(default=None, max_length=36)
 
 
 class AdminUserResponse(BaseModel):
@@ -298,6 +326,29 @@ def _usage_budget(cost_usd: float) -> AdminUsageBudget:
     )
 
 
+def _usage_evaluator_cache() -> AdminUsageEvaluatorCache:
+    stats = evaluator.evaluator_semantic_cache_stats()
+    hits = _safe_usage_int(stats.get("hits", 0))
+    misses = _safe_usage_int(stats.get("misses", 0))
+    requests = hits + misses
+    hit_rate = round(hits / requests, 4) if requests else 0.0
+    enabled = (
+        bool(settings.evaluator_semantic_cache_enabled)
+        and settings.evaluator_semantic_cache_ttl_seconds > 0
+        and settings.evaluator_semantic_cache_max_entries > 0
+    )
+    return AdminUsageEvaluatorCache(
+        enabled=enabled,
+        entries=_safe_usage_int(stats.get("entries", 0)),
+        hits=hits,
+        misses=misses,
+        stores=_safe_usage_int(stats.get("stores", 0)),
+        evictions=_safe_usage_int(stats.get("evictions", 0)),
+        requests=requests,
+        hit_rate=hit_rate,
+    )
+
+
 async def _runtime_health_metrics(*, db_ok: bool) -> RuntimeHealthMetrics:
     metrics = RuntimeHealthMetrics()
 
@@ -401,6 +452,27 @@ async def _usage_from_database(window_days: int) -> AdminUsageResponse:
             """,
             window_days,
         )
+        daily_rows = await conn.fetch(
+            """
+            SELECT
+              to_char(date_trunc('day', created_at), 'YYYY-MM-DD') AS day,
+              COUNT(*) AS turns,
+              COALESCE(SUM(tokens_in), 0)::bigint AS tokens_in,
+              COALESCE(SUM(tokens_out), 0)::bigint AS tokens_out,
+              COALESCE(SUM(cost_usd), 0)::numeric AS cost_usd
+            FROM app.turns
+            WHERE created_at >= now() - ($1::int * interval '1 day')
+              AND speaker = 'client'
+              AND (
+                llm_provider IS NOT NULL OR model IS NOT NULL
+                OR tokens_in IS NOT NULL OR tokens_out IS NOT NULL
+                OR cost_usd IS NOT NULL
+              )
+            GROUP BY 1
+            ORDER BY 1
+            """,
+            window_days,
+        )
 
     total_cost = round(_decimal_to_float(total_row["cost_usd"] if total_row else 0), 6)
     return AdminUsageResponse(
@@ -414,6 +486,7 @@ async def _usage_from_database(window_days: int) -> AdminUsageResponse:
         tokens_out=_safe_usage_int(total_row["tokens_out"] if total_row else 0),
         cost_usd=total_cost,
         budget=_usage_budget(total_cost),
+        evaluator_cache=_usage_evaluator_cache(),
         by_provider=[
             AdminUsageBreakdown(
                 provider=str(row["provider"] or "unknown"),
@@ -425,21 +498,32 @@ async def _usage_from_database(window_days: int) -> AdminUsageResponse:
             )
             for row in rows
         ],
+        daily_cost=[
+            AdminUsageDailyCost(
+                day=str(row["day"]),
+                turns=_safe_usage_int(row["turns"]),
+                tokens_in=_safe_usage_int(row["tokens_in"]),
+                tokens_out=_safe_usage_int(row["tokens_out"]),
+                cost_usd=round(_decimal_to_float(row["cost_usd"]), 6),
+            )
+            for row in daily_rows
+        ],
     )
 
 
 async def _record_health_events(
     *,
-    principal: Principal,
+    principal: Principal | None,
     overall_status: HealthStatus,
     environment: str,
     engine_mode: str,
     services: list[AdminServiceHealth],
-) -> None:
+) -> int:
     if not services:
-        return
+        return 0
+    captured_by = principal.user_id if principal is not None else None
     try:
-        async with acquire(role="admin", user_id=principal.user_id) as conn:
+        async with acquire(role="admin", user_id=captured_by) as conn:
             await conn.executemany(
                 """
                 INSERT INTO app.admin_health_event (
@@ -467,14 +551,15 @@ async def _record_health_events(
                         service.detail,
                         service.metric,
                         service.load,
-                        principal.user_id,
+                        captured_by,
                     )
                     for service in services
                 ],
             )
+            return len(services)
     except Exception:
         # 헬스 화면 자체가 장애 확인 경로라, 이력 적재 실패가 응답을 막으면 안 된다.
-        return
+        return 0
 
 
 async def _uptime_from_database(window_hours: int) -> AdminUptimeResponse:
@@ -498,23 +583,74 @@ async def _uptime_from_database(window_hours: int) -> AdminUptimeResponse:
             """,
             window_hours,
         )
+        rollup_rows = await conn.fetch(
+            """
+            SELECT
+              r.service_key,
+              r.service_name,
+              r.sample_count,
+              r.ok_samples,
+              r.degraded_samples,
+              r.down_samples,
+              r.latest_status,
+              r.last_observed_at,
+              r.last_down_at
+            FROM app.admin_health_daily_rollup AS r
+            WHERE r.rollup_date >= (now() - ($1::int * interval '1 hour'))::date
+              AND r.rollup_date < current_date
+              AND NOT EXISTS (
+                SELECT 1
+                FROM app.admin_health_event AS e
+                WHERE e.observed_at::date = r.rollup_date
+                  AND e.environment = r.environment
+                  AND e.engine_mode = r.engine_mode
+                  AND e.service_key = r.service_key
+              )
+            ORDER BY r.last_observed_at DESC, r.service_key
+            LIMIT 1000
+            """,
+            window_hours,
+        )
 
     events = [_health_event_from_row(row) for row in rows]
     service_buckets: dict[str, AdminUptimeServiceSummary] = {}
-    for event in events:
-        current = service_buckets.get(event.service_key)
+
+    def ensure_bucket(
+        *,
+        service_key: str,
+        service_name: str,
+        latest_status: HealthStatus,
+        latest_observed_at: float | None,
+    ) -> AdminUptimeServiceSummary:
+        current = service_buckets.get(service_key)
         if current is None:
             current = AdminUptimeServiceSummary(
-                service_key=event.service_key,
-                service_name=event.service_name,
+                service_key=service_key,
+                service_name=service_name,
                 samples=0,
                 ok_samples=0,
                 degraded_samples=0,
                 down_samples=0,
-                latest_status=event.service_status,
-                latest_observed_at=event.observed_at,
+                latest_status=latest_status,
+                latest_observed_at=latest_observed_at,
             )
-            service_buckets[event.service_key] = current
+            service_buckets[service_key] = current
+            return current
+        if latest_observed_at is not None and (
+            current.latest_observed_at is None
+            or latest_observed_at > current.latest_observed_at
+        ):
+            current.latest_status = latest_status
+            current.latest_observed_at = latest_observed_at
+        return current
+
+    for event in events:
+        current = ensure_bucket(
+            service_key=event.service_key,
+            service_name=event.service_name,
+            latest_status=event.service_status,
+            latest_observed_at=event.observed_at,
+        )
         current.samples += 1
         if event.service_status == "ok":
             current.ok_samples += 1
@@ -523,9 +659,29 @@ async def _uptime_from_database(window_hours: int) -> AdminUptimeResponse:
         else:
             current.down_samples += 1
 
-    ok_samples = sum(1 for event in events if event.service_status == "ok")
-    sample_count = len(events)
+    rollup_last_down: list[float] = []
+    for row in rollup_rows:
+        latest_observed_at = _row_ts(row["last_observed_at"])
+        current = ensure_bucket(
+            service_key=str(row["service_key"]),
+            service_name=str(row["service_name"]),
+            latest_status=row["latest_status"],
+            latest_observed_at=latest_observed_at,
+        )
+        current.samples += int(row["sample_count"] or 0)
+        current.ok_samples += int(row["ok_samples"] or 0)
+        current.degraded_samples += int(row["degraded_samples"] or 0)
+        current.down_samples += int(row["down_samples"] or 0)
+        last_down = _row_ts(row["last_down_at"])
+        if last_down is not None:
+            rollup_last_down.append(last_down)
+
+    ok_samples = sum(1 for event in events if event.service_status == "ok") + sum(
+        int(row["ok_samples"] or 0) for row in rollup_rows
+    )
+    sample_count = len(events) + sum(int(row["sample_count"] or 0) for row in rollup_rows)
     down_times = [event.observed_at for event in events if event.service_status == "down"]
+    down_times.extend(rollup_last_down)
     return AdminUptimeResponse(
         source="database",
         durable=True,
@@ -533,8 +689,10 @@ async def _uptime_from_database(window_hours: int) -> AdminUptimeResponse:
         generated_at=time.time(),
         sample_count=sample_count,
         ok_ratio=round(ok_samples / sample_count, 4) if sample_count else 0.0,
-        degraded_events=sum(1 for event in events if event.service_status == "degraded"),
-        down_events=sum(1 for event in events if event.service_status == "down"),
+        degraded_events=sum(1 for event in events if event.service_status == "degraded")
+        + sum(int(row["degraded_samples"] or 0) for row in rollup_rows),
+        down_events=sum(1 for event in events if event.service_status == "down")
+        + sum(int(row["down_samples"] or 0) for row in rollup_rows),
         last_down_at=max(down_times) if down_times else None,
         services=sorted(
             service_buckets.values(),
@@ -573,14 +731,32 @@ async def _tickets_from_database(
               t.subject,
               t.body,
               t.source_path,
+              t.fingerprint,
+              t.parent_ticket_id,
               t.assigned_group,
               t.resolution_note,
               t.created_at,
               t.updated_at,
               t.resolved_at,
+              COALESCE(dup.duplicate_count, 0) AS duplicate_count,
+              dup.parent_candidate_id AS duplicate_parent_candidate_id,
+              COALESCE(child.child_ticket_count, 0) AS child_ticket_count,
               COALESCE(ev.event_count, 0) AS event_count,
               ev.last_event_at
             FROM app.support_ticket AS t
+            LEFT JOIN LATERAL (
+              SELECT
+                count(*) FILTER (WHERE id <> t.id)::int AS duplicate_count,
+                (array_agg(id::text ORDER BY created_at ASC, id ASC))[1] AS parent_candidate_id
+              FROM app.support_ticket
+              WHERE fingerprint <> ''
+                AND fingerprint = t.fingerprint
+            ) AS dup ON TRUE
+            LEFT JOIN LATERAL (
+              SELECT count(*)::int AS child_ticket_count
+              FROM app.support_ticket
+              WHERE parent_ticket_id = t.id
+            ) AS child ON TRUE
             LEFT JOIN LATERAL (
               SELECT count(*)::int AS event_count, max(created_at) AS last_event_at
               FROM audit.audit_log
@@ -652,12 +828,14 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse:
     tokens_out = 0
     cost_usd = 0.0
     buckets: dict[tuple[str, str], dict[str, int | float]] = {}
+    daily_buckets: dict[str, dict[str, int | float]] = {}
 
     for sess in store.list():
         for turn in getattr(sess, "turns", []) or []:
             if getattr(turn, "speaker", "") != "client":
                 continue
-            if float(getattr(turn, "created_at", 0.0) or 0.0) < window_start:
+            created_at = float(getattr(turn, "created_at", 0.0) or 0.0)
+            if created_at < window_start:
                 continue
             total_turns += 1
             provider = str(getattr(turn, "llm_provider", None) or "unknown")
@@ -687,6 +865,15 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse:
             bucket["tokens_in"] = int(bucket["tokens_in"]) + turn_tokens_in
             bucket["tokens_out"] = int(bucket["tokens_out"]) + turn_tokens_out
             bucket["cost_usd"] = float(bucket["cost_usd"]) + turn_cost
+            day = datetime.fromtimestamp(created_at, timezone.utc).strftime("%Y-%m-%d")
+            daily_bucket = daily_buckets.setdefault(
+                day,
+                {"turns": 0, "tokens_in": 0, "tokens_out": 0, "cost_usd": 0.0},
+            )
+            daily_bucket["turns"] = int(daily_bucket["turns"]) + 1
+            daily_bucket["tokens_in"] = int(daily_bucket["tokens_in"]) + turn_tokens_in
+            daily_bucket["tokens_out"] = int(daily_bucket["tokens_out"]) + turn_tokens_out
+            daily_bucket["cost_usd"] = float(daily_bucket["cost_usd"]) + turn_cost
 
     by_provider = [
         AdminUsageBreakdown(
@@ -719,7 +906,18 @@ def _usage_from_runtime_store(window_days: int) -> AdminUsageResponse:
         tokens_out=tokens_out,
         cost_usd=total_cost,
         budget=_usage_budget(total_cost),
+        evaluator_cache=_usage_evaluator_cache(),
         by_provider=by_provider,
+        daily_cost=[
+            AdminUsageDailyCost(
+                day=day,
+                turns=int(values["turns"]),
+                tokens_in=int(values["tokens_in"]),
+                tokens_out=int(values["tokens_out"]),
+                cost_usd=round(float(values["cost_usd"]), 6),
+            )
+            for day, values in sorted(daily_buckets.items())
+        ],
     )
 
 
@@ -850,6 +1048,19 @@ def _ticket_from_row(row) -> AdminSupportTicketResponse:
         subject=row["subject"],
         body=row["body"],
         source_path=row["source_path"],
+        fingerprint=_row_value(row, "fingerprint", "") or "",
+        parent_ticket_id=(
+            str(_row_value(row, "parent_ticket_id"))
+            if _row_value(row, "parent_ticket_id") is not None
+            else None
+        ),
+        duplicate_count=int(_row_value(row, "duplicate_count", 0) or 0),
+        duplicate_parent_candidate_id=(
+            str(_row_value(row, "duplicate_parent_candidate_id"))
+            if _row_value(row, "duplicate_parent_candidate_id") is not None
+            else None
+        ),
+        child_ticket_count=int(_row_value(row, "child_ticket_count", 0) or 0),
         assigned_group=row["assigned_group"],
         resolution_note=row["resolution_note"],
         created_at=_row_ts(row["created_at"]) or 0.0,
@@ -925,12 +1136,31 @@ def _unavailable_tickets() -> AdminTicketsResponse:
     )
 
 
+def _normalize_parent_ticket_id(value: str | None) -> str | None:
+    if value is None:
+        return None
+    stripped = value.strip()
+    if not stripped:
+        return None
+    try:
+        return str(UUID(stripped))
+    except ValueError as exc:
+        raise HTTPException(
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="invalid parent ticket id",
+        ) from exc
+
+
 def _ticket_change_detail(old_row, new_row) -> dict[str, object]:
     changed_fields: list[str] = []
     detail: dict[str, object] = {"changed_fields": changed_fields}
-    for field in ("status", "priority", "assigned_group"):
+    for field in ("status", "priority", "assigned_group", "parent_ticket_id"):
         before = _row_value(old_row, field, "")
         after = _row_value(new_row, field, "")
+        if before is not None:
+            before = str(before)
+        if after is not None:
+            after = str(after)
         if before != after:
             changed_fields.append(field)
             detail[field] = {"from": before, "to": after}
@@ -1060,9 +1290,7 @@ async def _admin_user_response(user, *, durable: bool) -> AdminUserResponse:
     )
 
 
-@router.get("/health", response_model=AdminHealthResponse)
-async def admin_health(principal: AdminPrincipal) -> AdminHealthResponse:
-    """Return operational health from live backend checks."""
+async def _build_admin_health_response() -> AdminHealthResponse:
     current_engine = await _current_engine_config()
     db_ok = await healthcheck()
     engine_started = time.perf_counter()
@@ -1148,13 +1376,28 @@ async def admin_health(principal: AdminPrincipal) -> AdminHealthResponse:
         engine_mode=current_engine.engine_mode,
         services=services,
     )
-    await _record_health_events(
+    return response
+
+
+async def record_admin_health_sample(
+    *, principal: Principal | None = None
+) -> tuple[AdminHealthResponse, int]:
+    """Collect and persist one synthetic/admin health sample."""
+    response = await _build_admin_health_response()
+    recorded_count = await _record_health_events(
         principal=principal,
         overall_status=response.status,
         environment=response.environment,
         engine_mode=response.engine_mode,
         services=response.services,
     )
+    return response, recorded_count
+
+
+@router.get("/health", response_model=AdminHealthResponse)
+async def admin_health(principal: AdminPrincipal) -> AdminHealthResponse:
+    """Return operational health from live backend checks."""
+    response, _ = await record_admin_health_sample(principal=principal)
     return response
 
 
@@ -1229,7 +1472,8 @@ async def patch_ticket(
                   status,
                   source_path,
                   assigned_group,
-                  resolution_note
+                  resolution_note,
+                  parent_ticket_id
                 FROM app.support_ticket
                 WHERE id = $1::uuid
                 """,
@@ -1237,6 +1481,44 @@ async def patch_ticket(
             )
             if old_row is None:
                 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="ticket not found")
+            parent_specified = "parent_ticket_id" in body.model_fields_set
+            next_parent_id = (
+                _normalize_parent_ticket_id(body.parent_ticket_id)
+                if parent_specified
+                else None
+            )
+            if parent_specified and next_parent_id is not None:
+                if next_parent_id == str(UUID(ticket_id)):
+                    raise HTTPException(
+                        status.HTTP_422_UNPROCESSABLE_ENTITY,
+                        detail="ticket cannot be its own parent",
+                    )
+                parent_check = await conn.fetchrow(
+                    """
+                    WITH RECURSIVE ancestors AS (
+                      SELECT id, parent_ticket_id
+                      FROM app.support_ticket
+                      WHERE id = $1::uuid
+                      UNION ALL
+                      SELECT t.id, t.parent_ticket_id
+                      FROM app.support_ticket AS t
+                      JOIN ancestors AS a ON t.id = a.parent_ticket_id
+                      WHERE a.parent_ticket_id IS NOT NULL
+                    )
+                    SELECT
+                      EXISTS (SELECT 1 FROM ancestors) AS parent_exists,
+                      EXISTS (SELECT 1 FROM ancestors WHERE id = $2::uuid) AS creates_cycle
+                    """,
+                    next_parent_id,
+                    ticket_id,
+                )
+                if not parent_check or not parent_check["parent_exists"]:
+                    raise HTTPException(status.HTTP_404_NOT_FOUND, detail="parent ticket not found")
+                if parent_check["creates_cycle"]:
+                    raise HTTPException(
+                        status.HTTP_422_UNPROCESSABLE_ENTITY,
+                        detail="ticket parent would create a cycle",
+                    )
             row = await conn.fetchrow(
                 """
                 UPDATE app.support_ticket SET
@@ -1244,6 +1526,10 @@ async def patch_ticket(
                   priority = COALESCE($3, priority),
                   assigned_group = COALESCE($4, assigned_group),
                   resolution_note = COALESCE($5, resolution_note),
+                  parent_ticket_id = CASE
+                    WHEN $6::bool THEN $7::uuid
+                    ELSE parent_ticket_id
+                  END,
                   resolved_at = CASE
                     WHEN COALESCE($2, status) IN ('resolved', 'closed')
                       THEN COALESCE(resolved_at, now())
@@ -1266,6 +1552,8 @@ async def patch_ticket(
                   subject,
                   body,
                   source_path,
+                  fingerprint,
+                  parent_ticket_id,
                   assigned_group,
                   resolution_note,
                   created_at,
@@ -1277,6 +1565,8 @@ async def patch_ticket(
                 body.priority,
                 body.assigned_group.strip() if body.assigned_group is not None else None,
                 body.resolution_note.strip() if body.resolution_note is not None else None,
+                parent_specified,
+                next_parent_id,
             )
             if row is None:
                 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="ticket not found")
@@ -1302,14 +1592,32 @@ async def patch_ticket(
                       t.subject,
                       t.body,
                       t.source_path,
+                      t.fingerprint,
+                      t.parent_ticket_id,
                       t.assigned_group,
                       t.resolution_note,
                       t.created_at,
                       t.updated_at,
                       t.resolved_at,
+                      COALESCE(dup.duplicate_count, 0) AS duplicate_count,
+                      dup.parent_candidate_id AS duplicate_parent_candidate_id,
+                      COALESCE(child.child_ticket_count, 0) AS child_ticket_count,
                       COALESCE(ev.event_count, 0) AS event_count,
                       ev.last_event_at
                     FROM app.support_ticket AS t
+                    LEFT JOIN LATERAL (
+                      SELECT
+                        count(*) FILTER (WHERE id <> t.id)::int AS duplicate_count,
+                        (array_agg(id::text ORDER BY created_at ASC, id ASC))[1] AS parent_candidate_id
+                      FROM app.support_ticket
+                      WHERE fingerprint <> ''
+                        AND fingerprint = t.fingerprint
+                    ) AS dup ON TRUE
+                    LEFT JOIN LATERAL (
+                      SELECT count(*)::int AS child_ticket_count
+                      FROM app.support_ticket
+                      WHERE parent_ticket_id = t.id
+                    ) AS child ON TRUE
                     LEFT JOIN LATERAL (
                       SELECT count(*)::int AS event_count, max(created_at) AS last_event_at
                       FROM audit.audit_log
diff --git a/apps/api/app/routes/users.py b/apps/api/app/routes/users.py
index 3691e85..b0d1314 100644
--- a/apps/api/app/routes/users.py
+++ b/apps/api/app/routes/users.py
@@ -3,11 +3,12 @@
 from __future__ import annotations
 
 import secrets
+import time
 from pathlib import Path
 from typing import Literal
 
-from fastapi import APIRouter, File, HTTPException, UploadFile, status
-from pydantic import BaseModel, Field
+from fastapi import APIRouter, File, HTTPException, Query, UploadFile, status
+from pydantic import BaseModel, Field, model_validator
 
 from ..auth_types import RoleName
 from ..auth_sessions import (
@@ -21,6 +22,7 @@ from ..config import settings
 from ..db import acquire, get_pool
 from ..deps import CurrentPrincipal, Role
 from ..runtime_policy import require_runtime_fallback_allowed
+from ..services.support_tickets import support_ticket_fingerprint
 from ..services.voice import PRESET_RATE, PRESET_TO_OPENAI_VOICE
 
 router = APIRouter(prefix="/users", tags=["users"])
@@ -33,6 +35,14 @@ AVATAR_CONTENT_TYPES = {
     "image/jpeg": ("jpg", b"\xff\xd8\xff"),
     "image/webp": ("webp", b"RIFF"),
 }
+DEFAULT_PREPOST_PILOT_ID = "phase3-pilot-draft"
+DEFAULT_PREPOST_INSTRUMENT_VERSION = "pilot-prepost-scaffold-2026-06-28"
+PREPOST_MEASURE_NAMES = (
+    "self_efficacy",
+    "skill_proficiency",
+    "training_satisfaction",
+)
+PREPOST_TIMEPOINTS = ("pre", "post")
 
 TERMS_BODY = """Vignette 서비스 이용약관 초안
 
@@ -209,6 +219,9 @@ TicketCategory = Literal[
     "other",
 ]
 TicketPriority = Literal["low", "normal", "high", "urgent"]
+TicketStatus = Literal["open", "triaged", "in_progress", "resolved", "closed"]
+PrepostMeasureName = Literal["self_efficacy", "skill_proficiency", "training_satisfaction"]
+PrepostTimepoint = Literal["pre", "post"]
 
 
 class UserSupportTicketRequest(BaseModel):
@@ -228,6 +241,76 @@ class UserSupportTicketResponse(BaseModel):
     created_at: float
 
 
+class UserSupportTicketListItem(BaseModel):
+    ticket_id: str
+    status: TicketStatus
+    category: TicketCategory
+    priority: TicketPriority
+    subject: str
+    source_path: str
+    assigned_group: str
+    resolution_note: str
+    created_at: float
+    updated_at: float
+    resolved_at: float | None = None
+
+
+class UserSupportTicketsResponse(BaseModel):
+    source: Literal["database"]
+    durable: bool
+    generated_at: float
+    tickets: list[UserSupportTicketListItem]
+
+
+class UserPrepostMeasureRequest(BaseModel):
+    pilot_id: str = Field(default=DEFAULT_PREPOST_PILOT_ID, min_length=1, max_length=80)
+    measure_name: PrepostMeasureName
+    timepoint: PrepostTimepoint
+    raw_score: float
+    min_score: float = 1.0
+    max_score: float = 5.0
+    instrument_version: str = Field(
+        default=DEFAULT_PREPOST_INSTRUMENT_VERSION,
+        min_length=1,
+        max_length=120,
+    )
+    item_count: int = Field(default=1, ge=1, le=80)
+
+    @model_validator(mode="after")
+    def validate_score_range(self) -> "UserPrepostMeasureRequest":
+        if self.max_score <= self.min_score:
+            raise ValueError("max_score must be greater than min_score")
+        if self.raw_score < self.min_score or self.raw_score > self.max_score:
+            raise ValueError("raw_score must be within min_score and max_score")
+        return self
+
+
+class UserPrepostMeasureItem(BaseModel):
+    measure_id: str
+    pilot_id: str
+    measure_name: PrepostMeasureName
+    timepoint: PrepostTimepoint
+    raw_score: float
+    min_score: float
+    max_score: float
+    normalized_score: float
+    instrument_version: str
+    item_count: int
+    collected_at: float
+    updated_at: float
+
+
+class UserPrepostMeasuresResponse(BaseModel):
+    source: Literal["database"]
+    durable: bool
+    generated_at: float
+    pilot_id: str
+    required_measure_names: list[PrepostMeasureName]
+    required_timepoints: list[PrepostTimepoint]
+    complete_measure_pairs: int
+    measures: list[UserPrepostMeasureItem]
+
+
 _preferences: dict[str, UserPreferencesResponse] = {}
 
 VOICE_PRESET_META = {
@@ -514,6 +597,15 @@ async def create_support_ticket(
     profile = await _profile_for(principal)
     try:
         async with acquire(role=principal.role.value, user_id=principal.user_id) as conn:
+            subject = body.subject.strip()
+            ticket_body = body.body.strip()
+            source_path = body.source_path.strip()
+            fingerprint = support_ticket_fingerprint(
+                category=body.category,
+                subject=subject,
+                body=ticket_body,
+                source_path=source_path,
+            )
             row = await conn.fetchrow(
                 """
                 INSERT INTO app.support_ticket (
@@ -525,7 +617,8 @@ async def create_support_ticket(
                   priority,
                   subject,
                   body,
-                  source_path
+                  source_path,
+                  fingerprint
                 )
                 VALUES (
                   $1::uuid,
@@ -536,7 +629,8 @@ async def create_support_ticket(
                   $6,
                   $7,
                   $8,
-                  $9
+                  $9,
+                  $10
                 )
                 RETURNING id, category, priority, subject, EXTRACT(EPOCH FROM created_at) AS created_at
                 """,
@@ -546,9 +640,10 @@ async def create_support_ticket(
                 principal.role.value,
                 body.category,
                 body.priority,
-                body.subject.strip(),
-                body.body.strip(),
-                body.source_path.strip(),
+                subject,
+                ticket_body,
+                source_path,
+                fingerprint,
             )
             await conn.execute(
                 """
@@ -564,9 +659,10 @@ async def create_support_ticket(
                 {
                     "category": row["category"],
                     "priority": row["priority"],
-                    "source_path": body.source_path.strip(),
-                    "subject_present": bool(body.subject.strip()),
-                    "body_present": bool(body.body.strip()),
+                    "source_path": source_path,
+                    "fingerprint": fingerprint,
+                    "subject_present": bool(subject),
+                    "body_present": bool(ticket_body),
                 },
             )
     except Exception as exc:
@@ -584,6 +680,244 @@ async def create_support_ticket(
     )
 
 
+async def _support_tickets_for_user(
+    principal: CurrentPrincipal,
+    *,
+    limit: int = 20,
+) -> UserSupportTicketsResponse:
+    async with acquire(role=principal.role.value, user_id=principal.user_id) as conn:
+        rows = await conn.fetch(
+            """
+            SELECT
+              id,
+              category,
+              priority,
+              status,
+              subject,
+              source_path,
+              assigned_group,
+              resolution_note,
+              EXTRACT(EPOCH FROM created_at) AS created_at,
+              EXTRACT(EPOCH FROM updated_at) AS updated_at,
+              EXTRACT(EPOCH FROM resolved_at) AS resolved_at
+            FROM app.support_ticket
+            WHERE reporter_id = $1::uuid
+            ORDER BY created_at DESC
+            LIMIT $2
+            """,
+            principal.user_id,
+            limit,
+        )
+    return UserSupportTicketsResponse(
+        source="database",
+        durable=True,
+        generated_at=time.time(),
+        tickets=[
+            UserSupportTicketListItem(
+                ticket_id=str(row["id"]),
+                status=row["status"],
+                category=row["category"],
+                priority=row["priority"],
+                subject=row["subject"],
+                source_path=row["source_path"] or "",
+                assigned_group=row["assigned_group"] or "",
+                resolution_note=row["resolution_note"] or "",
+                created_at=float(row["created_at"] or 0.0),
+                updated_at=float(row["updated_at"] or 0.0),
+                resolved_at=float(row["resolved_at"]) if row["resolved_at"] is not None else None,
+            )
+            for row in rows
+        ],
+    )
+
+
+@router.get("/support-tickets", response_model=UserSupportTicketsResponse)
+async def list_my_support_tickets(
+    principal: CurrentPrincipal,
+) -> UserSupportTicketsResponse:
+    try:
+        return await _support_tickets_for_user(principal)
+    except Exception as exc:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+        detail="support ticket persistence unavailable",
+        ) from exc
+
+
+def _normalized_prepost_score(raw_score: float, min_score: float, max_score: float) -> float:
+    if max_score <= min_score:
+        return 0.0
+    return round(((raw_score - min_score) / (max_score - min_score)) * 100.0, 3)
+
+
+def _prepost_measure_from_row(row) -> UserPrepostMeasureItem:
+    raw_score = float(row["raw_score"])
+    min_score = float(row["min_score"])
+    max_score = float(row["max_score"])
+    return UserPrepostMeasureItem(
+        measure_id=str(row["id"]),
+        pilot_id=row["pilot_id"],
+        measure_name=row["measure_name"],
+        timepoint=row["timepoint"],
+        raw_score=raw_score,
+        min_score=min_score,
+        max_score=max_score,
+        normalized_score=_normalized_prepost_score(raw_score, min_score, max_score),
+        instrument_version=row["instrument_version"],
+        item_count=int(row["item_count"]),
+        collected_at=float(row["collected_at"] or 0.0),
+        updated_at=float(row["updated_at"] or 0.0),
+    )
+
+
+async def _prepost_measures_for_user(
+    principal: CurrentPrincipal,
+    *,
+    pilot_id: str = DEFAULT_PREPOST_PILOT_ID,
+) -> UserPrepostMeasuresResponse:
+    async with acquire(role=principal.role.value, user_id=principal.user_id) as conn:
+        rows = await conn.fetch(
+            """
+            SELECT
+              id,
+              pilot_id,
+              measure_name,
+              timepoint,
+              raw_score,
+              min_score,
+              max_score,
+              instrument_version,
+              item_count,
+              EXTRACT(EPOCH FROM collected_at) AS collected_at,
+              EXTRACT(EPOCH FROM updated_at) AS updated_at
+            FROM app.learner_prepost_measure
+            WHERE learner_id = $1::uuid
+              AND pilot_id = $2
+            ORDER BY measure_name, timepoint, instrument_version
+            """,
+            principal.user_id,
+            pilot_id.strip() or DEFAULT_PREPOST_PILOT_ID,
+        )
+    measures = [_prepost_measure_from_row(row) for row in rows]
+    pairs = {
+        item.measure_name
+        for item in measures
+        if {m.timepoint for m in measures if m.measure_name == item.measure_name} == {"pre", "post"}
+    }
+    return UserPrepostMeasuresResponse(
+        source="database",
+        durable=True,
+        generated_at=time.time(),
+        pilot_id=pilot_id.strip() or DEFAULT_PREPOST_PILOT_ID,
+        required_measure_names=list(PREPOST_MEASURE_NAMES),
+        required_timepoints=list(PREPOST_TIMEPOINTS),
+        complete_measure_pairs=len(pairs),
+        measures=measures,
+    )
+
+
+@router.get("/me/prepost-measures", response_model=UserPrepostMeasuresResponse)
+async def list_my_prepost_measures(
+    principal: CurrentPrincipal,
+    pilot_id: str = Query(default=DEFAULT_PREPOST_PILOT_ID, min_length=1, max_length=80),
+) -> UserPrepostMeasuresResponse:
+    try:
+        return await _prepost_measures_for_user(principal, pilot_id=pilot_id)
+    except Exception as exc:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="prepost measure persistence unavailable",
+        ) from exc
+
+
+@router.put("/me/prepost-measures", response_model=UserPrepostMeasureItem)
+async def upsert_my_prepost_measure(
+    body: UserPrepostMeasureRequest,
+    principal: CurrentPrincipal,
+) -> UserPrepostMeasureItem:
+    pilot_id = body.pilot_id.strip() or DEFAULT_PREPOST_PILOT_ID
+    instrument_version = body.instrument_version.strip() or DEFAULT_PREPOST_INSTRUMENT_VERSION
+    try:
+        async with acquire(role=principal.role.value, user_id=principal.user_id) as conn:
+            row = await conn.fetchrow(
+                """
+                INSERT INTO app.learner_prepost_measure (
+                  learner_id,
+                  pilot_id,
+                  measure_name,
+                  timepoint,
+                  raw_score,
+                  min_score,
+                  max_score,
+                  instrument_version,
+                  item_count
+                )
+                VALUES ($1::uuid, $2, $3, $4, $5, $6, $7, $8, $9)
+                ON CONFLICT (
+                  learner_id,
+                  pilot_id,
+                  measure_name,
+                  timepoint,
+                  instrument_version
+                )
+                DO UPDATE SET
+                  raw_score = EXCLUDED.raw_score,
+                  min_score = EXCLUDED.min_score,
+                  max_score = EXCLUDED.max_score,
+                  item_count = EXCLUDED.item_count,
+                  collected_at = now(),
+                  updated_at = now()
+                RETURNING
+                  id,
+                  pilot_id,
+                  measure_name,
+                  timepoint,
+                  raw_score,
+                  min_score,
+                  max_score,
+                  instrument_version,
+                  item_count,
+                  EXTRACT(EPOCH FROM collected_at) AS collected_at,
+                  EXTRACT(EPOCH FROM updated_at) AS updated_at
+                """,
+                principal.user_id,
+                pilot_id,
+                body.measure_name,
+                body.timepoint,
+                body.raw_score,
+                body.min_score,
+                body.max_score,
+                instrument_version,
+                body.item_count,
+            )
+            await conn.execute(
+                """
+                INSERT INTO audit.audit_log (
+                  actor_uid, action, target_kind, target_id, detail
+                )
+                VALUES ($1::uuid, $2, $3, $4, $5::jsonb)
+                """,
+                principal.user_id,
+                "prepost_measure_upsert",
+                "learner_prepost_measure",
+                str(row["id"]),
+                {
+                    "pilot_id": pilot_id,
+                    "measure_name": body.measure_name,
+                    "timepoint": body.timepoint,
+                    "instrument_version": instrument_version,
+                    "item_count": body.item_count,
+                    "score_recorded": True,
+                },
+            )
+    except Exception as exc:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="prepost measure persistence unavailable",
+        ) from exc
+    return _prepost_measure_from_row(row)
+
+
 @router.get("/me/preferences", response_model=UserPreferencesResponse)
 async def get_preferences(principal: CurrentPrincipal) -> UserPreferencesResponse:
     try:
diff --git a/apps/api/app/services/admin_health_maintenance.py b/apps/api/app/services/admin_health_maintenance.py
new file mode 100644
index 0000000..17be57d
--- /dev/null
+++ b/apps/api/app/services/admin_health_maintenance.py
@@ -0,0 +1,177 @@
+"""Maintenance helpers for persisted admin health samples."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ..db import acquire
+
+
+@dataclass(frozen=True)
+class AdminHealthMaintenanceResult:
+    applied: bool
+    rollup_days: int
+    retention_days: int
+    rollup_event_count: int
+    rollup_bucket_count: int
+    upserted_rollups: int
+    prunable_event_count: int
+    deleted_events: int
+
+
+def _validate_windows(*, rollup_days: int, retention_days: int) -> None:
+    if rollup_days < 1:
+        raise ValueError("rollup_days must be 1 or greater")
+    if retention_days < rollup_days:
+        raise ValueError("retention_days must be greater than or equal to rollup_days")
+
+
+def _deleted_count(command_tag: str) -> int:
+    parts = (command_tag or "").split()
+    if len(parts) >= 2 and parts[0].upper() == "DELETE":
+        try:
+            return int(parts[-1])
+        except ValueError:
+            return 0
+    return 0
+
+
+def _row_int(row, key: str) -> int:
+    if row is None:
+        return 0
+    try:
+        return int(row[key] or 0)
+    except (KeyError, TypeError, ValueError):
+        return 0
+
+
+async def maintain_admin_health_events(
+    *,
+    rollup_days: int,
+    retention_days: int,
+    apply: bool = False,
+) -> AdminHealthMaintenanceResult:
+    """Roll up old health samples and optionally prune retained raw rows.
+
+    Dry-run mode returns the affected event and bucket counts without mutating
+    data.  Apply mode upserts daily service rollups first, then deletes only raw
+    events older than the retention window.
+    """
+    _validate_windows(rollup_days=rollup_days, retention_days=retention_days)
+    async with acquire(role="admin") as conn:
+        rollup_stats = await conn.fetchrow(
+            """
+            SELECT
+              COUNT(*)::int AS event_count,
+              COUNT(DISTINCT (observed_at::date, environment, engine_mode, service_key))::int
+                AS bucket_count
+            FROM app.admin_health_event
+            WHERE observed_at::date < current_date - $1::int
+            """,
+            rollup_days,
+        )
+        prune_stats = await conn.fetchrow(
+            """
+            SELECT COUNT(*)::int AS event_count
+            FROM app.admin_health_event
+            WHERE observed_at::date < current_date - $1::int
+            """,
+            retention_days,
+        )
+        upserted_rollups = 0
+        deleted_events = 0
+        if apply:
+            rows = await conn.fetch(
+                """
+                WITH rolled AS (
+                  SELECT
+                    observed_at::date AS rollup_date,
+                    environment,
+                    engine_mode,
+                    service_key,
+                    (array_agg(service_name ORDER BY observed_at DESC, id DESC))[1] AS service_name,
+                    COUNT(*)::int AS sample_count,
+                    COUNT(*) FILTER (WHERE service_status = 'ok')::int AS ok_samples,
+                    COUNT(*) FILTER (WHERE service_status = 'degraded')::int AS degraded_samples,
+                    COUNT(*) FILTER (WHERE service_status = 'down')::int AS down_samples,
+                    MIN(observed_at) AS first_observed_at,
+                    MAX(observed_at) AS last_observed_at,
+                    (array_agg(service_status ORDER BY observed_at DESC, id DESC))[1] AS latest_status,
+                    MAX(observed_at) FILTER (WHERE service_status = 'down') AS last_down_at,
+                    MAX(load)::real AS max_load
+                  FROM app.admin_health_event
+                  WHERE observed_at::date < current_date - $1::int
+                  GROUP BY observed_at::date, environment, engine_mode, service_key
+                )
+                INSERT INTO app.admin_health_daily_rollup (
+                  rollup_date,
+                  environment,
+                  engine_mode,
+                  service_key,
+                  service_name,
+                  sample_count,
+                  ok_samples,
+                  degraded_samples,
+                  down_samples,
+                  first_observed_at,
+                  last_observed_at,
+                  latest_status,
+                  last_down_at,
+                  max_load,
+                  generated_at
+                )
+                SELECT
+                  rollup_date,
+                  environment,
+                  engine_mode,
+                  service_key,
+                  service_name,
+                  sample_count,
+                  ok_samples,
+                  degraded_samples,
+                  down_samples,
+                  first_observed_at,
+                  last_observed_at,
+                  latest_status,
+                  last_down_at,
+                  max_load,
+                  now()
+                FROM rolled
+                ON CONFLICT (rollup_date, environment, engine_mode, service_key)
+                DO UPDATE SET
+                  service_name = EXCLUDED.service_name,
+                  sample_count = EXCLUDED.sample_count,
+                  ok_samples = EXCLUDED.ok_samples,
+                  degraded_samples = EXCLUDED.degraded_samples,
+                  down_samples = EXCLUDED.down_samples,
+                  first_observed_at = EXCLUDED.first_observed_at,
+                  last_observed_at = EXCLUDED.last_observed_at,
+                  latest_status = EXCLUDED.latest_status,
+                  last_down_at = EXCLUDED.last_down_at,
+                  max_load = EXCLUDED.max_load,
+                  generated_at = now()
+                RETURNING 1
+                """,
+                rollup_days,
+            )
+            upserted_rollups = len(rows)
+            deleted_events = _deleted_count(
+                await conn.execute(
+                    """
+                    DELETE FROM app.admin_health_event
+                    WHERE observed_at::date < current_date - $1::int
+                    """,
+                    retention_days,
+                )
+            )
+
+    return AdminHealthMaintenanceResult(
+        applied=apply,
+        rollup_days=rollup_days,
+        retention_days=retention_days,
+        rollup_event_count=_row_int(rollup_stats, "event_count"),
+        rollup_bucket_count=_row_int(rollup_stats, "bucket_count"),
+        upserted_rollups=upserted_rollups,
+        prunable_event_count=_row_int(prune_stats, "event_count"),
+        deleted_events=deleted_events,
+    )
diff --git a/apps/api/app/services/phase3_kpi_export.py b/apps/api/app/services/phase3_kpi_export.py
new file mode 100644
index 0000000..b396817
--- /dev/null
+++ b/apps/api/app/services/phase3_kpi_export.py
@@ -0,0 +1,276 @@
+"""Phase 3 KPI evidence export helpers.
+
+This module turns persisted pre/post aggregate scores into the Phase 3 evidence
+shape checked by scripts/check-phase3-artifacts.py. It does not claim clinical
+effectiveness; it only produces pilot evidence files for operator review.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+from collections import defaultdict
+from datetime import UTC, date, datetime
+from decimal import Decimal
+from pathlib import Path
+from typing import Any, Iterable, Mapping, Sequence
+from uuid import UUID
+
+PREPOST_MEASURE_NAMES = (
+    "self_efficacy",
+    "skill_proficiency",
+    "training_satisfaction",
+)
+PREPOST_TIMEPOINTS = ("pre", "post")
+PHASE3_KPI_METRICS = (
+    "embedding_consistency",
+    "hallucination_rate",
+    "icc",
+    "inter_rater_kappa",
+    "pilot_completion",
+    "self_efficacy_prepost",
+    "session_completion",
+    "sus",
+    "top1",
+)
+PREPOST_CSV_PATH = "02-measures/prepost_measures.csv"
+KPI_REPORT_PATH = "02-measures/kpi_report.json"
+
+
+class ParticipantKeys:
+    def __init__(self) -> None:
+        self._keys: dict[str, str] = {}
+
+    def key(self, raw_id: Any) -> str:
+        value = str(raw_id or "unknown-participant")
+        if value not in self._keys:
+            self._keys[value] = f"P3-{len(self._keys) + 1:03d}"
+        return self._keys[value]
+
+    def __len__(self) -> int:
+        return len(self._keys)
+
+
+def json_safe(value: Any) -> Any:
+    if isinstance(value, datetime):
+        if value.tzinfo is None:
+            value = value.replace(tzinfo=UTC)
+        return value.isoformat().replace("+00:00", "Z")
+    if isinstance(value, date):
+        return value.isoformat()
+    if isinstance(value, (Decimal, UUID)):
+        return str(value)
+    if isinstance(value, Mapping):
+        return {str(key): json_safe(item) for key, item in value.items()}
+    if isinstance(value, list):
+        return [json_safe(item) for item in value]
+    if isinstance(value, tuple):
+        return [json_safe(item) for item in value]
+    return value
+
+
+def iso_timestamp(value: Any) -> str:
+    safe = json_safe(value)
+    return safe if isinstance(safe, str) else str(safe or "")
+
+
+def normalized_score(raw_score: float, min_score: float, max_score: float) -> float:
+    if max_score <= min_score:
+        return 0.0
+    return round(((raw_score - min_score) / (max_score - min_score)) * 100.0, 3)
+
+
+def latest_prepost_rows(rows: Iterable[Mapping[str, Any]]) -> list[dict[str, Any]]:
+    latest: dict[tuple[str, str, str], dict[str, Any]] = {}
+    for row in rows:
+        learner_id = str(row.get("learner_id") or row.get("participant_id") or "")
+        measure_name = str(row.get("measure_name") or "")
+        timepoint = str(row.get("timepoint") or "")
+        if measure_name not in PREPOST_MEASURE_NAMES or timepoint not in PREPOST_TIMEPOINTS:
+            continue
+        key = (learner_id, measure_name, timepoint)
+        current = dict(row)
+        current_order = iso_timestamp(current.get("updated_at") or current.get("collected_at"))
+        previous = latest.get(key)
+        previous_order = iso_timestamp(previous.get("updated_at") or previous.get("collected_at")) if previous else ""
+        if previous is None or current_order >= previous_order:
+            latest[key] = current
+    return sorted(
+        latest.values(),
+        key=lambda item: (
+            str(item.get("learner_id") or item.get("participant_id") or ""),
+            str(item.get("measure_name") or ""),
+            str(item.get("timepoint") or ""),
+        ),
+    )
+
+
+def build_prepost_csv_rows(
+    rows: Iterable[Mapping[str, Any]],
+    *,
+    participant_keys: ParticipantKeys | None = None,
+) -> list[dict[str, str]]:
+    keys = participant_keys if participant_keys is not None else ParticipantKeys()
+    output: list[dict[str, str]] = []
+    for row in latest_prepost_rows(rows):
+        raw_score = float(row.get("raw_score") or row.get("score") or 0.0)
+        output.append(
+            {
+                "participant_id": keys.key(row.get("learner_id") or row.get("participant_id")),
+                "measure_name": str(row.get("measure_name") or ""),
+                "timepoint": str(row.get("timepoint") or ""),
+                "score": _format_number(raw_score),
+                "collected_at": iso_timestamp(row.get("collected_at") or row.get("updated_at")),
+            }
+        )
+    return output
+
+
+def paired_prepost_summary(rows: Iterable[Mapping[str, Any]], measure_name: str) -> dict[str, Any]:
+    by_participant: dict[str, dict[str, float]] = defaultdict(dict)
+    for row in latest_prepost_rows(rows):
+        if str(row.get("measure_name") or "") != measure_name:
+            continue
+        participant_id = str(row.get("learner_id") or row.get("participant_id") or "")
+        raw_score = float(row.get("raw_score") or row.get("score") or 0.0)
+        min_score = float(row.get("min_score") or 1.0)
+        max_score = float(row.get("max_score") or 5.0)
+        by_participant[participant_id][str(row.get("timepoint") or "")] = normalized_score(
+            raw_score,
+            min_score,
+            max_score,
+        )
+
+    deltas: list[float] = []
+    pre_values: list[float] = []
+    post_values: list[float] = []
+    missing_pairs = 0
+    for values in by_participant.values():
+        if "pre" not in values or "post" not in values:
+            missing_pairs += 1
+            continue
+        pre_values.append(values["pre"])
+        post_values.append(values["post"])
+        deltas.append(values["post"] - values["pre"])
+
+    return {
+        "participants_with_any_measure": len(by_participant),
+        "complete_pairs": len(deltas),
+        "missing_pairs": missing_pairs,
+        "mean_pre": _mean(pre_values),
+        "mean_post": _mean(post_values),
+        "mean_delta": _mean(deltas),
+    }
+
+
+def build_kpi_report(
+    rows: Sequence[Mapping[str, Any]],
+    *,
+    pilot_id: str,
+    generated_at: str,
+    source_window: Mapping[str, Any] | None = None,
+    review_operator: str = "",
+) -> dict[str, Any]:
+    latest_rows = latest_prepost_rows(rows)
+    participants = {str(row.get("learner_id") or row.get("participant_id") or "") for row in latest_rows}
+    metrics = {name: _placeholder_metric(name) for name in PHASE3_KPI_METRICS}
+
+    self_efficacy = paired_prepost_summary(latest_rows, "self_efficacy")
+    metrics["self_efficacy_prepost"] = {
+        "value": self_efficacy["mean_delta"],
+        "threshold": 0.0,
+        "pass": False,
+        "numerator": self_efficacy["complete_pairs"],
+        "denominator": max(self_efficacy["participants_with_any_measure"], 0),
+        "method": "paired normalized post-pre delta for pilot review; no official pass/fail gate",
+        "source_files": [PREPOST_CSV_PATH],
+        "mean_pre": self_efficacy["mean_pre"],
+        "mean_post": self_efficacy["mean_post"],
+        "mean_delta": self_efficacy["mean_delta"],
+        "complete_pairs": self_efficacy["complete_pairs"],
+        "missing_pairs": self_efficacy["missing_pairs"],
+    }
+
+    for measure_name in ("skill_proficiency", "training_satisfaction"):
+        summary = paired_prepost_summary(latest_rows, measure_name)
+        metrics[f"{measure_name}_prepost"] = {
+            **_placeholder_metric(f"{measure_name}_prepost"),
+            **summary,
+            "value": summary["mean_delta"],
+            "numerator": summary["complete_pairs"],
+            "denominator": summary["participants_with_any_measure"],
+            "method": "paired normalized post-pre delta for pilot review; not a required KPI gate yet",
+            "source_files": [PREPOST_CSV_PATH],
+        }
+
+    return {
+        "pilot_id": pilot_id,
+        "generated_at": generated_at,
+        "source_window": dict(source_window or _source_window(latest_rows)),
+        "cohort_size": len(participants),
+        "metrics": metrics,
+        "exclusions": [],
+        "open_schema_gaps": [
+            "official item text and validated scoring rules are not encoded here",
+            "experimental/control assignment and statistical testing require owner/evaluation-design approval",
+        ],
+        "review": {
+            "operator": review_operator,
+            "reviewed_at": "",
+            "decision": "pending",
+        },
+    }
+
+
+def write_prepost_csv(rows: Sequence[Mapping[str, str]], path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", encoding="utf-8", newline="") as handle:
+        writer = csv.DictWriter(
+            handle,
+            fieldnames=("participant_id", "measure_name", "timepoint", "score", "collected_at"),
+        )
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def write_kpi_report(report: Mapping[str, Any], path: Path) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(json_safe(report), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+
+
+def _placeholder_metric(name: str) -> dict[str, Any]:
+    return {
+        "value": 0.0,
+        "threshold": 0.0,
+        "pass": False,
+        "numerator": 0,
+        "denominator": 0,
+        "method": f"not computed by prepost export scaffold: {name}",
+        "source_files": [],
+    }
+
+
+def _source_window(rows: Sequence[Mapping[str, Any]]) -> dict[str, str]:
+    timestamps = [
+        iso_timestamp(row.get("collected_at") or row.get("updated_at"))
+        for row in rows
+        if iso_timestamp(row.get("collected_at") or row.get("updated_at"))
+    ]
+    if not timestamps:
+        return {"started_at": "", "ended_at": ""}
+    return {"started_at": min(timestamps), "ended_at": max(timestamps)}
+
+
+def _mean(values: Sequence[float]) -> float:
+    if not values:
+        return 0.0
+    return round(sum(values) / len(values), 3)
+
+
+def _format_number(value: float) -> str:
+    if value.is_integer():
+        return str(int(value))
+    return f"{value:.3f}".rstrip("0").rstrip(".")
diff --git a/apps/api/app/services/support_tickets.py b/apps/api/app/services/support_tickets.py
new file mode 100644
index 0000000..399f6b8
--- /dev/null
+++ b/apps/api/app/services/support_tickets.py
@@ -0,0 +1,31 @@
+"""Support-ticket helpers shared across user and admin routes."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+
+_SPACE_RE = re.compile(r"\s+")
+
+
+def _normalize_fingerprint_part(value: str | None) -> str:
+    return _SPACE_RE.sub(" ", (value or "").strip().lower())
+
+
+def support_ticket_fingerprint(
+    *,
+    category: str,
+    subject: str,
+    body: str,
+    source_path: str,
+) -> str:
+    """Return a stable hash for exact-ish duplicate support-ticket hints."""
+    payload = {
+        "body": _normalize_fingerprint_part(body),
+        "category": _normalize_fingerprint_part(category),
+        "source_path": _normalize_fingerprint_part(source_path),
+        "subject": _normalize_fingerprint_part(subject),
+    }
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
diff --git a/apps/api/app/services/usage_report.py b/apps/api/app/services/usage_report.py
new file mode 100644
index 0000000..ada98c8
--- /dev/null
+++ b/apps/api/app/services/usage_report.py
@@ -0,0 +1,134 @@
+"""AI usage cost verification reports.
+
+The admin API already owns collection. This module turns that existing usage
+shape into a deterministic model/provider cost report for ops evidence without
+adding any enforcement policy.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+
+REPORT_SCHEMA = "vignette.ai_usage_model_cost_report.v1"
+
+
+def _number(value: Any, default: float = 0.0) -> float:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _integer(value: Any, default: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _ratio(part: float, whole: float) -> float:
+    if whole <= 0:
+        return 0.0
+    return round(part / whole, 6)
+
+
+def _cost_per_1k_tokens(cost_usd: float, tokens: int) -> float | None:
+    if tokens <= 0:
+        return None
+    return round(cost_usd / tokens * 1000.0, 6)
+
+
+def _cost_per_turn(cost_usd: float, turns: int) -> float | None:
+    if turns <= 0:
+        return None
+    return round(cost_usd / turns, 6)
+
+
+def build_model_cost_report(usage: Mapping[str, Any]) -> dict[str, Any]:
+    """Build an ops report from an AdminUsageResponse-like mapping."""
+    total_cost = round(_number(usage.get("cost_usd")), 6)
+    total_turns = _integer(usage.get("total_turns"))
+    metered_turns = _integer(usage.get("metered_turns"))
+    tokens_in = _integer(usage.get("tokens_in"))
+    tokens_out = _integer(usage.get("tokens_out"))
+    total_tokens = tokens_in + tokens_out
+    by_provider = list(usage.get("by_provider") or [])
+    budget = dict(usage.get("budget") or {})
+    evaluator_cache = dict(usage.get("evaluator_cache") or {})
+
+    models: list[dict[str, Any]] = []
+    for item in by_provider:
+        row = dict(item or {})
+        turns = _integer(row.get("turns"))
+        row_tokens_in = _integer(row.get("tokens_in"))
+        row_tokens_out = _integer(row.get("tokens_out"))
+        row_tokens = row_tokens_in + row_tokens_out
+        row_cost = round(_number(row.get("cost_usd")), 6)
+        models.append(
+            {
+                "provider": str(row.get("provider") or "unknown"),
+                "model": str(row.get("model") or "unknown"),
+                "turns": turns,
+                "tokens_in": row_tokens_in,
+                "tokens_out": row_tokens_out,
+                "tokens_total": row_tokens,
+                "cost_usd": row_cost,
+                "cost_share": _ratio(row_cost, total_cost),
+                "token_share": _ratio(float(row_tokens), float(total_tokens)),
+                "cost_per_turn_usd": _cost_per_turn(row_cost, turns),
+                "cost_per_1k_tokens_usd": _cost_per_1k_tokens(row_cost, row_tokens),
+            }
+        )
+    models.sort(key=lambda item: (-float(item["cost_usd"]), item["provider"], item["model"]))
+
+    warnings: list[str] = []
+    if total_turns > 0 and metered_turns < total_turns:
+        warnings.append("partial_metering")
+    if total_cost == 0 and metered_turns > 0:
+        warnings.append("zero_cost_metered_usage")
+    if str(budget.get("status") or "") in {"warn", "exceeded"}:
+        warnings.append(f"budget_{budget.get('status')}")
+    cache_hit_rate = _number(evaluator_cache.get("hit_rate"))
+    if bool(evaluator_cache.get("enabled")) and _integer(evaluator_cache.get("requests")) > 0:
+        if cache_hit_rate < 0.25:
+            warnings.append("low_evaluator_cache_hit_rate")
+    if models and models[0]["cost_share"] >= 0.8:
+        warnings.append("dominant_model_cost")
+
+    return {
+        "schema": REPORT_SCHEMA,
+        "source": str(usage.get("source") or "unknown"),
+        "durable": bool(usage.get("durable")),
+        "window_days": _integer(usage.get("window_days")),
+        "summary": {
+            "total_turns": total_turns,
+            "metered_turns": metered_turns,
+            "metered_coverage": _ratio(float(metered_turns), float(total_turns)),
+            "tokens_in": tokens_in,
+            "tokens_out": tokens_out,
+            "tokens_total": total_tokens,
+            "cost_usd": total_cost,
+            "cost_per_turn_usd": _cost_per_turn(total_cost, metered_turns),
+            "cost_per_1k_tokens_usd": _cost_per_1k_tokens(total_cost, total_tokens),
+        },
+        "budget": {
+            "status": str(budget.get("status") or "disabled"),
+            "limit_usd": round(_number(budget.get("limit_usd")), 6),
+            "used_ratio": round(_number(budget.get("used_ratio")), 6),
+            "remaining_usd": round(_number(budget.get("remaining_usd")), 6),
+        },
+        "evaluator_cache": {
+            "enabled": bool(evaluator_cache.get("enabled")),
+            "requests": _integer(evaluator_cache.get("requests")),
+            "hits": _integer(evaluator_cache.get("hits")),
+            "misses": _integer(evaluator_cache.get("misses")),
+            "hit_rate": round(cache_hit_rate, 6),
+        },
+        "models": models,
+        "top_cost_model": models[0] if models else None,
+        "warnings": warnings,
+    }
+
+
+__all__ = ["REPORT_SCHEMA", "build_model_cost_report"]
diff --git a/apps/api/app/test_admin_health_maintenance.py b/apps/api/app/test_admin_health_maintenance.py
new file mode 100644
index 0000000..f6044fa
--- /dev/null
+++ b/apps/api/app/test_admin_health_maintenance.py
@@ -0,0 +1,111 @@
+"""Admin health sample retention/rollup tests."""
+
+from __future__ import annotations
+
+import unittest
+import unittest.mock
+
+from .services import admin_health_maintenance as maintenance
+
+
+class _Acquire:
+    def __init__(self, conn):
+        self.conn = conn
+
+    async def __aenter__(self):
+        return self.conn
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+
+class AdminHealthMaintenanceTest(unittest.IsolatedAsyncioTestCase):
+    async def test_dry_run_counts_rollup_and_prunable_events_without_mutation(self) -> None:
+        class Conn:
+            def __init__(self):
+                self.fetch_calls = []
+
+            async def fetchrow(self, query, *args, **kwargs):
+                self.fetch_calls.append((query, args))
+                if len(self.fetch_calls) == 1:
+                    return {"event_count": 12, "bucket_count": 3}
+                return {"event_count": 4}
+
+            async def fetch(self, *args, **kwargs):
+                raise AssertionError("dry-run must not upsert rollups")
+
+            async def execute(self, *args, **kwargs):
+                raise AssertionError("dry-run must not delete raw events")
+
+        conn = Conn()
+        with unittest.mock.patch.object(maintenance, "acquire", return_value=_Acquire(conn)):
+            result = await maintenance.maintain_admin_health_events(
+                rollup_days=2,
+                retention_days=30,
+                apply=False,
+            )
+
+        self.assertFalse(result.applied)
+        self.assertEqual(result.rollup_event_count, 12)
+        self.assertEqual(result.rollup_bucket_count, 3)
+        self.assertEqual(result.prunable_event_count, 4)
+        self.assertEqual(result.upserted_rollups, 0)
+        self.assertEqual(result.deleted_events, 0)
+        self.assertEqual(conn.fetch_calls[0][1], (2,))
+        self.assertEqual(conn.fetch_calls[1][1], (30,))
+        self.assertIn("observed_at::date < current_date - $1::int", conn.fetch_calls[0][0])
+
+    async def test_apply_upserts_rollups_before_pruning_raw_events(self) -> None:
+        calls = []
+
+        class Conn:
+            async def fetchrow(self, query, *args, **kwargs):
+                calls.append(("fetchrow", query, args))
+                if len([call for call in calls if call[0] == "fetchrow"]) == 1:
+                    return {"event_count": 7, "bucket_count": 2}
+                return {"event_count": 5}
+
+            async def fetch(self, query, *args, **kwargs):
+                calls.append(("fetch", query, args))
+                return [1, 1]
+
+            async def execute(self, query, *args, **kwargs):
+                calls.append(("execute", query, args))
+                return "DELETE 5"
+
+        with unittest.mock.patch.object(maintenance, "acquire", return_value=_Acquire(Conn())):
+            result = await maintenance.maintain_admin_health_events(
+                rollup_days=3,
+                retention_days=30,
+                apply=True,
+            )
+
+        self.assertTrue(result.applied)
+        self.assertEqual(result.upserted_rollups, 2)
+        self.assertEqual(result.deleted_events, 5)
+        self.assertEqual([call[0] for call in calls], ["fetchrow", "fetchrow", "fetch", "execute"])
+        self.assertIn("INSERT INTO app.admin_health_daily_rollup", calls[2][1])
+        self.assertIn("last_down_at", calls[2][1])
+        self.assertIn("DELETE FROM app.admin_health_event", calls[3][1])
+
+    def test_invalid_windows_are_rejected(self) -> None:
+        with self.assertRaises(ValueError):
+            maintenance._validate_windows(rollup_days=0, retention_days=30)
+        with self.assertRaises(ValueError):
+            maintenance._validate_windows(rollup_days=7, retention_days=2)
+
+    def test_schema_defines_rollup_table_and_rls(self) -> None:
+        from pathlib import Path
+
+        root = Path(__file__).resolve().parents[3]
+        schema = (root / "infra" / "db" / "init" / "05_runtime_auth.sql").read_text(
+            encoding="utf-8"
+        )
+
+        self.assertIn("CREATE TABLE IF NOT EXISTS app.admin_health_daily_rollup", schema)
+        self.assertIn("PRIMARY KEY (rollup_date, environment, engine_mode, service_key)", schema)
+        self.assertIn("last_down_at TIMESTAMPTZ", schema)
+        self.assertIn("CREATE POLICY p_admin_health_daily_rollup_select", schema)
+        self.assertIn("CREATE POLICY p_admin_health_daily_rollup_update", schema)
+        self.assertIn("CREATE POLICY p_admin_health_event_delete", schema)
+        self.assertNotIn("CREATE POLICY p_admin_health_daily_rollup_delete", schema)
diff --git a/apps/api/app/test_admin_ops.py b/apps/api/app/test_admin_ops.py
index bb86044..209da92 100644
--- a/apps/api/app/test_admin_ops.py
+++ b/apps/api/app/test_admin_ops.py
@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import unittest
 from datetime import datetime, timedelta, timezone
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 from .deps import Principal, Role
 from .routes import admin as admin_routes
@@ -26,6 +26,9 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase):
         case = self
 
         class Conn:
+            def __init__(self) -> None:
+                self.fetch_calls = 0
+
             async def fetchrow(self, query, *args, **kwargs):
                 return {
                     "total_turns": 2,
@@ -36,6 +39,18 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase):
                 }
 
             async def fetch(self, query, *args, **kwargs):
+                self.fetch_calls += 1
+                if self.fetch_calls == 2:
+                    case.assertIn("date_trunc('day', created_at)", query)
+                    return [
+                        {
+                            "day": "2026-06-28",
+                            "turns": 1,
+                            "tokens_in": 11,
+                            "tokens_out": 13,
+                            "cost_usd": 0.0042,
+                        }
+                    ]
                 case.assertIn(
                     "COALESCE(SUM(tokens_in), 0) + COALESCE(SUM(tokens_out), 0) DESC",
                     query,
@@ -52,7 +67,11 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase):
                     }
                 ]
 
-        with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())):
+        with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())), patch.object(
+            admin_routes.evaluator,
+            "evaluator_semantic_cache_stats",
+            return_value={"hits": 3, "misses": 1, "stores": 2, "evictions": 0, "entries": 2},
+        ):
             usage = await admin_routes._usage_from_database(window_days=7)
 
         self.assertTrue(usage.durable)
@@ -60,12 +79,23 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(usage.total_turns, 2)
         self.assertEqual(usage.metered_turns, 1)
         self.assertEqual(usage.by_provider[0].provider, "claude_cli")
+        self.assertTrue(usage.evaluator_cache.enabled)
+        self.assertEqual(usage.evaluator_cache.requests, 4)
+        self.assertEqual(usage.evaluator_cache.hit_rate, 0.75)
+        self.assertEqual(usage.daily_cost[0].day, "2026-06-28")
+        self.assertEqual(usage.daily_cost[0].cost_usd, 0.0042)
 
     async def test_uptime_from_database_aggregates_health_samples(self) -> None:
         now = datetime.now(timezone.utc)
 
         class Conn:
+            def __init__(self):
+                self.calls = 0
+
             async def fetch(self, *args, **kwargs):
+                self.calls += 1
+                if self.calls == 2:
+                    return []
                 return [
                     {
                         "id": 3,
@@ -113,6 +143,113 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase):
         self.assertIsNotNone(uptime.last_down_at)
         self.assertEqual({item.service_key for item in uptime.services}, {"db", "engine", "voice"})
 
+    async def test_uptime_from_database_combines_raw_events_and_daily_rollups(self) -> None:
+        now = datetime.now(timezone.utc)
+
+        class Conn:
+            def __init__(self):
+                self.calls = 0
+
+            async def fetch(self, *args, **kwargs):
+                self.calls += 1
+                if self.calls == 1:
+                    return [
+                        {
+                            "id": 10,
+                            "observed_at": now,
+                            "overall_status": "ok",
+                            "service_key": "engine",
+                            "service_name": "응답 생성",
+                            "service_status": "ok",
+                            "detail": "ready",
+                            "metric": "21ms",
+                            "load": 0.02,
+                        }
+                    ]
+                return [
+                    {
+                        "service_key": "engine",
+                        "service_name": "응답 생성",
+                        "sample_count": 3,
+                        "ok_samples": 1,
+                        "degraded_samples": 1,
+                        "down_samples": 1,
+                        "latest_status": "down",
+                        "last_observed_at": now - timedelta(days=2),
+                        "last_down_at": now - timedelta(days=2, hours=1),
+                    }
+                ]
+
+        with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())):
+            uptime = await admin_routes._uptime_from_database(window_hours=168)
+
+        self.assertEqual(uptime.sample_count, 4)
+        self.assertEqual(uptime.down_events, 1)
+        self.assertEqual(uptime.degraded_events, 1)
+        self.assertAlmostEqual(uptime.ok_ratio, 0.5, places=4)
+        self.assertIsNotNone(uptime.last_down_at)
+        engine = next(service for service in uptime.services if service.service_key == "engine")
+        self.assertEqual(engine.samples, 4)
+        self.assertEqual(engine.latest_status, "ok")
+        self.assertEqual(len(uptime.events), 1)
+
+    async def test_record_health_events_allows_synthetic_capture_without_actor(self) -> None:
+        calls = []
+
+        class Conn:
+            async def executemany(self, query, rows, **kwargs):
+                calls.append((query, rows))
+
+        service = admin_routes.AdminServiceHealth(
+            key="engine",
+            name="응답 생성",
+            status="ok",
+            detail="ready",
+            metric="12ms",
+            load=0.01,
+        )
+        with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())):
+            recorded = await admin_routes._record_health_events(
+                principal=None,
+                overall_status="ok",
+                environment="dev",
+                engine_mode="claude_cli",
+                services=[service],
+            )
+
+        self.assertEqual(recorded, 1)
+        query, rows = calls[0]
+        self.assertIn("INSERT INTO app.admin_health_event", query)
+        self.assertIn("captured_by", query)
+        self.assertIsNone(rows[0][-1])
+
+    async def test_record_admin_health_sample_reuses_health_contract(self) -> None:
+        engine_config = admin_routes.AdminEngineConfigResponse(
+            engine_mode="claude_cli",
+            engine_url="http://127.0.0.1:9099",
+            model="gateway-default",
+            durable=True,
+            source="database",
+        )
+        recorder = AsyncMock(return_value=5)
+
+        with (
+            patch.object(admin_routes, "_current_engine_config", AsyncMock(return_value=engine_config)),
+            patch.object(admin_routes, "healthcheck", AsyncMock(return_value=True)),
+            patch.object(admin_routes.engine_client, "health_detail", AsyncMock(return_value={"ok": True})),
+            patch.object(admin_routes.voice_service, "is_available", return_value=True),
+            patch.object(admin_routes, "_runtime_health_metrics", AsyncMock(return_value=admin_routes.RuntimeHealthMetrics())),
+            patch.object(admin_routes, "_record_health_events", recorder),
+        ):
+            health, recorded = await admin_routes.record_admin_health_sample(principal=None)
+
+        self.assertEqual(recorded, 5)
+        self.assertEqual(health.status, "ok")
+        self.assertEqual(health.engine_mode, "claude_cli")
+        recorder.assert_awaited_once()
+        self.assertIsNone(recorder.await_args.kwargs["principal"])
+        self.assertEqual(len(recorder.await_args.kwargs["services"]), 5)
+
     async def test_tickets_from_database_returns_summary_without_synthetic_rows(self) -> None:
         now = datetime.now(timezone.utc)
 
@@ -289,6 +426,108 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase):
         self.assertNotIn("body", detail)
         self.assertNotIn("subject", detail)
 
+    async def test_patch_ticket_links_manual_duplicate_parent(self) -> None:
+        now = datetime.now(timezone.utc)
+        ticket_id = "00000000-0000-0000-0000-000000000101"
+        parent_id = "00000000-0000-0000-0000-000000000102"
+        audit_calls: list[tuple[str, tuple[object, ...]]] = []
+
+        class Conn:
+            def __init__(self):
+                self.fetchrow_calls = 0
+
+            async def fetchrow(self, query, *args, **kwargs):
+                self.fetchrow_calls += 1
+                if self.fetchrow_calls == 1:
+                    return {
+                        "id": ticket_id,
+                        "category": "voice_browser",
+                        "priority": "normal",
+                        "status": "open",
+                        "source_path": "/learn/session/voice",
+                        "assigned_group": "",
+                        "resolution_note": "",
+                        "parent_ticket_id": None,
+                    }
+                if self.fetchrow_calls == 2:
+                    case.assertIn("WITH RECURSIVE ancestors", query)
+                    return {"parent_exists": True, "creates_cycle": False}
+                if self.fetchrow_calls == 3:
+                    case.assertIn("parent_ticket_id = CASE", query)
+                    return {
+                        "id": ticket_id,
+                        "reporter_id": "00000000-0000-0000-0000-000000000201",
+                        "reporter_email": "learner@hs.ac.kr",
+                        "reporter_name": "Learner",
+                        "reporter_role": "learner",
+                        "category": "voice_browser",
+                        "priority": "normal",
+                        "status": "open",
+                        "subject": "마이크 권한 오류",
+                        "body": "브라우저 마이크 권한 실패",
+                        "source_path": "/learn/session/voice",
+                        "fingerprint": "dup-fingerprint",
+                        "parent_ticket_id": parent_id,
+                        "assigned_group": "",
+                        "resolution_note": "",
+                        "created_at": now,
+                        "updated_at": now,
+                        "resolved_at": None,
+                    }
+                return {
+                    "id": ticket_id,
+                    "reporter_id": "00000000-0000-0000-0000-000000000201",
+                    "reporter_email": "learner@hs.ac.kr",
+                    "reporter_name": "Learner",
+                    "reporter_role": "learner",
+                    "category": "voice_browser",
+                    "priority": "normal",
+                    "status": "open",
+                    "subject": "마이크 권한 오류",
+                    "body": "브라우저 마이크 권한 실패",
+                    "source_path": "/learn/session/voice",
+                    "fingerprint": "dup-fingerprint",
+                    "parent_ticket_id": parent_id,
+                    "duplicate_count": 1,
+                    "duplicate_parent_candidate_id": parent_id,
+                    "child_ticket_count": 0,
+                    "assigned_group": "",
+                    "resolution_note": "",
+                    "created_at": now,
+                    "updated_at": now,
+                    "resolved_at": None,
+                    "event_count": 1,
+                    "last_event_at": now,
+                }
+
+            async def execute(self, query, *args, **kwargs):
+                audit_calls.append((query, args))
+
+        case = self
+        principal = Principal(
+            user_id="00000000-0000-0000-0000-000000000301",
+            role=Role.ADMIN,
+            admin_access=True,
+            email="admin@hs.ac.kr",
+            display_name="Admin",
+        )
+        with patch.object(admin_routes, "acquire", return_value=_Acquire(Conn())):
+            updated = await admin_routes.patch_ticket(
+                ticket_id,
+                admin_routes.AdminTicketPatch(parent_ticket_id=parent_id),
+                principal,
+            )
+
+        self.assertEqual(updated.parent_ticket_id, parent_id)
+        self.assertEqual(updated.duplicate_count, 1)
+        self.assertEqual(updated.event_count, 1)
+        self.assertEqual(len(audit_calls), 1)
+        detail = audit_calls[0][1][4]
+        self.assertEqual(detail["changed_fields"], ["parent_ticket_id"])
+        self.assertEqual(detail["parent_ticket_id"]["to"], parent_id)
+        self.assertNotIn("body", detail)
+        self.assertNotIn("subject", detail)
+
     def test_admin_ops_schema_enables_rls(self) -> None:
         from pathlib import Path
 
@@ -297,5 +536,19 @@ class AdminOpsTest(unittest.IsolatedAsyncioTestCase):
             encoding="utf-8"
         )
         self.assertIn("ALTER TABLE app.admin_health_event ENABLE ROW LEVEL SECURITY", schema)
+        self.assertIn("CREATE TABLE IF NOT EXISTS app.admin_health_daily_rollup", schema)
+        self.assertIn("last_down_at TIMESTAMPTZ", schema)
         self.assertIn("ALTER TABLE app.support_ticket ENABLE ROW LEVEL SECURITY", schema)
+        self.assertIn("CREATE POLICY p_admin_health_event_delete", schema)
+        self.assertIn("CREATE POLICY p_admin_health_daily_rollup_select", schema)
+        self.assertIn("CREATE POLICY p_admin_health_daily_rollup_update", schema)
+        self.assertNotIn("CREATE POLICY p_admin_health_daily_rollup_delete", schema)
         self.assertIn("CREATE POLICY p_support_ticket_insert", schema)
+        self.assertIn("reporter_id = app.current_uid()", schema)
+        self.assertIn("fingerprint TEXT NOT NULL DEFAULT ''", schema)
+        self.assertIn("parent_ticket_id UUID REFERENCES app.support_ticket(id)", schema)
+        self.assertIn("idx_support_ticket_fingerprint", schema)
+        self.assertIn("CREATE TABLE IF NOT EXISTS app.learner_prepost_measure", schema)
+        self.assertIn("measure_name IN ('self_efficacy','skill_proficiency','training_satisfaction')", schema)
+        self.assertIn("ALTER TABLE app.learner_prepost_measure ENABLE ROW LEVEL SECURITY", schema)
+        self.assertIn("CREATE POLICY p_learner_prepost_measure_insert", schema)
diff --git a/apps/api/app/test_phase3_kpi_export.py b/apps/api/app/test_phase3_kpi_export.py
new file mode 100644
index 0000000..f60f283
--- /dev/null
+++ b/apps/api/app/test_phase3_kpi_export.py
@@ -0,0 +1,163 @@
+import json
+import tempfile
+import unittest
+from datetime import UTC, datetime
+from pathlib import Path
+
+from app.services.phase3_kpi_export import (
+    KPI_REPORT_PATH,
+    PHASE3_KPI_METRICS,
+    PREPOST_CSV_PATH,
+    ParticipantKeys,
+    build_kpi_report,
+    build_prepost_csv_rows,
+    paired_prepost_summary,
+    write_kpi_report,
+    write_prepost_csv,
+)
+
+
+REQUIRED_METRIC_KEYS = {
+    "denominator",
+    "method",
+    "numerator",
+    "pass",
+    "source_files",
+    "threshold",
+    "value",
+}
+
+
+def fixture_rows():
+    return [
+        {
+            "learner_id": "11111111-1111-1111-1111-111111111111",
+            "measure_name": "self_efficacy",
+            "timepoint": "pre",
+            "raw_score": 2,
+            "min_score": 1,
+            "max_score": 5,
+            "collected_at": datetime(2026, 6, 27, 0, 0, tzinfo=UTC),
+            "updated_at": datetime(2026, 6, 27, 0, 1, tzinfo=UTC),
+        },
+        {
+            "learner_id": "11111111-1111-1111-1111-111111111111",
+            "measure_name": "self_efficacy",
+            "timepoint": "pre",
+            "raw_score": 3,
+            "min_score": 1,
+            "max_score": 5,
+            "collected_at": datetime(2026, 6, 27, 0, 2, tzinfo=UTC),
+            "updated_at": datetime(2026, 6, 27, 0, 3, tzinfo=UTC),
+        },
+        {
+            "learner_id": "11111111-1111-1111-1111-111111111111",
+            "measure_name": "self_efficacy",
+            "timepoint": "post",
+            "raw_score": 4,
+            "min_score": 1,
+            "max_score": 5,
+            "collected_at": datetime(2026, 6, 27, 1, 0, tzinfo=UTC),
+            "updated_at": datetime(2026, 6, 27, 1, 0, tzinfo=UTC),
+        },
+        {
+            "learner_id": "22222222-2222-2222-2222-222222222222",
+            "measure_name": "self_efficacy",
+            "timepoint": "pre",
+            "raw_score": 4,
+            "min_score": 1,
+            "max_score": 5,
+            "collected_at": datetime(2026, 6, 27, 0, 5, tzinfo=UTC),
+            "updated_at": datetime(2026, 6, 27, 0, 5, tzinfo=UTC),
+        },
+        {
+            "learner_id": "11111111-1111-1111-1111-111111111111",
+            "measure_name": "skill_proficiency",
+            "timepoint": "pre",
+            "raw_score": 2,
+            "min_score": 1,
+            "max_score": 5,
+            "collected_at": datetime(2026, 6, 27, 0, 10, tzinfo=UTC),
+            "updated_at": datetime(2026, 6, 27, 0, 10, tzinfo=UTC),
+        },
+        {
+            "learner_id": "11111111-1111-1111-1111-111111111111",
+            "measure_name": "skill_proficiency",
+            "timepoint": "post",
+            "raw_score": 5,
+            "min_score": 1,
+            "max_score": 5,
+            "collected_at": datetime(2026, 6, 27, 1, 10, tzinfo=UTC),
+            "updated_at": datetime(2026, 6, 27, 1, 10, tzinfo=UTC),
+        },
+    ]
+
+
+class Phase3KpiExportTests(unittest.TestCase):
+    def test_prepost_csv_rows_use_pseudonymous_participant_ids_and_latest_score(self) -> None:
+        keys = ParticipantKeys()
+        rows = build_prepost_csv_rows(fixture_rows(), participant_keys=keys)
+
+        self.assertEqual(len(keys), 2)
+        self.assertTrue(all(row["participant_id"].startswith("P3-") for row in rows))
+        blob = json.dumps(rows, ensure_ascii=False)
+        self.assertNotIn("11111111-1111-1111-1111-111111111111", blob)
+        self.assertNotIn("22222222-2222-2222-2222-222222222222", blob)
+        p1_pre = [
+            row for row in rows
+            if row["participant_id"] == "P3-001"
+            and row["measure_name"] == "self_efficacy"
+            and row["timepoint"] == "pre"
+        ]
+        self.assertEqual([{"participant_id": "P3-001", "measure_name": "self_efficacy", "timepoint": "pre", "score": "3", "collected_at": "2026-06-27T00:02:00Z"}], p1_pre)
+
+    def test_paired_prepost_summary_uses_normalized_scores(self) -> None:
+        summary = paired_prepost_summary(fixture_rows(), "self_efficacy")
+
+        self.assertEqual(summary["participants_with_any_measure"], 2)
+        self.assertEqual(summary["complete_pairs"], 1)
+        self.assertEqual(summary["missing_pairs"], 1)
+        self.assertEqual(summary["mean_pre"], 50.0)
+        self.assertEqual(summary["mean_post"], 75.0)
+        self.assertEqual(summary["mean_delta"], 25.0)
+
+    def test_kpi_report_contains_checker_required_metric_shape(self) -> None:
+        report = build_kpi_report(
+            fixture_rows(),
+            pilot_id="phase3-pilot-draft",
+            generated_at="2026-06-28T00:00:00Z",
+            review_operator="operator",
+        )
+
+        self.assertEqual(report["pilot_id"], "phase3-pilot-draft")
+        self.assertEqual(report["cohort_size"], 2)
+        self.assertTrue(set(PHASE3_KPI_METRICS).issubset(report["metrics"]))
+        for metric in report["metrics"].values():
+            self.assertTrue(REQUIRED_METRIC_KEYS.issubset(metric))
+        self_efficacy = report["metrics"]["self_efficacy_prepost"]
+        self.assertFalse(self_efficacy["pass"])
+        self.assertEqual(self_efficacy["value"], 25.0)
+        self.assertEqual(self_efficacy["source_files"], [PREPOST_CSV_PATH])
+
+    def test_writers_create_phase3_evidence_files(self) -> None:
+        rows = build_prepost_csv_rows(fixture_rows(), participant_keys=ParticipantKeys())
+        report = build_kpi_report(
+            fixture_rows(),
+            pilot_id="phase3-pilot-draft",
+            generated_at="2026-06-28T00:00:00Z",
+        )
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            write_prepost_csv(rows, root / PREPOST_CSV_PATH)
+            write_kpi_report(report, root / KPI_REPORT_PATH)
+
+            self.assertTrue((root / PREPOST_CSV_PATH).exists())
+            self.assertTrue((root / KPI_REPORT_PATH).exists())
+            csv_text = (root / PREPOST_CSV_PATH).read_text(encoding="utf-8")
+            self.assertTrue(csv_text.startswith("participant_id,measure_name,timepoint,score,collected_at"))
+            saved = json.loads((root / KPI_REPORT_PATH).read_text(encoding="utf-8"))
+            self.assertEqual(saved["metrics"]["self_efficacy_prepost"]["complete_pairs"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/apps/api/app/test_runtime_policy.py b/apps/api/app/test_runtime_policy.py
index dfce7d6..af5df54 100644
--- a/apps/api/app/test_runtime_policy.py
+++ b/apps/api/app/test_runtime_policy.py
@@ -148,9 +148,15 @@ class RuntimeFallbackPolicyTest(unittest.IsolatedAsyncioTestCase):
                     "has_state_columns": False,
                     "has_stage_defs": False,
                     "has_admin_health_event": False,
+                    "has_admin_health_daily_rollup": False,
+                    "has_admin_health_daily_rollup_columns": False,
                     "has_support_ticket": False,
+                    "has_support_ticket_duplicate_columns": False,
+                    "has_learner_prepost_measure": False,
                     "has_admin_health_event_policies": False,
+                    "has_admin_health_daily_rollup_policies": False,
                     "has_support_ticket_policies": False,
+                    "has_learner_prepost_measure_policies": False,
                     "has_session_write_policies": False,
                     "removed_old_session_policy": False,
                     "has_turn_write_policies": False,
diff --git a/apps/api/app/test_usage_report.py b/apps/api/app/test_usage_report.py
new file mode 100644
index 0000000..fd1deb0
--- /dev/null
+++ b/apps/api/app/test_usage_report.py
@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import unittest
+
+from .services.usage_report import REPORT_SCHEMA, build_model_cost_report
+
+
+class UsageReportTest(unittest.TestCase):
+    def test_model_cost_report_derives_cost_efficiency_and_warnings(self) -> None:
+        report = build_model_cost_report(
+            {
+                "source": "database",
+                "durable": True,
+                "window_days": 7,
+                "total_turns": 5,
+                "metered_turns": 4,
+                "tokens_in": 300,
+                "tokens_out": 100,
+                "cost_usd": 0.4,
+                "budget": {
+                    "status": "warn",
+                    "limit_usd": 0.5,
+                    "used_ratio": 0.8,
+                    "remaining_usd": 0.1,
+                },
+                "evaluator_cache": {
+                    "enabled": True,
+                    "requests": 10,
+                    "hits": 2,
+                    "misses": 8,
+                    "hit_rate": 0.2,
+                },
+                "by_provider": [
+                    {
+                        "provider": "cheap",
+                        "model": "mini",
+                        "turns": 3,
+                        "tokens_in": 100,
+                        "tokens_out": 50,
+                        "cost_usd": 0.04,
+                    },
+                    {
+                        "provider": "claude_cli",
+                        "model": "opus",
+                        "turns": 1,
+                        "tokens_in": 200,
+                        "tokens_out": 50,
+                        "cost_usd": 0.36,
+                    },
+                ],
+            }
+        )
+
+        self.assertEqual(report["schema"], REPORT_SCHEMA)
+        self.assertEqual(report["summary"]["metered_coverage"], 0.8)
+        self.assertEqual(report["summary"]["cost_per_1k_tokens_usd"], 1.0)
+        self.assertEqual(report["models"][0]["provider"], "claude_cli")
+        self.assertEqual(report["models"][0]["cost_share"], 0.9)
+        self.assertEqual(report["models"][0]["cost_per_1k_tokens_usd"], 1.44)
+        self.assertEqual(report["models"][1]["cost_per_turn_usd"], 0.013333)
+        self.assertIn("partial_metering", report["warnings"])
+        self.assertIn("budget_warn", report["warnings"])
+        self.assertIn("low_evaluator_cache_hit_rate", report["warnings"])
+        self.assertIn("dominant_model_cost", report["warnings"])
+
+    def test_model_cost_report_handles_empty_usage(self) -> None:
+        report = build_model_cost_report({"budget": {}, "evaluator_cache": {}})
+
+        self.assertEqual(report["models"], [])
+        self.assertIsNone(report["top_cost_model"])
+        self.assertEqual(report["summary"]["metered_coverage"], 0.0)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/apps/api/app/test_user_support_tickets.py b/apps/api/app/test_user_support_tickets.py
new file mode 100644
index 0000000..d6ff363
--- /dev/null
+++ b/apps/api/app/test_user_support_tickets.py
@@ -0,0 +1,262 @@
+"""Current-user support ticket visibility tests."""
+
+from __future__ import annotations
+
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import AsyncMock, patch
+
+from .deps import Principal, Role
+from .routes import users as user_routes
+from .services.support_tickets import support_ticket_fingerprint
+
+
+class _Acquire:
+    def __init__(self, conn):
+        self.conn = conn
+
+    async def __aenter__(self):
+        return self.conn
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return None
+
+
+class UserSupportTicketTest(unittest.IsolatedAsyncioTestCase):
+    async def test_create_support_ticket_stores_fingerprint_without_audit_body(self) -> None:
+        ticket_id = "00000000-0000-0000-0000-000000000101"
+        audit_calls: list[tuple[str, tuple[object, ...]]] = []
+
+        class Conn:
+            async def fetchrow(self, query, *args, **kwargs):
+                self.query = query
+                self.args = args
+                return {
+                    "id": ticket_id,
+                    "category": args[4],
+                    "priority": args[5],
+                    "subject": args[6],
+                    "created_at": 1234.0,
+                }
+
+            async def execute(self, query, *args, **kwargs):
+                audit_calls.append((query, args))
+
+        principal = Principal(
+            user_id="00000000-0000-0000-0000-000000000201",
+            role=Role.LEARNER,
+            email="learner@hs.ac.kr",
+            display_name="Learner",
+        )
+        body = user_routes.UserSupportTicketRequest(
+            category="voice_browser",
+            priority="high",
+            subject="마이크 권한 오류",
+            body="브라우저에서 마이크 권한을 눌렀는데 다시 실패합니다.",
+            source_path="/learn/session/voice",
+        )
+        expected_fingerprint = support_ticket_fingerprint(
+            category=body.category,
+            subject=body.subject,
+            body=body.body,
+            source_path=body.source_path,
+        )
+        profile = user_routes.UserProfileResponse(
+            user_id=principal.user_id,
+            email=principal.email,
+            display_name="Learner",
+            role="learner",
+            cohort_ids=[],
+            affiliation="한신대학교",
+        )
+        conn = Conn()
+
+        with (
+            patch.object(user_routes, "acquire", return_value=_Acquire(conn)),
+            patch.object(user_routes, "_profile_for", AsyncMock(return_value=profile)),
+        ):
+            created = await user_routes.create_support_ticket(body, principal)
+
+        self.assertEqual(created.ticket_id, ticket_id)
+        self.assertIn("fingerprint", conn.query)
+        self.assertEqual(conn.args[9], expected_fingerprint)
+        self.assertEqual(len(audit_calls), 1)
+        detail = audit_calls[0][1][4]
+        self.assertEqual(detail["fingerprint"], expected_fingerprint)
+        self.assertNotIn("body", detail)
+        self.assertNotIn("subject", detail)
+
+    async def test_user_ticket_list_scopes_to_current_reporter_without_body(self) -> None:
+        now = datetime.now(timezone.utc)
+
+        class Conn:
+            async def fetch(self, query, *args, **kwargs):
+                self.query = query
+                self.args = args
+                return [
+                    {
+                        "id": "00000000-0000-0000-0000-000000000101",
+                        "category": "session_review",
+                        "priority": "high",
+                        "status": "resolved",
+                        "subject": "리뷰 생성 지연",
+                        "source_path": "/learn/session/1/review",
+                        "assigned_group": "서비스 운영자",
+                        "resolution_note": "재처리 완료",
+                        "created_at": (now - timedelta(days=1)).timestamp(),
+                        "updated_at": now.timestamp(),
+                        "resolved_at": now.timestamp(),
+                    }
+                ]
+
+        conn = Conn()
+        principal = Principal(
+            user_id="00000000-0000-0000-0000-000000000201",
+            role=Role.LEARNER,
+            email="learner@hs.ac.kr",
+            display_name="Learner",
+        )
+        with patch.object(user_routes, "acquire", return_value=_Acquire(conn)):
+            tickets = await user_routes._support_tickets_for_user(principal)
+
+        self.assertIn("WHERE reporter_id = $1::uuid", conn.query)
+        self.assertNotIn(" body", conn.query.lower())
+        self.assertEqual(conn.args, (principal.user_id, 20))
+        self.assertEqual(tickets.source, "database")
+        self.assertTrue(tickets.durable)
+        self.assertEqual(len(tickets.tickets), 1)
+        ticket = tickets.tickets[0]
+        self.assertEqual(ticket.status, "resolved")
+        self.assertEqual(ticket.resolution_note, "재처리 완료")
+        self.assertFalse(hasattr(ticket, "body"))
+
+
+class UserPrepostMeasureTest(unittest.IsolatedAsyncioTestCase):
+    async def test_upsert_prepost_measure_scopes_to_current_user_without_score_in_audit(self) -> None:
+        measure_id = "00000000-0000-0000-0000-000000000301"
+        audit_calls: list[tuple[str, tuple[object, ...]]] = []
+
+        class Conn:
+            async def fetchrow(self, query, *args, **kwargs):
+                self.query = query
+                self.args = args
+                return {
+                    "id": measure_id,
+                    "pilot_id": args[1],
+                    "measure_name": args[2],
+                    "timepoint": args[3],
+                    "raw_score": args[4],
+                    "min_score": args[5],
+                    "max_score": args[6],
+                    "instrument_version": args[7],
+                    "item_count": args[8],
+                    "collected_at": 1234.0,
+                    "updated_at": 1235.0,
+                }
+
+            async def execute(self, query, *args, **kwargs):
+                audit_calls.append((query, args))
+
+        principal = Principal(
+            user_id="00000000-0000-0000-0000-000000000201",
+            role=Role.LEARNER,
+            email="learner@hs.ac.kr",
+            display_name="Learner",
+        )
+        body = user_routes.UserPrepostMeasureRequest(
+            pilot_id="phase3-pilot-local",
+            measure_name="self_efficacy",
+            timepoint="pre",
+            raw_score=4.0,
+            min_score=1.0,
+            max_score=5.0,
+            instrument_version="draft-scale-v1",
+            item_count=6,
+        )
+        conn = Conn()
+
+        with patch.object(user_routes, "acquire", return_value=_Acquire(conn)):
+            saved = await user_routes.upsert_my_prepost_measure(body, principal)
+
+        self.assertIn("learner_prepost_measure", conn.query)
+        self.assertEqual(conn.args[0], principal.user_id)
+        self.assertEqual(conn.args[1], "phase3-pilot-local")
+        self.assertEqual(conn.args[2], "self_efficacy")
+        self.assertEqual(conn.args[3], "pre")
+        self.assertEqual(saved.normalized_score, 75.0)
+        self.assertEqual(len(audit_calls), 1)
+        detail = audit_calls[0][1][4]
+        self.assertEqual(detail["measure_name"], "self_efficacy")
+        self.assertEqual(detail["timepoint"], "pre")
+        self.assertTrue(detail["score_recorded"])
+        self.assertNotIn("raw_score", detail)
+        self.assertNotIn("score", detail)
+
+    async def test_prepost_measure_list_scopes_to_current_user_and_counts_pairs(self) -> None:
+        class Conn:
+            async def fetch(self, query, *args, **kwargs):
+                self.query = query
+                self.args = args
+                return [
+                    {
+                        "id": "00000000-0000-0000-0000-000000000301",
+                        "pilot_id": "phase3-pilot-local",
+                        "measure_name": "self_efficacy",
+                        "timepoint": "pre",
+                        "raw_score": 3.0,
+                        "min_score": 1.0,
+                        "max_score": 5.0,
+                        "instrument_version": "draft-scale-v1",
+                        "item_count": 6,
+                        "collected_at": 1234.0,
+                        "updated_at": 1234.0,
+                    },
+                    {
+                        "id": "00000000-0000-0000-0000-000000000302",
+                        "pilot_id": "phase3-pilot-local",
+                        "measure_name": "self_efficacy",
+                        "timepoint": "post",
+                        "raw_score": 4.0,
+                        "min_score": 1.0,
+                        "max_score": 5.0,
+                        "instrument_version": "draft-scale-v1",
+                        "item_count": 6,
+                        "collected_at": 2234.0,
+                        "updated_at": 2234.0,
+                    },
+                    {
+                        "id": "00000000-0000-0000-0000-000000000303",
+                        "pilot_id": "phase3-pilot-local",
+                        "measure_name": "skill_proficiency",
+                        "timepoint": "pre",
+                        "raw_score": 2.0,
+                        "min_score": 1.0,
+                        "max_score": 5.0,
+                        "instrument_version": "draft-scale-v1",
+                        "item_count": 6,
+                        "collected_at": 3234.0,
+                        "updated_at": 3234.0,
+                    },
+                ]
+
+        principal = Principal(
+            user_id="00000000-0000-0000-0000-000000000201",
+            role=Role.LEARNER,
+            email="learner@hs.ac.kr",
+            display_name="Learner",
+        )
+        conn = Conn()
+
+        with patch.object(user_routes, "acquire", return_value=_Acquire(conn)):
+            response = await user_routes._prepost_measures_for_user(
+                principal,
+                pilot_id="phase3-pilot-local",
+            )
+
+        self.assertIn("WHERE learner_id = $1::uuid", conn.query)
+        self.assertEqual(conn.args, (principal.user_id, "phase3-pilot-local"))
+        self.assertEqual(response.source, "database")
+        self.assertTrue(response.durable)
+        self.assertEqual(response.complete_measure_pairs, 1)
+        self.assertEqual(len(response.measures), 3)
+        self.assertEqual(response.measures[0].normalized_score, 50.0)
diff --git a/infra/db/init/05_runtime_auth.sql b/infra/db/init/05_runtime_auth.sql
index 2672c43..e6defe7 100644
--- a/infra/db/init/05_runtime_auth.sql
+++ b/infra/db/init/05_runtime_auth.sql
@@ -92,6 +92,28 @@ CREATE TABLE IF NOT EXISTS app.admin_health_event (
 CREATE INDEX IF NOT EXISTS idx_admin_health_event_observed
   ON app.admin_health_event(observed_at DESC, service_key);
 
+CREATE TABLE IF NOT EXISTS app.admin_health_daily_rollup (
+  rollup_date DATE NOT NULL,
+  environment TEXT NOT NULL,
+  engine_mode TEXT NOT NULL,
+  service_key TEXT NOT NULL,
+  service_name TEXT NOT NULL,
+  sample_count INTEGER NOT NULL DEFAULT 0 CHECK (sample_count >= 0),
+  ok_samples INTEGER NOT NULL DEFAULT 0 CHECK (ok_samples >= 0),
+  degraded_samples INTEGER NOT NULL DEFAULT 0 CHECK (degraded_samples >= 0),
+  down_samples INTEGER NOT NULL DEFAULT 0 CHECK (down_samples >= 0),
+  first_observed_at TIMESTAMPTZ NOT NULL,
+  last_observed_at TIMESTAMPTZ NOT NULL,
+  latest_status TEXT NOT NULL CHECK (latest_status IN ('ok','degraded','down')),
+  last_down_at TIMESTAMPTZ,
+  max_load REAL NOT NULL DEFAULT 0.0 CHECK (max_load >= 0.0 AND max_load <= 1.0),
+  generated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  PRIMARY KEY (rollup_date, environment, engine_mode, service_key)
+);
+
+CREATE INDEX IF NOT EXISTS idx_admin_health_daily_rollup_latest
+  ON app.admin_health_daily_rollup(last_observed_at DESC, service_key);
+
 CREATE TABLE IF NOT EXISTS app.support_ticket (
   id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
   reporter_id UUID REFERENCES app.app_user(user_id) ON DELETE SET NULL,
@@ -115,6 +137,8 @@ CREATE TABLE IF NOT EXISTS app.support_ticket (
   subject TEXT NOT NULL,
   body TEXT NOT NULL,
   source_path TEXT NOT NULL DEFAULT '',
+  fingerprint TEXT NOT NULL DEFAULT '',
+  parent_ticket_id UUID REFERENCES app.support_ticket(id) ON DELETE SET NULL,
   assigned_group TEXT NOT NULL DEFAULT '',
   resolution_note TEXT NOT NULL DEFAULT '',
   created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
@@ -129,15 +153,68 @@ CREATE INDEX IF NOT EXISTS idx_support_ticket_status_priority
 CREATE INDEX IF NOT EXISTS idx_support_ticket_reporter
   ON app.support_ticket(reporter_id, created_at DESC);
 
+CREATE INDEX IF NOT EXISTS idx_support_ticket_fingerprint
+  ON app.support_ticket(fingerprint, created_at DESC)
+  WHERE fingerprint <> '';
+
+CREATE INDEX IF NOT EXISTS idx_support_ticket_parent
+  ON app.support_ticket(parent_ticket_id)
+  WHERE parent_ticket_id IS NOT NULL;
+
+CREATE TABLE IF NOT EXISTS app.learner_prepost_measure (
+  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+  learner_id UUID NOT NULL REFERENCES app.app_user(user_id) ON DELETE CASCADE,
+  pilot_id TEXT NOT NULL DEFAULT 'phase3-pilot-draft',
+  measure_name TEXT NOT NULL CHECK (
+    measure_name IN ('self_efficacy','skill_proficiency','training_satisfaction')
+  ),
+  timepoint TEXT NOT NULL CHECK (timepoint IN ('pre','post')),
+  raw_score NUMERIC(8,3) NOT NULL,
+  min_score NUMERIC(8,3) NOT NULL DEFAULT 1.0,
+  max_score NUMERIC(8,3) NOT NULL DEFAULT 5.0,
+  instrument_version TEXT NOT NULL DEFAULT 'pilot-prepost-scaffold-2026-06-28',
+  item_count INTEGER NOT NULL DEFAULT 1 CHECK (item_count >= 1 AND item_count <= 80),
+  collected_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+  CHECK (max_score > min_score),
+  CHECK (raw_score >= min_score AND raw_score <= max_score),
+  UNIQUE (learner_id, pilot_id, measure_name, timepoint, instrument_version)
+);
+
+CREATE INDEX IF NOT EXISTS idx_learner_prepost_measure_pilot
+  ON app.learner_prepost_measure(pilot_id, measure_name, timepoint);
+
+CREATE INDEX IF NOT EXISTS idx_learner_prepost_measure_learner
+  ON app.learner_prepost_measure(learner_id, pilot_id);
+
 ALTER TABLE app.admin_health_event ENABLE ROW LEVEL SECURITY;
 DROP POLICY IF EXISTS p_admin_health_event_select ON app.admin_health_event;
 DROP POLICY IF EXISTS p_admin_health_event_insert ON app.admin_health_event;
+DROP POLICY IF EXISTS p_admin_health_event_delete ON app.admin_health_event;
 CREATE POLICY p_admin_health_event_select
   ON app.admin_health_event FOR SELECT
   USING (app.current_role_name() = 'admin');
 CREATE POLICY p_admin_health_event_insert
   ON app.admin_health_event FOR INSERT
   WITH CHECK (app.current_role_name() = 'admin');
+CREATE POLICY p_admin_health_event_delete
+  ON app.admin_health_event FOR DELETE
+  USING (app.current_role_name() = 'admin');
+
+ALTER TABLE app.admin_health_daily_rollup ENABLE ROW LEVEL SECURITY;
+DROP POLICY IF EXISTS p_admin_health_daily_rollup_select ON app.admin_health_daily_rollup;
+DROP POLICY IF EXISTS p_admin_health_daily_rollup_insert ON app.admin_health_daily_rollup;
+DROP POLICY IF EXISTS p_admin_health_daily_rollup_update ON app.admin_health_daily_rollup;
+CREATE POLICY p_admin_health_daily_rollup_select
+  ON app.admin_health_daily_rollup FOR SELECT
+  USING (app.current_role_name() = 'admin');
+CREATE POLICY p_admin_health_daily_rollup_insert
+  ON app.admin_health_daily_rollup FOR INSERT
+  WITH CHECK (app.current_role_name() = 'admin');
+CREATE POLICY p_admin_health_daily_rollup_update
+  ON app.admin_health_daily_rollup FOR UPDATE
+  USING (app.current_role_name() = 'admin')
+  WITH CHECK (app.current_role_name() = 'admin');
 
 ALTER TABLE app.support_ticket ENABLE ROW LEVEL SECURITY;
 DROP POLICY IF EXISTS p_support_ticket_select ON app.support_ticket;
@@ -164,6 +241,41 @@ CREATE POLICY p_support_ticket_delete
   ON app.support_ticket FOR DELETE
   USING (app.current_role_name() = 'admin');
 
+ALTER TABLE app.learner_prepost_measure ENABLE ROW LEVEL SECURITY;
+DROP POLICY IF EXISTS p_learner_prepost_measure_select ON app.learner_prepost_measure;
+DROP POLICY IF EXISTS p_learner_prepost_measure_insert ON app.learner_prepost_measure;
+DROP POLICY IF EXISTS p_learner_prepost_measure_update ON app.learner_prepost_measure;
+CREATE POLICY p_learner_prepost_measure_select
+  ON app.learner_prepost_measure FOR SELECT
+  USING (
+    app.current_role_name() = 'admin'
+    OR learner_id = app.current_uid()
+    OR (
+      app.current_role_name() = 'instructor'
+      AND EXISTS (
+        SELECT 1 FROM app.app_user u
+        WHERE u.user_id = app.learner_prepost_measure.learner_id
+          AND u.cohort = current_setting('app.current_cohort', true)
+      )
+    )
+  );
+CREATE POLICY p_learner_prepost_measure_insert
+  ON app.learner_prepost_measure FOR INSERT
+  WITH CHECK (
+    app.current_role_name() = 'admin'
+    OR learner_id = app.current_uid()
+  );
+CREATE POLICY p_learner_prepost_measure_update
+  ON app.learner_prepost_measure FOR UPDATE
+  USING (
+    app.current_role_name() = 'admin'
+    OR learner_id = app.current_uid()
+  )
+  WITH CHECK (
+    app.current_role_name() = 'admin'
+    OR learner_id = app.current_uid()
+  );
+
 ALTER TABLE app.sessions
   ADD COLUMN IF NOT EXISTS runtime_case_id UUID,
   ADD COLUMN IF NOT EXISTS persona_code TEXT,
diff --git a/scripts/export-phase3-kpi.py b/scripts/export-phase3-kpi.py
new file mode 100644
index 0000000..dcff1fa
--- /dev/null
+++ b/scripts/export-phase3-kpi.py
@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+API_ROOT = REPO_ROOT / "apps" / "api"
+sys.path.insert(0, str(API_ROOT))
+
+import asyncpg  # noqa: E402
+
+from app.services.phase3_kpi_export import (  # noqa: E402
+    KPI_REPORT_PATH,
+    PREPOST_CSV_PATH,
+    ParticipantKeys,
+    build_kpi_report,
+    build_prepost_csv_rows,
+    write_kpi_report,
+    write_prepost_csv,
+)
+
+
+PREPOST_QUERY = """
+WITH ranked AS (
+  SELECT
+    learner_id::text AS learner_id,
+    pilot_id,
+    measure_name,
+    timepoint,
+    raw_score,
+    min_score,
+    max_score,
+    instrument_version,
+    item_count,
+    collected_at,
+    updated_at,
+    row_number() OVER (
+      PARTITION BY learner_id, measure_name, timepoint
+      ORDER BY updated_at DESC, collected_at DESC, instrument_version DESC
+    ) AS rn
+  FROM app.learner_prepost_measure
+  WHERE pilot_id = $1
+    AND ($2::timestamptz IS NULL OR collected_at >= $2::timestamptz)
+    AND ($3::timestamptz IS NULL OR collected_at < $3::timestamptz)
+)
+SELECT
+  learner_id,
+  pilot_id,
+  measure_name,
+  timepoint,
+  raw_score,
+  min_score,
+  max_score,
+  instrument_version,
+  item_count,
+  collected_at,
+  updated_at
+FROM ranked
+WHERE rn = 1
+ORDER BY learner_id, measure_name, timepoint
+"""
+
+
+def parse_time(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=UTC)
+    return parsed
+
+
+def now_utc() -> str:
+    return datetime.now(UTC).isoformat().replace("+00:00", "Z")
+
+
+async def fetch_rows(args: argparse.Namespace) -> list[dict[str, Any]]:
+    conn = await asyncpg.connect(args.database_url)
+    try:
+        records = await conn.fetch(
+            PREPOST_QUERY,
+            args.pilot_id,
+            parse_time(args.started_at),
+            parse_time(args.ended_at),
+        )
+    finally:
+        await conn.close()
+    return [dict(record) for record in records]
+
+
+async def run(args: argparse.Namespace) -> int:
+    rows = await fetch_rows(args)
+    keys = ParticipantKeys()
+    csv_rows = build_prepost_csv_rows(rows, participant_keys=keys)
+    generated_at = args.generated_at or now_utc()
+    source_window = None
+    if args.started_at or args.ended_at:
+        source_window = {
+            "started_at": args.started_at or "",
+            "ended_at": args.ended_at or "",
+        }
+    report = build_kpi_report(
+        rows,
+        pilot_id=args.pilot_id,
+        generated_at=generated_at,
+        source_window=source_window,
+        review_operator=args.operator,
+    )
+    output_root = Path(args.output_root)
+    prepost_path = output_root / PREPOST_CSV_PATH
+    report_path = output_root / KPI_REPORT_PATH
+    write_prepost_csv(csv_rows, prepost_path)
+    write_kpi_report(report, report_path)
+    print(
+        json.dumps(
+            {
+                "prepost_csv": str(prepost_path),
+                "kpi_report": str(report_path),
+                "rows": len(csv_rows),
+                "participants": len(keys),
+                "pilot_id": args.pilot_id,
+            },
+            ensure_ascii=False,
+        )
+    )
+    return 0
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Export Phase 3 pre/post KPI evidence CSV and kpi_report.json from app.learner_prepost_measure."
+    )
+    parser.add_argument("--database-url", default=os.environ.get("DATABASE_URL"), required=os.environ.get("DATABASE_URL") is None)
+    parser.add_argument("--output-root", default=str(REPO_ROOT / "data" / "phase3-dry-run"))
+    parser.add_argument("--pilot-id", default="phase3-pilot-draft")
+    parser.add_argument("--started-at")
+    parser.add_argument("--ended-at")
+    parser.add_argument("--generated-at")
+    parser.add_argument("--operator", default="")
+    return parser
+
+
+def main() -> int:
+    parser = build_parser()
+    args = parser.parse_args()
+    return asyncio.run(run(args))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/install-health-sampler-task.ps1 b/scripts/install-health-sampler-task.ps1
new file mode 100644
index 0000000..cc6f7cc
--- /dev/null
+++ b/scripts/install-health-sampler-task.ps1
@@ -0,0 +1,85 @@
+param(
+  [string]$Workspace = "D:\workspace\vignette",
+  [string]$TaskName = "VignetteAdminHealthSampler",
+  [int]$IntervalMinutes = 5,
+  [string]$Python = "$env:LOCALAPPDATA\Programs\Python\Python311\python.exe",
+  [switch]$RunNow,
+  [switch]$PrintOnly
+)
+
+$ErrorActionPreference = "Stop"
+
+if ($IntervalMinutes -lt 1) {
+  throw "IntervalMinutes must be 1 or greater"
+}
+
+$sampleScript = Join-Path $Workspace "scripts\record-admin-health-sample.py"
+if (!(Test-Path $sampleScript)) {
+  throw "Health sampler script not found at $sampleScript"
+}
+
+if (!(Test-Path $Python)) {
+  $pythonCommand = Get-Command python.exe -ErrorAction Stop
+  $Python = $pythonCommand.Source
+}
+
+$userId = [System.Security.Principal.WindowsIdentity]::GetCurrent().Name
+$actionArguments = @(
+  "-B",
+  "`"$sampleScript`"",
+  "--json"
+)
+
+if ($PrintOnly) {
+  Write-Output "Task: $TaskName"
+  Write-Output "User: $userId"
+  Write-Output "Action: $Python $($actionArguments -join ' ')"
+  Write-Output "WorkingDirectory: $Workspace"
+  Write-Output "Interval: every $IntervalMinutes minute(s), plus at user logon"
+  return
+}
+
+$action = New-ScheduledTaskAction `
+  -Execute $Python `
+  -Argument ($actionArguments -join " ") `
+  -WorkingDirectory $Workspace
+
+$logonTrigger = New-ScheduledTaskTrigger -AtLogOn -User $userId
+$repeatTrigger = New-ScheduledTaskTrigger `
+  -Once `
+  -At (Get-Date).AddMinutes(1) `
+  -RepetitionInterval (New-TimeSpan -Minutes $IntervalMinutes)
+
+$settings = New-ScheduledTaskSettingsSet `
+  -AllowStartIfOnBatteries `
+  -DontStopIfGoingOnBatteries `
+  -ExecutionTimeLimit (New-TimeSpan -Minutes 3) `
+  -MultipleInstances IgnoreNew `
+  -RestartCount 2 `
+  -RestartInterval (New-TimeSpan -Minutes 1) `
+  -StartWhenAvailable `
+  -WakeToRun
+
+$principal = New-ScheduledTaskPrincipal `
+  -UserId $userId `
+  -LogonType Interactive `
+  -RunLevel Limited
+
+$description = "Records Vignette admin health samples into app.admin_health_event as a one-shot synthetic monitor. Secrets stay in apps/api/.env or the user environment; the task command stores no secrets."
+$task = New-ScheduledTask `
+  -Action $action `
+  -Trigger @($logonTrigger, $repeatTrigger) `
+  -Settings $settings `
+  -Principal $principal `
+  -Description $description
+
+Register-ScheduledTask -TaskName $TaskName -InputObject $task -Force | Out-Null
+
+Write-Output "Installed scheduled task '$TaskName' for $userId"
+Write-Output "Action: $Python $($actionArguments -join ' ')"
+Write-Output "Interval: every $IntervalMinutes minute(s), plus at user logon"
+
+if ($RunNow) {
+  Start-ScheduledTask -TaskName $TaskName
+  Write-Output "Started scheduled task '$TaskName'"
+}
diff --git a/scripts/maintain-admin-health-events.py b/scripts/maintain-admin-health-events.py
new file mode 100644
index 0000000..0c0ae13
--- /dev/null
+++ b/scripts/maintain-admin-health-events.py
@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+"""Roll up and optionally prune persisted admin health samples."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+from dataclasses import asdict
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+API_ROOT = REPO_ROOT / "apps" / "api"
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+if hasattr(sys.stderr, "reconfigure"):
+    sys.stderr.reconfigure(encoding="utf-8")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Roll up app.admin_health_event into app.admin_health_daily_rollup "
+            "and optionally prune old raw samples."
+        )
+    )
+    parser.add_argument("--rollup-days", type=int, required=True, help="Roll up raw samples older than this many days.")
+    parser.add_argument("--retention-days", type=int, required=True, help="Prune raw samples older than this many days.")
+    parser.add_argument("--apply", action="store_true", help="Mutate the database. Default is dry-run.")
+    parser.add_argument(
+        "--allow-non-dev-apply",
+        action="store_true",
+        help="Required with --apply outside ENVIRONMENT=dev.",
+    )
+    parser.add_argument("--json", action="store_true", help="Print compact JSON.")
+    return parser
+
+
+def _prepare_imports() -> None:
+    os.chdir(API_ROOT)
+    api_root = str(API_ROOT)
+    if api_root not in sys.path:
+        sys.path.insert(0, api_root)
+
+
+async def _run(args: argparse.Namespace) -> dict[str, object]:
+    _prepare_imports()
+    from app.auth_sessions import ensure_runtime_tables
+    from app.config import settings
+    from app.db import close_pool, init_pool
+    from app.services.admin_health_maintenance import maintain_admin_health_events
+
+    if args.apply and settings.environment != "dev" and not args.allow_non_dev_apply:
+        raise RuntimeError("--apply outside ENVIRONMENT=dev requires --allow-non-dev-apply")
+
+    await init_pool()
+    try:
+        await ensure_runtime_tables()
+        result = await maintain_admin_health_events(
+            rollup_days=args.rollup_days,
+            retention_days=args.retention_days,
+            apply=args.apply,
+        )
+    finally:
+        await close_pool()
+    return asdict(result)
+
+
+def _print_result(result: dict[str, object], *, as_json: bool) -> None:
+    if as_json:
+        print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))
+        return
+    mode = "applied" if result["applied"] else "dry-run"
+    print(
+        "{mode}: rollup_events={rollup_event_count}, rollup_buckets={rollup_bucket_count}, "
+        "upserted_rollups={upserted_rollups}, prunable_events={prunable_event_count}, "
+        "deleted_events={deleted_events}".format(mode=mode, **result)
+    )
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    try:
+        result = asyncio.run(_run(args))
+    except Exception as exc:  # noqa: BLE001 - CLI should report compact failure.
+        payload = {"ok": False, "error_type": exc.__class__.__name__, "error": str(exc)}
+        print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")), file=sys.stderr)
+        return 2
+    _print_result(result, as_json=args.json)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/record-admin-health-sample.py b/scripts/record-admin-health-sample.py
new file mode 100644
index 0000000..e781b7e
--- /dev/null
+++ b/scripts/record-admin-health-sample.py
@@ -0,0 +1,139 @@
+#!/usr/bin/env python3
+"""Record one Vignette admin health sample into app.admin_health_event.
+
+This is a one-shot synthetic monitor entrypoint.  It reuses the same health
+calculation as GET /admin/health, but does not require an interactive admin
+browser session.  Schedule it with scripts/install-health-sampler-task.ps1 on
+Windows, or run it manually for a local proof.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+API_ROOT = REPO_ROOT / "apps" / "api"
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+if hasattr(sys.stderr, "reconfigure"):
+    sys.stderr.reconfigure(encoding="utf-8")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Collect one live Vignette health sample and persist it to app.admin_health_event."
+    )
+    parser.add_argument(
+        "--json",
+        action="store_true",
+        help="Print a compact JSON result instead of a short text summary.",
+    )
+    parser.add_argument(
+        "--allow-unrecorded",
+        action="store_true",
+        help="Exit 0 even when the health check ran but no DB event rows were recorded.",
+    )
+    return parser
+
+
+def _prepare_imports() -> None:
+    os.chdir(API_ROOT)
+    api_root = str(API_ROOT)
+    if api_root not in sys.path:
+        sys.path.insert(0, api_root)
+
+
+def _service_payload(service: Any) -> dict[str, Any]:
+    return {
+        "key": service.key,
+        "status": service.status,
+        "metric": service.metric,
+        "load": service.load,
+    }
+
+
+async def _collect() -> dict[str, Any]:
+    _prepare_imports()
+    from app.auth_sessions import ensure_runtime_tables
+    from app.db import close_pool, init_pool
+    from app.engine_client import engine_client
+    from app.routes.admin import apply_engine_config_from_store, record_admin_health_sample
+    from app.services.voice import voice_service
+
+    await init_pool()
+    try:
+        await ensure_runtime_tables()
+        await apply_engine_config_from_store()
+        await engine_client.startup()
+        await voice_service.startup()
+        health, recorded_count = await record_admin_health_sample(principal=None)
+    finally:
+        await voice_service.shutdown()
+        await engine_client.shutdown()
+        await close_pool()
+    return {
+        "ok": recorded_count > 0,
+        "recorded_count": recorded_count,
+        "generated_at": time.time(),
+        "status": health.status,
+        "environment": health.environment,
+        "engine_mode": health.engine_mode,
+        "services": [_service_payload(service) for service in health.services],
+    }
+
+
+def _error_payload(exc: BaseException) -> dict[str, Any]:
+    return {
+        "ok": False,
+        "recorded_count": 0,
+        "generated_at": time.time(),
+        "error_type": exc.__class__.__name__,
+        "error": str(exc),
+    }
+
+
+def _print_result(result: dict[str, Any], *, as_json: bool) -> None:
+    if as_json:
+        print(json.dumps(result, ensure_ascii=False, separators=(",", ":")))
+        return
+    if result.get("ok"):
+        print(
+            "recorded {count} health events: status={status}, env={env}, engine={engine}".format(
+                count=result["recorded_count"],
+                status=result["status"],
+                env=result["environment"],
+                engine=result["engine_mode"],
+            )
+        )
+    else:
+        print(
+            "health sample not recorded: {error}".format(
+                error=result.get("error") or f"recorded_count={result.get('recorded_count', 0)}"
+            ),
+            file=sys.stderr,
+        )
+
+
+def main() -> int:
+    args = build_parser().parse_args()
+    try:
+        result = asyncio.run(_collect())
+    except Exception as exc:  # noqa: BLE001 - CLI should report a compact failure.
+        result = _error_payload(exc)
+    _print_result(result, as_json=args.json)
+    if result.get("ok") or args.allow_unrecorded:
+        return 0
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/scripts/report-ai-usage.py b/scripts/report-ai-usage.py
new file mode 100644
index 0000000..b5fae9e
--- /dev/null
+++ b/scripts/report-ai-usage.py
@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+"""Build a deterministic model-cost report from an admin usage JSON export."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import sys
+
+
+def _repo_root() -> Path:
+    return Path(__file__).resolve().parents[1]
+
+
+def _load_builder():
+    api_root = _repo_root() / "apps" / "api"
+    sys.path.insert(0, str(api_root))
+    from app.services.usage_report import build_model_cost_report
+
+    return build_model_cost_report
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Create a model/provider AI usage cost report from AdminUsageResponse JSON.",
+    )
+    parser.add_argument("--input", required=True, help="Path to AdminUsageResponse JSON.")
+    parser.add_argument("--output", help="Optional output JSON path. Defaults to stdout.")
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    usage = json.loads(Path(args.input).read_text(encoding="utf-8"))
+    report = _load_builder()(usage)
+    payload = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+    if args.output:
+        Path(args.output).write_text(payload, encoding="utf-8")
+    else:
+        sys.stdout.write(payload)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())