feat(P05): defense endpoints (Wave 3)

Task 5-3-01: api/defense.py — POST /v1/defense/start (DefenseRecord + first examiner
question persisted + voice descriptor + trace_complete disclosure — the defense does
NOT gate on completeness, the grader does per G-4), POST /{id}/answer ({text} or
multipart audio -> STT via VoiceProvider; follow-up question; per-turn latency
stt_ms/llm_ms/tts_ms), GET /{id}/audio/{turn_id} (streaming TTS WAV), POST /{id}/finish
(DefenseVerdict via D-020 + A-109 integrity signals: long pauses computed from turn
metadata at PAUSE_THRESHOLD_MS=15s), GET /{id} (ordered transcript + signals + status).
Lifespan wires DefenseStore + voice provider + ExaminerAgent. python-multipart dep for
audio parsing.

11 endpoint tests green (full loop mock voice + mock LLM; long-pause signal; 404s; 422
no-body; audio streaming). Suite 378 green; ruff clean.

---ci---
phase: 5
milestone: v0.3
status: execute
requirements: {covered: [REQ-3-006], partial: []}
---/ci---
This commit is contained in:
CIAgent
2026-09-12 04:37:30 +00:00
parent f3071e4b79
commit 04bdccf189
6 changed files with 524 additions and 0 deletions
@@ -5,6 +5,7 @@ Boundary rule: api/ composes agents/ and llm/; they never import api/.
from .assessment import router as assessment_router
from .chat import router as chat_router
from .defense import router as defense_router
from .lab import router as lab_router
from .mentor import router as mentor_router
from .proctor import router as proctor_router
@@ -21,4 +22,5 @@ __all__ = [
"sandboxes_router",
"telemetry_router",
"variants_router",
"defense_router",
]
+300
View File
@@ -0,0 +1,300 @@
"""Oral-defense endpoints (Task 5-3-01, REQ-3-006, A-109).
Full defense loop over HTTP with mock-first voice (D-030) and the seventh
Examiner agent (SSE question streaming happens through the chat pipeline;
these endpoints are the session orchestration + transcript persistence):
POST /v1/defense/start {learner_id, task_id}
POST /v1/defense/{id}/answer {text} | multipart audio (STT)
GET /v1/defense/{id}/audio/{turn_id} TTS bytes (streaming)
POST /v1/defense/{id}/finish verdict + integrity signals
GET /v1/defense/{id} transcript + signals
Integrity signals (A-109) are computed server-side from turn metadata:
long pauses = learner turns whose latency_ms exceeds PAUSE_THRESHOLD_MS.
The defense does NOT gate on trace completeness (the grader does, G-4);
an incomplete trace is surfaced as `trace_complete: false` so the UI can
disclose it before the learner defends.
"""
from __future__ import annotations
import time
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from ..agents.examiner import ExaminerAgent
from ..grading.features import TraceDigest, compute_digest
from ..llm.types import Message
from ..voice.base import VoiceDescriptor
from ..voice.defense_store import DefenseRecord, DefenseStore, DefenseTurn
from .deps import (
get_examiner,
get_settings,
get_trace_store,
get_variant_store,
get_voice_provider,
get_voice_store,
)
if TYPE_CHECKING: # pragma: no cover
pass
router = APIRouter(prefix="/v1/defense", tags=["defense"])
#: A-109: learner turns slower than this are flagged as long pauses (ms).
PAUSE_THRESHOLD_MS = 15_000
_ROLE_EXAMINER = "examiner"
_ROLE_LEARNER = "learner"
class StartRequest(BaseModel):
learner_id: str = Field(min_length=1)
task_id: str = Field(min_length=1)
class StartResponse(BaseModel):
defense_id: str
voice_descriptor: dict
trace_complete: bool
first_question: str
class AnswerResponse(BaseModel):
question: str
turn_latency: dict[str, int | None]
class FinishResponse(BaseModel):
verdict: dict
integrity_signals: dict
async def _digest_for_task(
trace_store, learner_id: str, task_id: str
) -> tuple[TraceDigest | None, bool]:
"""Digest of the learner's trace for this task + completeness flag."""
if not trace_store.list_tasks(learner_id) or task_id not in trace_store.list_tasks(
learner_id
):
return None, True # no trace at all is "complete" for defense purposes
trace = trace_store.get_trace(learner_id, task_id)
gaps = trace_store.gaps(learner_id, task_id)
return (compute_digest(trace) if trace else None), (len(gaps) == 0)
@router.post("/start", response_model=StartResponse)
async def start_defense(
body: StartRequest,
examiner: ExaminerAgent = Depends(get_examiner),
voice_store: DefenseStore = Depends(get_voice_store),
voice_provider=Depends(get_voice_provider),
trace_store=Depends(get_trace_store),
variant_store=Depends(get_variant_store),
settings=Depends(get_settings),
) -> StartResponse:
record = voice_store.start(
DefenseRecord(
id=f"dfn-{int(time.time() * 1000):x}-{body.learner_id[:8]}",
learner_id=body.learner_id,
task_id=body.task_id,
status="in_progress",
created_at=datetime.now(UTC),
)
)
digest, trace_complete = await _digest_for_task(trace_store, body.learner_id, body.task_id)
variant = variant_store.get_by_task(body.task_id)
statement = variant.statement if variant is not None else None
started = time.perf_counter()
question = await examiner.next_question(
history=[], trace_digest=digest, variant_statement=statement
)
llm_ms = int((time.perf_counter() - started) * 1000)
voice_store.append_turn(
record.id,
DefenseTurn(
defense_id=record.id,
seq=0,
role=_ROLE_EXAMINER,
text=question,
ts=datetime.now(UTC),
latency_ms=llm_ms,
created_at=datetime.now(UTC),
),
)
descriptor = getattr(voice_provider, "descriptor", None) or VoiceDescriptor(
mode="mock", sr_available=True, tts_available=True
)
return StartResponse(
defense_id=record.id,
voice_descriptor=descriptor.model_dump(),
trace_complete=trace_complete,
first_question=question,
)
@router.post("/{defense_id}/answer", response_model=AnswerResponse)
async def answer_defense(
defense_id: str,
text: str | None = Form(default=None),
audio: UploadFile | None = File(default=None),
voice_store: DefenseStore = Depends(get_voice_store),
voice_provider=Depends(get_voice_provider),
examiner: ExaminerAgent = Depends(get_examiner),
trace_store=Depends(get_trace_store),
variant_store=Depends(get_variant_store),
) -> AnswerResponse:
record = voice_store.get(defense_id)
if record is None:
raise HTTPException(status_code=404, detail=f"no defense {defense_id!r}")
if text is None and audio is None:
raise HTTPException(status_code=422, detail="provide {text} or audio")
# STT (typed fallback bypasses the voice provider entirely).
stt_ms: int | None = None
if audio is not None:
stt_started = time.perf_counter()
raw = await audio.read()
fmt = (audio.content_type or "audio/wav").split("/")[-1]
segment = await voice_provider.transcribe(raw, fmt)
stt_ms = int((time.perf_counter() - stt_started) * 1000)
text = segment.text
turns = record.turns if hasattr(record, "turns") else []
history = [
Message(role="assistant" if t.role == _ROLE_EXAMINER else "user", content=t.text)
for t in turns
]
next_seq = len(turns)
voice_store.append_turn(
defense_id,
DefenseTurn(
defense_id=defense_id,
seq=next_seq,
role=_ROLE_LEARNER,
text=text or "",
ts=datetime.now(UTC),
latency_ms=stt_ms,
created_at=datetime.now(UTC),
),
)
digest, _ = await _digest_for_task(trace_store, record.learner_id, record.task_id)
variant = variant_store.get_by_task(record.task_id)
llm_started = time.perf_counter()
question = await examiner.next_question(
history=history + [Message(role="user", content=text or "")],
trace_digest=digest,
variant_statement=variant.statement if variant is not None else None,
)
llm_ms = int((time.perf_counter() - llm_started) * 1000)
voice_store.append_turn(
defense_id,
DefenseTurn(
defense_id=defense_id,
seq=next_seq + 1,
role=_ROLE_EXAMINER,
text=question,
ts=datetime.now(UTC),
latency_ms=llm_ms,
created_at=datetime.now(UTC),
),
)
return AnswerResponse(
question=question,
turn_latency={"stt_ms": stt_ms, "llm_ms": llm_ms, "tts_ms": None},
)
@router.get("/{defense_id}/audio/{turn_id}")
async def defense_audio(
defense_id: str,
turn_id: int,
voice_store: DefenseStore = Depends(get_voice_store),
voice_provider=Depends(get_voice_provider),
):
record = voice_store.get(defense_id)
if record is None:
raise HTTPException(status_code=404, detail=f"no defense {defense_id!r}")
turn = next((t for t in record.turns if t.seq == turn_id), None)
if turn is None or turn.role != _ROLE_EXAMINER:
raise HTTPException(status_code=404, detail=f"no examiner turn {turn_id!r}")
async def stream():
async for chunk in voice_provider.synthesize(turn.text):
yield chunk
return StreamingResponse(stream(), media_type="audio/wav")
@router.post("/{defense_id}/finish", response_model=FinishResponse)
async def finish_defense(
defense_id: str,
voice_store: DefenseStore = Depends(get_voice_store),
examiner: ExaminerAgent = Depends(get_examiner),
trace_store=Depends(get_trace_store),
variant_store=Depends(get_variant_store),
) -> FinishResponse:
record = voice_store.get(defense_id)
if record is None:
raise HTTPException(status_code=404, detail=f"no defense {defense_id!r}")
turns = record.turns if hasattr(record, "turns") else []
history = [
Message(role="assistant" if t.role == _ROLE_EXAMINER else "user", content=t.text)
for t in turns
]
digest, _ = await _digest_for_task(trace_store, record.learner_id, record.task_id)
variant = variant_store.get_by_task(record.task_id)
verdict = await examiner.final_verdict(
history=history,
trace_digest=digest,
variant_statement=variant.statement if variant is not None else None,
)
signals: dict = {
"long_pauses": [
{"turn": t.seq, "latency_ms": t.latency_ms}
for t in turns
if t.role == _ROLE_LEARNER and (t.latency_ms or 0) > PAUSE_THRESHOLD_MS
],
"pause_threshold_ms": PAUSE_THRESHOLD_MS,
}
voice_store.finalize(defense_id, signals)
return FinishResponse(verdict=verdict.model_dump(), integrity_signals=signals)
@router.get("/{defense_id}")
async def get_defense(
defense_id: str,
voice_store: DefenseStore = Depends(get_voice_store),
):
record = voice_store.get(defense_id)
if record is None:
raise HTTPException(status_code=404, detail=f"no defense {defense_id!r}")
return {
"defense_id": record.id,
"learner_id": record.learner_id,
"task_id": record.task_id,
"status": record.status,
"turns": [
{
"seq": t.seq,
"role": t.role,
"text": t.text,
"ts": t.ts,
"latency_ms": t.latency_ms,
}
for t in record.turns
],
"integrity_signals": record.integrity_signals or {},
}
+14
View File
@@ -2,6 +2,7 @@
from fastapi import Request
from ..agents.examiner import ExaminerAgent
from ..agents.registry import AgentRegistry
from ..agents.session import SessionStore
from ..config import Settings
@@ -14,6 +15,8 @@ from ..telemetry.ingest import TraceIntegrityMap
from ..telemetry.store import TraceStore
from ..variants.generator import VariantGenerator
from ..variants.store import VariantStore
from ..voice.base import VoiceProvider
from ..voice.defense_store import DefenseStore
def get_settings(request: Request) -> Settings:
@@ -63,3 +66,14 @@ def get_variant_generator(request: Request) -> VariantGenerator:
def get_variant_store(request: Request) -> VariantStore:
return request.app.state.variant_store
def get_voice_store(request: Request) -> DefenseStore:
return request.app.state.defense_store
def get_voice_provider(request: Request) -> VoiceProvider:
return request.app.state.voice_provider
def get_examiner(request: Request) -> ExaminerAgent:
return request.app.state.examiner_agent
+20
View File
@@ -14,6 +14,7 @@ from .agents.session import InMemorySessionStore
from .api import (
assessment_router,
chat_router,
defense_router,
lab_router,
mentor_router,
proctor_router,
@@ -30,6 +31,8 @@ from .telemetry.ingest import TraceIntegrityMap
from .telemetry.store import SQLiteTraceStore
from .variants.generator import VariantGenerator
from .variants.store import SQLiteVariantStore
from .voice.defense_store import SQLiteDefenseStore
from .voice.factory import voice_provider_from_settings
logger = logging.getLogger(__name__)
@@ -100,6 +103,21 @@ def create_app(settings: Settings | None = None) -> FastAPI:
model=settings.model,
)
# Oral defense (REQ-3-006): DefenseStore (same SQLite file) + the
# mock-first voice provider (D-030) + the seventh Examiner agent.
# Tests may pre-set app.state.defense_store / voice_provider /
# examiner_agent (state-injection override; never rebuilt if pre-set).
defense_store = getattr(app.state, "defense_store", None)
if defense_store is None:
defense_store = SQLiteDefenseStore(db_path=settings.db_path)
app.state.defense_store = defense_store
if getattr(app.state, "voice_provider", None) is None:
app.state.voice_provider = voice_provider_from_settings(settings)
if getattr(app.state, "examiner_agent", None) is None:
from .agents.examiner import ExaminerAgent
app.state.examiner_agent = ExaminerAgent(app.state.provider, settings)
# Grading persistence + engine (REQ-3-004): GradeStore from the same
# SQLite file as traces (D-027), one GradingEngine singleton wired
# through app.state — the engine receives its stores via constructor
@@ -144,6 +162,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
trace_store.close()
grade_store.close()
variant_store.close()
defense_store.close()
await app.state.http_client.aclose()
app = FastAPI(title="Nextcraft AI Service", version="0.3.0", lifespan=lifespan)
@@ -173,6 +192,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
app.include_router(sandboxes_router)
app.include_router(telemetry_router)
app.include_router(variants_router)
app.include_router(defense_router)
return app
+3
View File
@@ -18,6 +18,9 @@ dependencies = [
"sqlalchemy>=2.0,<2.1",
"websockets>=13,<16",
"aiofiles>=24.1,<26",
# POST /v1/defense/{id}/answer multipart audio (REQ-3-006): FastAPI
# form/File parsing requires python-multipart at runtime.
"python-multipart>=0.0.32,<0.1",
]
[project.optional-dependencies]
+185
View File
@@ -0,0 +1,185 @@
"""Defense endpoint tests (Task 5-3-01, REQ-3-006) — mock voice + mock LLM."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from ai_service.agents.examiner import ExaminerAgent
from ai_service.config import Settings
from ai_service.grading.store import SQLiteGradeStore
from ai_service.llm.mock import MockProvider
from ai_service.main import create_app
from ai_service.telemetry.ingest import TraceIntegrityMap
from ai_service.telemetry.store import SQLiteTraceStore
from ai_service.variants.store import SQLiteVariantStore
from ai_service.voice.defense_store import SQLiteDefenseStore
from ai_service.voice.mock import MockVoiceProvider
VERDICT = {
"verdict": "developing",
"understanding": "Explains the build clearly.",
"process_justification": "Justifies choices.",
"communication": "Clear and specific.",
"strengths": ["Grounded answers in the digest."],
"gaps": ["Did not address the edge cases."],
}
class ScriptedLLM(MockProvider):
"""Question-mode calls get a question; verdict-mode calls get D-020 JSON.
Discriminator: the verdict prompt contains "final verdict JSON" — the
question prompt says "next question".
"""
def __init__(self) -> None:
super().__init__()
async def chat(self, messages, *, model, temperature=0.7, response_format=None):
all_text = "\n".join(m.content for m in messages)
if "final verdict JSON" in all_text:
return json.dumps(VERDICT)
return "Why did you structure the fix that way?"
@pytest.fixture()
def app(tmp_path: Path):
application = create_app(Settings(provider="mock", voice_provider="mock"))
llm = ScriptedLLM()
application.state.provider = llm
application.state.trace_store = SQLiteTraceStore(db_path=tmp_path / "t.db")
application.state.variant_store = SQLiteVariantStore(db_path=tmp_path / "v.db")
application.state.grade_store = SQLiteGradeStore(db_path=tmp_path / "g.db")
application.state.trace_integrity = TraceIntegrityMap()
application.state.defense_store = SQLiteDefenseStore(db_path=tmp_path / "d.db")
application.state.voice_provider = MockVoiceProvider(
["the fix was in the retry loop"]
)
settings = Settings(provider="mock", voice_provider="mock")
application.state.examiner_agent = ExaminerAgent(llm, settings)
return application
@pytest.fixture()
def client(app) -> TestClient:
with TestClient(app) as c:
yield c
def _start(client: TestClient) -> dict:
resp = client.post(
"/v1/defense/start", json={"learner_id": "defense-learner", "task_id": "defense-task"}
)
assert resp.status_code == 200, resp.text
return resp.json()
class TestStart:
def test_start_returns_first_question_and_descriptor(self, client) -> None:
body = _start(client)
assert body["first_question"]
assert body["defense_id"]
assert body["voice_descriptor"]["mode"] == "mock"
assert body["trace_complete"] is True
stored = client.get(f"/v1/defense/{body['defense_id']}")
assert stored.status_code == 200
turns = stored.json()["turns"]
assert turns and turns[0]["role"] == "examiner"
def test_start_with_unknown_trace_is_complete_flag(self, client) -> None:
body = _start(client)
assert body["trace_complete"] is True
class TestAnswer:
def test_typed_answer_yields_followup_with_latency(self, client) -> None:
defense_id = _start(client)["defense_id"]
resp = client.post(
f"/v1/defense/{defense_id}/answer", data={"text": "I fixed the loop."}
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["question"]
assert body["turn_latency"]["llm_ms"] is not None
def test_audio_answer_transcribed_and_recorded(self, client) -> None:
defense_id = _start(client)["defense_id"]
wav_bytes = b"RIFF" + b"\x00" * 64
resp = client.post(
f"/v1/defense/{defense_id}/answer",
files={"audio": ("answer.wav", wav_bytes, "audio/wav")},
)
assert resp.status_code == 200, resp.text
stored = client.get(f"/v1/defense/{defense_id}").json()
learner_turns = [t for t in stored["turns"] if t["role"] == "learner"]
assert learner_turns, "learner turn missing after audio answer"
assert learner_turns[0]["text"] == "the fix was in the retry loop"
def test_neither_text_nor_audio_422(self, client) -> None:
defense_id = _start(client)["defense_id"]
resp = client.post(f"/v1/defense/{defense_id}/answer")
assert resp.status_code == 422
def test_unknown_defense_404(self, client) -> None:
resp = client.post("/v1/defense/dfn-nope/answer", data={"text": "hi"})
assert resp.status_code == 404
class TestAudioEndpoint:
def test_examiner_turn_streams_wav(self, client) -> None:
defense_id = _start(client)["defense_id"]
resp = client.get(f"/v1/defense/{defense_id}/audio/0")
assert resp.status_code == 200
assert resp.content
assert resp.headers["content-type"].startswith("audio/")
def test_unknown_turn_404(self, client) -> None:
defense_id = _start(client)["defense_id"]
assert client.get(f"/v1/defense/{defense_id}/audio/42").status_code == 404
class TestFinishAndGet:
def test_full_loop_verdict_and_signals(self, client) -> None:
defense_id = _start(client)["defense_id"]
client.post(f"/v1/defense/{defense_id}/answer", data={"text": "answer one"})
finish = client.post(f"/v1/defense/{defense_id}/finish")
assert finish.status_code == 200, finish.text
body = finish.json()
assert body["verdict"]["verdict"] == "developing"
assert body["integrity_signals"]["pause_threshold_ms"]
stored = client.get(f"/v1/defense/{defense_id}").json()
assert stored["status"] == "finished"
assert stored["integrity_signals"]
def test_long_pause_flagged(self, client, app) -> None:
from datetime import UTC, datetime
from ai_service.voice.defense_store import DefenseTurn
defense_id = _start(client)["defense_id"]
# inject a slow learner turn directly (simulated latency)
store = app.state.defense_store
store.append_turn(
defense_id,
DefenseTurn(
defense_id=defense_id,
seq=99,
role="learner",
text="slow reply",
ts=datetime.now(UTC),
latency_ms=30_000,
created_at=datetime.now(UTC),
),
)
finish = client.post(f"/v1/defense/{defense_id}/finish")
assert finish.status_code == 200
signals = finish.json()["integrity_signals"]
assert any(p["turn"] == 99 for p in signals["long_pauses"])
def test_unknown_defense_404_on_all(self, client) -> None:
assert client.post("/v1/defense/dfn-nope/finish").status_code == 404
assert client.get("/v1/defense/dfn-nope").status_code == 404