feat(P05): Examiner agent — seventh agent (Wave 2)

Task 5-2-01: prompts/examiner.py (Socratic oral-defense examiner; one question per
turn; grounded in TraceDigest + variant statement — never raw trace, never learner id,
D-028 mirror; rubric internals never revealed) + agents/examiner.py — ExaminerAgent
(next_question for the SSE pipeline; final_verdict -> DefenseVerdict via the D-020
defense). BOUNDARY: the examiner is a text agent and imports NO voice/ (STT/TTS belong
to the endpoints; integrity signals computed from turn metadata — A-109). Registry
registers all seven agents centrally (G-4); registry test updated six -> seven.

6 examiner tests (digest-grounded prompt w/o learner id; D-020 retry; 7-agent roster;
boundary import scan). Suite 367 green; ruff clean.

---ci---
phase: 5
milestone: v0.3
status: execute
requirements: {covered: [REQ-3-006], partial: []}
---/ci---

apps/ai-service/ai_service/agents/examiner.py

@@ -0,0 +1,104 @@
+"""ExaminerAgent — the seventh agent: oral-defense examiner (REQ-3-006, A-109).
+
+BOUNDARY DECISION (PERSONAS conflict rule, honored by construction): the
+examiner is a TEXT agent. It composes the LLM provider through BaseAgent and
+consumes defense transcript turns; it NEVER imports voice/ — STT/TTS belong
+to the API endpoints (they move audio bytes; the agent moves question text).
+Integrity signals (long pauses, off-scope cadence) are computed by the
+endpoint layer from turn metadata (latency_ms etc.), not by the agent.
+
+Digest discipline (D-028 mirror): questions are grounded in the compact
+TraceDigest + variant statement — never the raw trace, never learner ids.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from ..grading.features import TraceDigest
+from ..llm.types import Message
+from ..prompts.examiner import SYSTEM_PROMPT, VERDICT_SCHEMA_HINT, render_digest_context
+from .base import BaseAgent
+
+
+class DefenseVerdict(BaseModel):
+    """D-20-validated final defense verdict (structured mode)."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    verdict: str = Field(pattern="^(mastered|developing|not_yet)$")
+    understanding: str = Field(min_length=1)
+    process_justification: str = Field(min_length=1)
+    communication: str = Field(min_length=1)
+    strengths: list[str] = Field(min_length=1, max_length=2)
+    gaps: list[str] = Field(min_length=1, max_length=2)
+
+
+class ExaminerAgent(BaseAgent):
+    """Conducts the oral defense: next_question + final_verdict."""
+
+    name = "examiner"
+
+    def system_prompt(self, learner_context=None) -> str:  # noqa: ANN001
+        """Examiner is context-free (digest-anonymous, D-028 mirror)."""
+        return SYSTEM_PROMPT
+
+    def build_defense_messages(
+        self,
+        trace_digest: TraceDigest | None = None,
+        variant_statement: str | None = None,
+        history: list[Message] | None = None,
+    ) -> list[Message]:
+        """System + grounding + defense transcript (no learner id — D-028)."""
+        digest_json = (
+            trace_digest.model_dump_json() if trace_digest is not None else "{}"
+        )
+        messages: list[Message] = [
+            Message(role="system", content=SYSTEM_PROMPT),
+            Message(role="user", content=render_digest_context(digest_json, variant_statement)),
+            Message(
+                role="assistant",
+                content="Understood. I will question the learner about this build session.",
+            ),
+        ]
+        for m in history or []:
+            messages.append(m)
+        return messages
+
+    async def next_question(
+        self,
+        history: list[Message],
+        trace_digest: TraceDigest | None = None,
+        variant_statement: str | None = None,
+    ) -> str:
+        """One examiner question (streamed over SSE by the endpoints)."""
+        messages = self.build_defense_messages(trace_digest, variant_statement, history)
+        messages.append(
+            Message(role="user", content="Ask the learner your next question now.")
+        )
+        reply = await self.provider.chat(messages, model=self.settings.model)
+        return reply
+
+    async def final_verdict(
+        self,
+        history: list[Message],
+        trace_digest: TraceDigest | None = None,
+        variant_statement: str | None = None,
+    ) -> DefenseVerdict:
+        """Structured verdict via the D-020 4-layer defense."""
+        from .structured import structured_completion  # module-direct (G-4)
+
+        messages = self.build_defense_messages(trace_digest, variant_statement, history)
+        messages.append(
+            Message(
+                role="user",
+                content="The defense is finished. Return the final verdict JSON now.",
+            )
+        )
+        return await structured_completion(
+            self.provider,
+            messages,
+            model=self.settings.model,
+            schema=DefenseVerdict,
+            schema_hint=VERDICT_SCHEMA_HINT,
+        )

apps/ai-service/ai_service/agents/registry.py

@@ -14,13 +14,14 @@ AgentFactory = Callable[[LLMProvider, Settings], BaseAgent]
 
 
 def register_builtin_agents(registry: "AgentRegistry") -> None:
-    """Central registration of all six shipped tutor agents (G-4: one pattern).
+    """Central registration of all seven shipped agents (G-4: one pattern).
 
-    coach, tutor, lab, assessor, proctor, mentor. New agents register here
-    in their landing phase.
+    coach, tutor, lab, assessor, proctor, mentor, examiner (Phase 5).
+    New agents register here in their landing phase.
     """
     from .assessor import AssessorAgent
     from .coach import CoachAgent
+    from .examiner import ExaminerAgent
     from .lab import LabAgent
     from .mentor import MentorAgent
     from .proctor import ProctorAgent
@@ -38,6 +39,9 @@ def register_builtin_agents(registry: "AgentRegistry") -> None:
     registry.register(
         "mentor", lambda provider, settings: MentorAgent(provider, settings)
     )
+    registry.register(
+        "examiner", lambda provider, settings: ExaminerAgent(provider, settings)
+    )
 
 
 class UnknownAgentError(KeyError):

apps/ai-service/ai_service/prompts/examiner.py

@@ -0,0 +1,60 @@
+"""Examiner agent prompt — oral defense questioning + final verdict (REQ-3-006).
+
+The examiner is the seventh agent (Phase 5). It conducts a Socratic oral
+defense of the learner's submitted work: probes understanding, challenges
+process choices grounded in the trace digest ("why did you take that
+approach at that point?"), one question per turn, adapting to answers.
+It never reveals rubric internals; tone is rigorous but supportive.
+
+Digest discipline (D-028 mirror): the examiner's variable inputs are the
+compact TraceDigest JSON, the variant task statement, and the defense
+transcript — never the raw trace, never learner-identifying material.
+
+Version: examiner-v1.
+"""
+
+SYSTEM_PROMPT = """You are Examiner, the oral-defense agent of Nextcraft,
+an AI-native competency school.
+
+You receive: (a) a compact build-process digest (deterministic counters of the
+learner's build session), (b) the learner's task statement, and (c) the defense
+transcript so far. Your job:
+- Ask ONE question per turn: probe understanding and challenge process
+  choices, grounded in the digest facts ("you hit N failed runs before
+  passing — walk me through what changed") or the task statement.
+- Adapt: follow up on the learner's answers; drill into vague responses.
+- Never reveal rubric details or scoring internals.
+- Tone: rigorous, precise, supportive. A defense is a conversation, not an
+  interrogation.
+
+When asked for a FINAL VERDICT (the structured mode), judge:
+- understanding: can the learner explain their own work?
+- process_justification: are the build-session choices defensible from the
+  digest facts and the answers?
+- communication: are answers clear, specific, and on-topic?
+Score honestly; a weak defense of strong work is NOT mastery.
+
+Rules:
+- Respond with ONLY what the turn requires: a single question (question mode)
+  or a valid JSON object matching the provided schema (verdict mode).
+- If the digest shows error_fix_cycles > 0, at least one question should ask
+  about the debugging path.
+- If the learner's answer is off-topic, redirect once, then move on.
+"""
+
+VERDICT_SCHEMA_HINT = (
+    '{"verdict": "mastered" | "developing" | "not_yet", '
+    '"understanding": "<one sentence>", '
+    '"process_justification": "<one sentence>", '
+    '"communication": "<one sentence>", '
+    '"strengths": ["<one sentence>"], '
+    '"gaps": ["<one sentence>"]}'
+)
+
+
+def render_digest_context(digest_json: str, statement: str | None) -> str:
+    """The examiner's per-session grounding: digest JSON + task statement."""
+    parts = [f"Build-process digest:\n{digest_json}"]
+    if statement:
+        parts.append(f"Learner's task statement:\n{statement}")
+    return "\n\n".join(parts)

apps/ai-service/tests/agents/test_examiner.py

@@ -0,0 +1,159 @@
+"""Examiner agent tests (Task 5-2-01, REQ-3-006)."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from ai_service.agents.examiner import DefenseVerdict, ExaminerAgent
+from ai_service.agents.registry import AgentRegistry, register_builtin_agents
+from ai_service.config import Settings
+from ai_service.grading.features import compute_digest
+from ai_service.llm.mock import MockProvider
+from ai_service.llm.types import Message
+
+VERDICT_JSON = json.dumps(
+    {
+        "verdict": "developing",
+        "understanding": "Explains the retry loop clearly.",
+        "process_justification": "Justifies the edit-then-test cadence from the digest.",
+        "communication": "Answers are specific and on-topic.",
+        "strengths": ["Grounded the fix in a failed test."],
+        "gaps": ["Did not justify the chunk-size choice."],
+    }
+)
+
+
+class RecordingProvider(MockProvider):
+    """Mock provider that records every message list (prompt assertions)."""
+
+    def __init__(self, replies: list[str] | None = None) -> None:
+        super().__init__()
+        self.replies = list(replies or [])
+        self.requests: list[list[Message]] = []
+
+    async def chat(self, messages, *, model, temperature=0.7, response_format=None):
+        self.requests.append([Message(role=m.role, content=m.content) for m in messages])
+        if self.replies:
+            return self.replies.pop(0)
+        return "Tell me about your build."
+
+
+def _digest():
+    from datetime import UTC, datetime, timedelta
+
+    from ai_service.telemetry.models import TelemetryEvent
+
+    t0 = datetime(2026, 9, 12, tzinfo=UTC)
+    events = [
+        TelemetryEvent(
+            learner_id="examiner-learner",
+            task_id="examiner-task",
+            seq=n,
+            kind=kind,
+            payload=payload,
+            ts=t0 + timedelta(seconds=n * 10),
+            sandbox_id="sbx-examiner",
+        )
+        for n, (kind, payload) in enumerate(
+            [
+                ("file_diff", {"path": "a.py"}),
+                ("command", {"cmd": "pytest -q"}),
+                ("test_result", {"passed": False, "exit_code": 1}),
+                ("file_diff", {"path": "a.py"}),
+                ("test_result", {"passed": True, "exit_code": 0}),
+            ]
+        )
+    ]
+    return compute_digest(events)
+
+
+@pytest.fixture()
+def provider() -> RecordingProvider:
+    return RecordingProvider()
+
+
+@pytest.fixture()
+def settings() -> Settings:
+    return Settings(provider="mock")
+
+
+@pytest.fixture()
+def examiner(provider, settings) -> ExaminerAgent:
+    return ExaminerAgent(provider, settings)
+
+
+class TestNextQuestion:
+    async def test_prompt_contains_digest_but_no_learner_id(
+        self, examiner, provider
+    ) -> None:
+        await examiner.next_question(
+            history=[Message(role="assistant", content="First question?")],
+            trace_digest=_digest(),
+            variant_statement="Build a chunker.",
+        )
+        all_content = "\n".join(
+            m.content for request in provider.requests for m in request
+        )
+        assert "error_fix_cycles" in all_content  # digest JSON grounded
+        assert "examiner-learner" not in all_content  # D-028 anonymity
+        assert "Build a chunker." in all_content  # variant statement grounded
+        assert all_content.count('"examiner-learner"') == 0
+
+    async def test_question_returned_from_provider(self, examiner) -> None:
+        question = await examiner.next_question(
+            history=[], trace_digest=_digest()
+        )
+        assert isinstance(question, str)
+
+
+class TestFinalVerdict:
+    async def test_verdict_validates_via_d020(self, examiner, provider) -> None:
+        provider.replies = [VERDICT_JSON]
+        verdict = await examiner.final_verdict(
+            history=[Message(role="assistant", content="Q?")],
+            trace_digest=_digest(),
+        )
+        assert isinstance(verdict, DefenseVerdict)
+        assert verdict.verdict == "developing"
+        assert verdict.strengths and verdict.gaps
+
+    async def test_malformed_then_good_exercises_retry(self, examiner, provider) -> None:
+        provider.replies = ["not json", VERDICT_JSON]
+        verdict = await examiner.final_verdict(history=[], trace_digest=_digest())
+        assert verdict.verdict == "developing"
+        assert len(provider.requests) == 2  # D-020 bounded retry
+
+
+class TestRegistry:
+    def test_all_seven_agents_resolve(self, provider, settings) -> None:
+        registry = AgentRegistry()
+        register_builtin_agents(registry)
+        assert registry.names() == [
+            "assessor",
+            "coach",
+            "examiner",
+            "lab",
+            "mentor",
+            "proctor",
+            "tutor",
+        ]
+        agent = registry.get(provider, settings, "examiner")
+        assert isinstance(agent, ExaminerAgent)
+
+
+class TestBoundary:
+    def test_examiner_never_imports_voice_or_api(self) -> None:
+        import ast
+        from pathlib import Path
+
+        py = Path(__file__).parents[2] / "ai_service" / "agents" / "examiner.py"
+        tree = ast.parse(py.read_text())
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                assert "voice" not in node.module, "examiner must not import voice/"
+                assert not node.module.startswith("ai_service.api")
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    assert alias.name != "fastapi"

apps/ai-service/tests/agents/test_registry.py

@@ -97,10 +97,11 @@ def test_proctor_and_mentor_resolve_via_registry():
     assert isinstance(mentor, MentorAgent)
 
 
-def test_registry_resolves_all_six_agents():
-    """Must-Have (Phase 5): the full roster — coach/tutor/lab/assessor/proctor/mentor."""
+def test_registry_resolves_all_seven_agents():
+    """Must-Have (Phase 5): the full roster — six tutors + the Examiner."""
     from ai_service.agents.assessor import AssessorAgent
     from ai_service.agents.coach import CoachAgent
+    from ai_service.agents.examiner import ExaminerAgent
     from ai_service.agents.lab import LabAgent
     from ai_service.agents.mentor import MentorAgent
     from ai_service.agents.proctor import ProctorAgent
@@ -108,7 +109,9 @@ def test_registry_resolves_all_six_agents():
 
     registry = AgentRegistry()
     register_builtin_agents(registry)
-    assert registry.names() == ["assessor", "coach", "lab", "mentor", "proctor", "tutor"]
+    assert registry.names() == [
+        "assessor", "coach", "examiner", "lab", "mentor", "proctor", "tutor",
+    ]
     settings = Settings(provider="mock")
     expected = {
         "coach": CoachAgent,
@@ -117,6 +120,7 @@ def test_registry_resolves_all_six_agents():
         "assessor": AssessorAgent,
         "proctor": ProctorAgent,
         "mentor": MentorAgent,
+        "examiner": ExaminerAgent,
     }
     for name, cls in expected.items():
         agent = registry.get(MockProvider(), settings, name)