Praxis CI
ec397f2c65
v0.5 (Live Assist — on-the-job voice companion) milestone complete. 4 phases: P0 (pre-execution, v0.1.10) → P1 (assist core + guardrail, v0.1.11) → P2 (integration + tech-debt + NFR, v0.1.12) → P3 (final review + ship, v0.1.13 = milestone release). 16/16 REQs covered (3 ASSIST + 4 NFR + 9 IDEATE). 4 v0.6 backlog. 469 tests passed, 0 failed. 1 P0 fixed (guardrail processor safety). 8 P1+ flagged for v0.6. 8 v0.4 P1+ tech-debt addressed. G-049 + G-067 grill MUSTs resolved. ESCALATION-01 (PIPEDA) OPEN for human legal review before assist surface go-live. ---ci--- project: praxis phase: 3 milestone: v0.5 status: complete requirements: covered: [REQ-ASSIST-01, REQ-ASSIST-02, REQ-ASSIST-03, REQ-NFR-ASSIST-01, REQ-NFR-ASSIST-02, REQ-NFR-ASSIST-03, REQ-NFR-ASSIST-04, REQ-IDEATE-01, REQ-IDEATE-02, REQ-IDEATE-03, REQ-IDEATE-04, REQ-IDEATE-05, REQ-IDEATE-06, REQ-IDEATE-07, REQ-IDEATE-08, REQ-IDEATE-09] partial: [] ---/ci---
290 lines
11 KiB
Python
290 lines
11 KiB
Python
"""NFR measurement tests (TASK-09-03, REQ-NFR-ASSIST-01, REQ-IDEATE-04, D-072).
|
|
|
|
Tests the measurement infrastructure (NOT the actual latency — that's a Phase-1
|
|
live measurement, not a CI test):
|
|
- AssistLatencyMetrics: p95/p50/p99 computed correctly from mock records.
|
|
D-072: within_target = (p95 < 600), within_pilot = (p95 <= 650).
|
|
- GuardrailMetrics: false_positive_rate on the tuning corpus, false_negative_rate
|
|
on the direct-answer corpus, nightly_trend on mock turns.
|
|
|
|
D-072 binding: the pilot tolerance is ≤ 650ms. The target is < 600ms (C-8). The
|
|
test ASSERTS that the measurement infrastructure works (percentiles + flags),
|
|
not that the actual latency is under budget.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import datetime as _dt
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from server.assist.guardrail_metrics import GuardrailMetrics
|
|
from server.assist.latency_metrics import (
|
|
PILOT_TOLERANCE_MS,
|
|
TARGET_MS,
|
|
AssistLatencyMetrics,
|
|
)
|
|
from server.latency import LatencyRecord
|
|
|
|
|
|
# ── AssistLatencyMetrics ────────────────────────────────────────────────────
|
|
|
|
|
|
def _record(e2e_ms: float) -> LatencyRecord:
|
|
"""Build a LatencyRecord with a specific e2e_asr_to_tts_ms value."""
|
|
# e2e = tts_first_audio_ms - transcript_ready_ms. Use a non-zero base
|
|
# because LatencyRecord.e2e_asr_to_tts_ms guards on truthiness (0.0 is falsy).
|
|
base = 100.0
|
|
return LatencyRecord(
|
|
transcript_ready_ms=base,
|
|
tts_first_audio_ms=base + e2e_ms,
|
|
)
|
|
|
|
|
|
def test_latency_empty_returns_none():
|
|
m = AssistLatencyMetrics()
|
|
assert m.p50() is None
|
|
assert m.p95() is None
|
|
assert m.p99() is None
|
|
s = m.summary()
|
|
assert s["count"] == 0
|
|
assert s["p95"] is None
|
|
assert s["within_target"] is False # no records → not within target
|
|
assert s["within_pilot"] is False
|
|
|
|
|
|
def test_latency_p95_p50_p99_computed():
|
|
"""100 mock records: some <600ms, some 600-650ms, some >650ms.
|
|
|
|
Verifies p50/p95/p99 are computed correctly + the within_target/within_pilot
|
|
flags reflect the p95 against the D-072 thresholds.
|
|
"""
|
|
m = AssistLatencyMetrics()
|
|
# 80 records < 600ms (within target), 15 records 600-650ms (within pilot),
|
|
# 5 records > 650ms (over pilot tolerance).
|
|
for i in range(80):
|
|
m.record(_record(500.0 + i)) # 500..579ms
|
|
for i in range(15):
|
|
m.record(_record(610.0 + i)) # 610..624ms
|
|
for i in range(5):
|
|
m.record(_record(700.0 + i)) # 700..704ms
|
|
|
|
s = m.summary()
|
|
assert s["count"] == 100
|
|
assert s["p50"] is not None
|
|
assert s["p95"] is not None
|
|
assert s["p99"] is not None
|
|
# p50 should be in the < 600ms range (median of the 80 < 600ms records).
|
|
assert s["p50"] < 600.0
|
|
# p95: nearest-rank index = ceil(0.95 * 100) - 1 = 94 (0-indexed) → the 95th
|
|
# sorted value. 80 records are 500..579, 15 are 610..624, 5 are 700..704.
|
|
# Sorted: [500..579 (80), 610..624 (15), 700..704 (5)]. Index 94 → 610..624
|
|
# range (index 80..94 = the 610..624 set; index 94 = 624.0).
|
|
assert 610.0 <= s["p95"] <= 625.0
|
|
# p99: index = ceil(0.99 * 100) - 1 = 98 → the 99th sorted value (700..704).
|
|
assert s["p99"] >= 700.0
|
|
# D-072: within_target = (p95 < 600). p95 is ~624 → not within target.
|
|
assert s["within_target"] is False
|
|
# D-072: within_pilot = (p95 <= 650). p95 is ~624 → within pilot.
|
|
assert s["within_pilot"] is True
|
|
# D-072 thresholds documented in the summary.
|
|
assert s["target_ms"] == TARGET_MS == 600
|
|
assert s["pilot_tolerance_ms"] == PILOT_TOLERANCE_MS == 650
|
|
|
|
|
|
def test_latency_within_target_when_p95_under_600():
|
|
"""All records < 600ms → within_target=True, within_pilot=True."""
|
|
m = AssistLatencyMetrics()
|
|
for i in range(20):
|
|
m.record(_record(400.0 + i)) # 400..419ms
|
|
s = m.summary()
|
|
assert s["p95"] < 600.0
|
|
assert s["within_target"] is True
|
|
assert s["within_pilot"] is True
|
|
|
|
|
|
def test_latency_over_pilot_when_p95_over_650():
|
|
"""All records > 650ms → within_target=False, within_pilot=False."""
|
|
m = AssistLatencyMetrics()
|
|
for i in range(20):
|
|
m.record(_record(700.0 + i)) # 700..719ms
|
|
s = m.summary()
|
|
assert s["p95"] > 650.0
|
|
assert s["within_target"] is False
|
|
assert s["within_pilot"] is False
|
|
|
|
|
|
def test_latency_pilot_boundary_exactly_650():
|
|
"""D-072 boundary: p95 == 650 → within_pilot=True (≤ is inclusive)."""
|
|
m = AssistLatencyMetrics()
|
|
# 20 records all exactly 650ms → p95 = 650.0
|
|
for _ in range(20):
|
|
m.record(_record(650.0))
|
|
s = m.summary()
|
|
assert s["p95"] == 650.0
|
|
assert s["within_pilot"] is True # ≤ 650 (inclusive)
|
|
assert s["within_target"] is False # < 600 (strict)
|
|
|
|
|
|
def test_latency_d072_thresholds_documented():
|
|
"""D-072: the pilot tolerance (≤650ms) + target (<600ms) are documented."""
|
|
assert TARGET_MS == 600
|
|
assert PILOT_TOLERANCE_MS == 650
|
|
assert PILOT_TOLERANCE_MS > TARGET_MS # pilot tolerance is more lenient
|
|
|
|
|
|
# ── GuardrailMetrics ────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_guardrail_fp_rate_on_coaching_corpus():
|
|
"""FP rate on the tuning corpus < 5% (REQ-IDEATE-04 target)."""
|
|
gm = GuardrailMetrics()
|
|
rate, mis, total = await gm.false_positive_rate()
|
|
print(f"\n[nfr] guardrail FP rate: {rate:.1%} ({mis}/{total})")
|
|
assert rate < 0.05, (
|
|
f"guardrail FP rate {rate:.1%} exceeds 5% target — the regex is "
|
|
f"over-matching coaching responses. {mis}/{total} blocked."
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_guardrail_fn_rate_on_direct_corpus():
|
|
"""FN rate on the direct-answer corpus < 5% (REQ-IDEATE-04 target)."""
|
|
gm = GuardrailMetrics()
|
|
rate, mis, total = await gm.false_negative_rate()
|
|
print(f"\n[nfr] guardrail FN rate: {rate:.1%} ({mis}/{total})")
|
|
assert rate < 0.05, (
|
|
f"guardrail FN rate {rate:.1%} exceeds 5% target — the regex is "
|
|
f"under-matching direct answers. {mis}/{total} allowed."
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_guardrail_adversarial_fn_measured():
|
|
"""Adversarial FN rate measured + reported (G-067 — ≤ 20% pilot threshold).
|
|
|
|
This test does NOT assert the 5% target (the adversarial set is the
|
|
residual-risk set, not the tuning target). It asserts the measurement
|
|
infrastructure works + the rate is within the G-067 pilot threshold (≤ 20%).
|
|
"""
|
|
gm = GuardrailMetrics()
|
|
rate, mis, total = await gm.adversarial_false_negative_rate()
|
|
print(f"\n[nfr] guardrail adversarial FN rate: {rate:.1%} ({mis}/{total})")
|
|
# G-067: ≤ 20% pilot threshold (the binding contract from GRILL-v0.5).
|
|
assert rate <= 0.20, (
|
|
f"adversarial FN rate {rate:.1%} exceeds G-067 ≤20% threshold — "
|
|
f"re-tune the regex or escalate. {mis}/{total} slipped through."
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_guardrail_nightly_trend_on_mock_turns(tmp_path: Path):
|
|
"""nightly_trend() samples 24h of assist turns + reports fn_candidates.
|
|
|
|
Seeds a temp SQLite store with assist turns (some coaching, some with
|
|
direct-answer heuristic patterns) + verifies the nightly trend detects
|
|
fn_candidates.
|
|
"""
|
|
from db.migrate import apply_migrations
|
|
from db.store import PraxisStore
|
|
|
|
db = tmp_path / "test_nfr_nightly.db"
|
|
apply_migrations(db)
|
|
store = PraxisStore(db)
|
|
await store.init()
|
|
|
|
# Seed an assist session + turns.
|
|
session_id = await store.start_session_typed(
|
|
"learner-1", "assist:refund", session_type="assist"
|
|
)
|
|
# Turn 1: a coaching response (allowed, no fn_candidate).
|
|
await store.log_turn_with_verdict(
|
|
session_id, 0, role="assistant",
|
|
asr_text="customer wants refund",
|
|
tts_text="What do you think the customer needs right now?",
|
|
latency_ms=580.0,
|
|
guardrail_verdict_json=json.dumps({"allowed": True, "category": "coaching"}),
|
|
)
|
|
# Turn 2: a direct-answer response that slipped past the guardrail
|
|
# (allowed=True in the verdict, but the heuristic catches it).
|
|
await store.log_turn_with_verdict(
|
|
session_id, 1, role="assistant",
|
|
asr_text="what should I say",
|
|
tts_text="You should say: I'm sorry, here's a refund.",
|
|
latency_ms=590.0,
|
|
guardrail_verdict_json=json.dumps({"allowed": True, "category": "coaching"}),
|
|
)
|
|
# Turn 3: a blocked response (guardrail caught it).
|
|
await store.log_turn_with_verdict(
|
|
session_id, 2, role="assistant",
|
|
asr_text="help me",
|
|
tts_text="Tell the customer: we will issue a full refund now.",
|
|
latency_ms=570.0,
|
|
guardrail_verdict_json=json.dumps({"allowed": False, "category": "blocked_direct_script"}),
|
|
)
|
|
|
|
gm = GuardrailMetrics()
|
|
trend = await gm.nightly_trend(store)
|
|
|
|
assert trend["total_turns"] == 3
|
|
assert trend["blocked"] == 1
|
|
# Turn 2 should be flagged as an fn_candidate. The guardrail re-check may
|
|
# catch it as a regression (it now blocks what it previously allowed) OR
|
|
# the heuristic may catch it as a direct-answer pattern. Either way, it
|
|
# must appear in fn_candidates.
|
|
assert len(trend["fn_candidates"]) >= 1
|
|
seqs = [c.get("turn_seq") for c in trend["fn_candidates"]]
|
|
assert 1 in seqs, "turn 2 (direct-answer that slipped past) must be flagged"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_guardrail_nightly_trend_empty_store(tmp_path: Path):
|
|
"""nightly_trend() on an empty store returns zeros + no fn_candidates."""
|
|
from db.migrate import apply_migrations
|
|
from db.store import PraxisStore
|
|
|
|
db = tmp_path / "test_nfr_nightly_empty.db"
|
|
apply_migrations(db)
|
|
store = PraxisStore(db)
|
|
await store.init()
|
|
|
|
gm = GuardrailMetrics()
|
|
trend = await gm.nightly_trend(store)
|
|
assert trend["total_turns"] == 0
|
|
assert trend["blocked"] == 0
|
|
assert trend["fn_candidates"] == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_guardrail_nightly_trend_excludes_practice_turns(tmp_path: Path):
|
|
"""nightly_trend() only samples assist turns (not practice turns)."""
|
|
from db.migrate import apply_migrations
|
|
from db.store import PraxisStore
|
|
|
|
db = tmp_path / "test_nfr_nightly_practice.db"
|
|
apply_migrations(db)
|
|
store = PraxisStore(db)
|
|
await store.init()
|
|
|
|
# Seed a practice session (NOT assist) with a turn.
|
|
practice_id = await store.start_session_typed(
|
|
"learner-1", "cs_refund_ca_v01", session_type="practice"
|
|
)
|
|
await store.log_turn_with_verdict(
|
|
practice_id, 0, role="assistant",
|
|
asr_text="hello",
|
|
tts_text="You should say sorry.",
|
|
latency_ms=500.0,
|
|
guardrail_verdict_json=json.dumps({"allowed": True, "category": "coaching"}),
|
|
)
|
|
|
|
gm = GuardrailMetrics()
|
|
trend = await gm.nightly_trend(store)
|
|
# Practice turns must NOT appear in the assist nightly trend.
|
|
assert trend["total_turns"] == 0
|
|
assert trend["fn_candidates"] == [] |