feat(P06): agent re-grounding on real engine inputs + corpus dormancy (Wave 1)

Tasks 6-1-01..04 (REQ-3-007): Lab consumes the LIVE trace digest (compute_digest over
TraceStore events; empty trace coaches the baseline); Assessor renders coaching FROM
the stored GradeRecord (it never invents scores — the grading engine owns that;
evaluate endpoint re-grounded: 404 without a grade); Proctor consumes digest +
DefenseStore long-pause signals + variant seed cross-check. Corpus telemetry/artifacts
DORMANT (headers + AST dormancy test: zero production importers; learner_context stays
active; retained as Phase-3 calibration history). lifespan now adopts a pre-set
provider (state-injection pattern).

v0.2 corpus-path endpoint tests updated honestly to the live contract (learner_id+task_id).
392 tests green; ruff clean.

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

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

@@ -1,53 +1,34 @@
-"""AssessorAgent — rubric application to pre-baked artifacts (REQ-2-008).
+"""AssessorAgent — rubric coaching over REAL grading output (REQ-3-007).
 
-Structured-output showcase: applies the 4-layer defense (D-020) to return
-a pydantic-validated rubric score. Mock engine inputs (corpus artifacts +
-transcripts); real process-trace grading is v0.3+.
+v0.3 re-grounding: the Assessor no longer invents scores from corpus
+artifacts — the process-trace grading engine (Phase 3) computes and
+persists the validated RubricScore. This agent now renders the STORED
+grade as rubric-anchored coaching: explains the criteria, cites strengths
+and gaps, and frames next steps. Corpus artifacts are retired from this
+path (corpus dormancy, Task 6-1-04).
 """
 
 from pydantic import BaseModel, Field
 
-from ..corpus.artifacts import (
-    ArtifactSubmission,
-    AssessmentRubric,
-    DefenseTranscript,
-    render_rubric,
-    render_transcript,
-)
 from ..corpus.learner_context import LearnerContext, get_learner_context
+from ..grading.store import GradeRecord
 from ..prompts.assessor import SYSTEM_PROMPT, render_context
 from .base import BaseAgent
 
 
-class CriterionScore(BaseModel):
-    criterion_id: str
-    name: str
-    score: int = Field(ge=0, le=100)
-    evidence: str
+class GradeCoaching(BaseModel):
+    """Rubric-anchored coaching rendered FROM the stored grade (not invented)."""
+
+    summary: str = Field(min_length=1)
+    strengths: list[str] = Field(min_length=1, max_length=3)
+    gaps: list[str] = Field(min_length=1, max_length=3)
+    next_steps: list[str] = Field(min_length=1, max_length=3)
 
 
-class RubricScore(BaseModel):
-    rubric_id: str
-    artifact_id: str
-    competency_id: str
-    scores: list[CriterionScore]
-    strengths: list[str] = Field(min_length=1, max_length=2)
-    gaps: list[str] = Field(min_length=1, max_length=2)
-    verdict: str  # "mastered" | "developing" | "not_yet"
-
-    def weighted_total(self, rubric: AssessmentRubric) -> float:
-        by_id = {c.criterion_id: c for c in rubric.criteria}
-        total = 0.0
-        for s in self.scores:
-            total += s.score * by_id[s.criterion_id].weight
-        return total
-
-
-RUBRIC_SCORE_SCHEMA_HINT = (
-    '{"rubric_id": "<id>", "artifact_id": "<id>", "competency_id": "<id>", '
-    '"scores": [{"criterion_id": "<id>", "name": "<name>", "score": <0-100>, '
-    '"evidence": "<one sentence>"}], "strengths": ["<one sentence>"], '
-    '"gaps": ["<one sentence>"], "verdict": "mastered"|"developing"|"not_yet"}'
+GRADE_COACHING_SCHEMA_HINT = (
+    '{"summary": "<two sentences on the grade>", '
+    '"strengths": ["<one sentence>"], "gaps": ["<one sentence>"], '
+    '"next_steps": ["<one sentence>"]}'
 )
 
 
@@ -58,35 +39,25 @@ class AssessorAgent(BaseAgent):
         ctx = learner_context or get_learner_context()
         return SYSTEM_PROMPT.format_map(render_context(ctx))
 
-    def build_evaluation_input(
+    async def coach_grade(
         self,
-        artifact: ArtifactSubmission,
-        rubric: AssessmentRubric,
-        transcript: DefenseTranscript | None,
-    ) -> str:
-        parts = [
-            f"ARTIFACT: {artifact.name} ({artifact.artifact_id})",
-            f"Evidence excerpt: {artifact.evidence_excerpt}",
-            "",
-            render_rubric(rubric),
-        ]
-        if transcript is not None:
-            parts += ["", render_transcript(transcript)]
-        return "\n".join(parts)
-
-    async def evaluate(
-        self,
-        artifact: ArtifactSubmission,
-        rubric: AssessmentRubric,
-        transcript: DefenseTranscript | None,
+        grade: GradeRecord,
         learner_context: LearnerContext | None = None,
-    ) -> RubricScore:
-        evaluation_input = self.build_evaluation_input(artifact, rubric, transcript)
-        score: RubricScore = await self.structured_reply(
+    ) -> GradeCoaching:
+        """Render the STORED grade as coaching via the D-020 defense."""
+        grade_json = {
+            "verdict": grade.verdict,
+            "scores": grade.scores,
+            "digest": grade.digest,
+        }
+        coaching: GradeCoaching = await self.structured_reply(
             history=None,
-            user_input=evaluation_input,
+            user_input=(
+                "The learner's process-trace grade (computed by the grading "
+                f"engine) is:\n{grade_json!r}\nExplain it as coaching."
+            ),
             learner_context=learner_context,
-            schema=RubricScore,
-            schema_hint=RUBRIC_SCORE_SCHEMA_HINT,
+            schema=GradeCoaching,
+            schema_hint=GRADE_COACHING_SCHEMA_HINT,
         )
-        return score
+        return coaching

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

@@ -1,19 +1,24 @@
-"""LabAgent — in-flow feedback over simulated sandbox telemetry (REQ-2-007).
+"""LabAgent — in-flow feedback over LIVE sandbox telemetry (REQ-3-007).
 
-Scenario-driven: consumes a LabTelemetryScenario from the corpus, renders
-the event timeline into the conversation, streams concrete feedback.
-No session chat — each request is one scenario read.
+v0.3 re-grounding: consumes a TraceDigest computed from the learner's real
+trace (grading/features.compute_digest over TraceStore events) — the v0.2
+corpus scenarios are retired from this path (corpus dormancy, Task 6-1-04).
+No session chat — each request is one live-trace read.
 """
 
 from collections.abc import AsyncIterator
+from typing import TYPE_CHECKING
 
 from ..config import Settings
 from ..corpus.learner_context import LearnerContext, get_learner_context
-from ..corpus.telemetry import LabTelemetryScenario, summarize_scenario
+from ..grading.features import TraceDigest
 from ..llm.base import LLMProvider
-from ..prompts.lab import SYSTEM_PROMPT, render_context
+from ..prompts.lab import SYSTEM_PROMPT, render_context, render_digest_timeline
 from .base import BaseAgent
 
+if TYPE_CHECKING:  # pragma: no cover
+    pass
+
 
 class LabAgent(BaseAgent):
     name = "lab"
@@ -27,10 +32,11 @@ class LabAgent(BaseAgent):
 
     async def stream_feedback(
         self,
-        scenario: LabTelemetryScenario,
+        digest: TraceDigest | None,
         learner_context: LearnerContext | None = None,
     ) -> AsyncIterator[str]:
-        timeline = summarize_scenario(scenario)
+        """Feedback grounded in the learner's live trace digest."""
+        timeline = render_digest_timeline(digest)
         async for token in self.stream_reply(
             history=None, user_input=timeline, learner_context=learner_context
         ):

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

@@ -1,33 +1,41 @@
-"""ProctorAgent — integrity signals + coaching interventions (REQ-2-009).
+"""ProctorAgent — integrity signals + coaching over REAL inputs (REQ-3-007).
 
-Consumes a ProctorScenario from the corpus, returns pydantic-validated
-signal classifications via structured_reply (4-layer defense).
-Mock engine inputs; real identity/attention signals are v0.3+.
+v0.3 re-grounding: consumes the learner's live trace digest (idle gaps,
+command cadence), the DefenseStore integrity signals (long pauses from the
+oral defense), and the variant seed cross-check — NOT v0.2 corpus
+scenarios. The proctor COACHES: it classifies signals supportively and
+recommends one intervention; it never punishes and never accuses.
+
+Integrity inputs (computed server-side, passed in by the API layer):
+- trace digest: idle_gap_count/total, command_categories histogram,
+  error/fix cycles, huge-burst indicators (edit_count vs test runs)
+- defense signals: long_pauses list from the finished defense (A-109)
+- variant: seed + params when the task is variant-derived (off-template
+  work is a cross-check input, not an accusation)
 """
 
 from pydantic import BaseModel, Field
 
 from ..corpus.learner_context import LearnerContext, get_learner_context
-from ..corpus.telemetry import ProctorScenario, summarize_proctor_scenario
+from ..grading.features import TraceDigest
 from ..prompts.proctor import SYSTEM_PROMPT, render_context
 from .base import BaseAgent
 
 
 class IntegritySignal(BaseModel):
-    signal_type: str  # e.g. "context_switch" | "idle_gap" | "large_paste"
+    signal_type: str  # "idle_gap" | "long_pause" | "burst_edit" | "off_template"
     severity: str  # "low" | "medium" | "high"
     note: str
 
 
 class ProctorAssessment(BaseModel):
-    scenario_id: str
     signals: list[IntegritySignal] = Field(min_length=0)
     intervention: str  # ONE supportive coaching recommendation
     summary: str
 
 
 PROCTOR_ASSESSMENT_SCHEMA_HINT = (
-    '{"scenario_id": "<id>", "signals": [{"signal_type": "<type>", '
+    '{"signals": [{"signal_type": "<type>", '
     '"severity": "low"|"medium"|"high", "note": "<one sentence>"}], '
     '"intervention": "<one supportive recommendation>", '
     '"summary": "<one sentence>"}'
@@ -43,13 +51,27 @@ class ProctorAgent(BaseAgent):
 
     async def assess(
         self,
-        scenario: ProctorScenario,
+        digest: TraceDigest | None,
+        defense_signals: dict | None = None,
+        variant_context: dict | None = None,
         learner_context: LearnerContext | None = None,
     ) -> ProctorAssessment:
-        timeline = summarize_proctor_scenario(scenario)
+        """Classify REAL integrity inputs into supportive signals + coaching."""
+        parts: list[str] = []
+        if digest is not None:
+            parts.append(f"Build-session digest:\n{digest.model_dump_json()}")
+        else:
+            parts.append("No build telemetry recorded for this task yet.")
+        if defense_signals:
+            parts.append(f"Oral-defense integrity signals:\n{defense_signals}")
+        if variant_context:
+            parts.append(f"Variant audit context (seed + params):\n{variant_context}")
         assessment: ProctorAssessment = await self.structured_reply(
             history=None,
-            user_input=timeline,
+            user_input=(
+                "Assess this learner's integrity signals supportively.\n\n"
+                + "\n\n".join(parts)
+            ),
             learner_context=learner_context,
             schema=ProctorAssessment,
             schema_hint=PROCTOR_ASSESSMENT_SCHEMA_HINT,

apps/ai-service/ai_service/api/assessment.py

@@ -59,11 +59,9 @@ from typing import Any
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel, Field
 
-from ..agents.assessor import RubricScore
 from ..agents.registry import AgentRegistry
 from ..agents.structured import StructuredOutputError
 from ..config import Settings
-from ..corpus.artifacts import get_artifact_bundle, get_transcript_for_artifact
 from ..corpus.learner_context import get_learner_context
 from ..grading.engine import GradingEngine
 from ..grading.store import GradeRecord, GradeStore
@@ -81,37 +79,49 @@ router = APIRouter(prefix="/v1")
 # --- v0.2 artifact evaluation (REQ-2-008) --------------------------------------
 
 
-class AssessmentRequest(BaseModel):
-    artifact_id: str = Field(min_length=1)
-    learner_id: str | None = None
+class EvaluateRequest(BaseModel):
+    learner_id: str = Field(min_length=1)
+    task_id: str = Field(min_length=1)
 
 
-@router.post("/assessment/evaluate", response_model=RubricScore)
+@router.post("/assessment/evaluate")
 async def assessment_evaluate(
-    body: AssessmentRequest,
+    body: EvaluateRequest,
     registry: AgentRegistry = Depends(get_agent_registry),
     settings: Settings = Depends(get_settings),
     provider=Depends(get_provider),
-) -> RubricScore:
-    bundle = get_artifact_bundle(body.artifact_id)
-    if bundle is None:
+    grade_store=Depends(get_grade_store),
+) -> dict:
+    """Assessor coaching rendered FROM the learner's stored grade (REQ-3-007).
+
+    The grading engine computes the scores (POST /assessment/grade); this
+    endpoint explains them. No stored grade yet -> 404 (grade first).
+    """
+    grade = grade_store.get(body.learner_id, body.task_id)
+    if grade is None:
         raise HTTPException(
-            status_code=404, detail=f"unknown artifact {body.artifact_id!r}"
+            status_code=404,
+            detail=f"no stored grade for {body.learner_id}/{body.task_id} - grade first",
         )
-    artifact, rubric = bundle
-    transcript = get_transcript_for_artifact(artifact.artifact_id)
     agent = registry.get(provider, settings, "assessor")
     learner_context = get_learner_context(body.learner_id)
     try:
-        return await agent.evaluate(artifact, rubric, transcript, learner_context)
+        coaching = await agent.coach_grade(grade, learner_context)
     except Exception as exc:
         raise HTTPException(
             status_code=502,
             detail=f"assessment evaluation failed: {exc}",
         ) from exc
+    return {
+        "learner_id": grade.learner_id,
+        "task_id": grade.task_id,
+        "grade_verdict": grade.verdict,
+        "grade_scores": grade.scores,
+        "coaching": coaching.model_dump(),
+    }
 
 
-# --- v0.3 trace grading (REQ-3-004) ---------------------------------------------
+# --- # --- v0.3 trace grading (REQ-3-004) ---------------------------------------------
 
 
 class GradeRequest(BaseModel):

apps/ai-service/ai_service/api/lab.py

@@ -1,27 +1,36 @@
-"""POST /v1/lab/feedback — SSE stream of Lab in-flow feedback (REQ-2-007).
+"""POST /v1/lab/feedback — SSE stream of Lab in-flow feedback (REQ-3-007).
 
-D-016 envelope with agent=lab. Unknown scenario → 404 before streaming.
+v0.3 re-grounding: LIVE trace digest. Request carries {learner_id, task_id};
+the digest is computed from the learner's real TraceStore events (D-028)
+and handed to the Lab agent. No corpus scenarios. Empty/unknown trace is NOT
+an error — Lab gets a "no telemetry yet" timeline and coaches the baseline.
+D-016 envelope with agent=lab.
 """
 
 import json
 from collections.abc import AsyncIterator
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends
 from pydantic import BaseModel, Field
 from sse_starlette.sse import EventSourceResponse
 
 from ..agents.registry import AgentRegistry
 from ..config import Settings
 from ..corpus.learner_context import get_learner_context
-from ..corpus.telemetry import get_lab_scenario
-from .deps import get_agent_registry, get_provider, get_settings
+from ..grading.features import compute_digest
+from .deps import (
+    get_agent_registry,
+    get_provider,
+    get_settings,
+    get_trace_store,
+)
 
 router = APIRouter(prefix="/v1")
 
 
 class LabFeedbackRequest(BaseModel):
-    scenario_id: str = Field(min_length=1)
-    learner_id: str | None = None
+    learner_id: str = Field(min_length=1)
+    task_id: str = Field(min_length=1)
 
 
 @router.post("/lab/feedback")
@@ -30,25 +39,27 @@ async def lab_feedback(
     registry: AgentRegistry = Depends(get_agent_registry),
     settings: Settings = Depends(get_settings),
     provider=Depends(get_provider),
+    trace_store=Depends(get_trace_store),
 ) -> EventSourceResponse:
-    scenario = get_lab_scenario(body.scenario_id)
-    if scenario is None:
-        raise HTTPException(
-            status_code=404, detail=f"unknown scenario {body.scenario_id!r}"
-        )
     agent = registry.get(provider, settings, "lab")
     learner_context = get_learner_context(body.learner_id)
+    trace = (
+        trace_store.get_trace(body.learner_id, body.task_id)
+        if body.task_id in trace_store.list_tasks(body.learner_id)
+        else []
+    )
+    digest = compute_digest(trace) if trace else None
 
     async def event_stream() -> AsyncIterator[dict]:
         yield {"event": "message", "data": json.dumps({
             "type": "meta",
             "agent": "lab",
-            "scenario_id": body.scenario_id,
+            "task_id": body.task_id,
             "model": settings.model,
         })}
         first_byte = True
         try:
-            async for token in agent.stream_feedback(scenario, learner_context):
+            async for token in agent.stream_feedback(digest, learner_context):
                 first_byte = False
                 yield {"event": "message", "data": json.dumps({
                     "type": "delta", "content": token

apps/ai-service/ai_service/api/proctor.py

@@ -1,7 +1,8 @@
-"""POST /v1/proctor/signals — structured integrity signals (REQ-2-009).
+"""POST /v1/proctor/signals — integrity signals over REAL inputs (REQ-3-007).
 
-JSON response (not SSE): a pydantic-validated ProctorAssessment.
-Unknown scenario → 404. Coaching-shaped interventions only.
+v0.3 re-grounding: live trace digest + DefenseStore long-pause signals +
+variant seed cross-check, no corpus scenarios. The proctor coaches:
+a pydantic-validated ProctorAssessment (JSON response, not SSE).
 """
 
 from fastapi import APIRouter, Depends, HTTPException
@@ -11,15 +12,22 @@ from ..agents.proctor import ProctorAssessment
 from ..agents.registry import AgentRegistry
 from ..config import Settings
 from ..corpus.learner_context import get_learner_context
-from ..corpus.telemetry import get_proctor_scenario
-from .deps import get_agent_registry, get_provider, get_settings
+from ..grading.features import compute_digest
+from .deps import (
+    get_agent_registry,
+    get_provider,
+    get_settings,
+    get_trace_store,
+    get_variant_store,
+    get_voice_store,
+)
 
 router = APIRouter(prefix="/v1")
 
 
 class ProctorRequest(BaseModel):
-    scenario_id: str = Field(min_length=1)
-    learner_id: str | None = None
+    learner_id: str = Field(min_length=1)
+    task_id: str = Field(min_length=1)
 
 
 @router.post("/proctor/signals", response_model=ProctorAssessment)
@@ -28,16 +36,40 @@ async def proctor_signals(
     registry: AgentRegistry = Depends(get_agent_registry),
     settings: Settings = Depends(get_settings),
     provider=Depends(get_provider),
+    trace_store=Depends(get_trace_store),
+    variant_store=Depends(get_variant_store),
+    voice_store=Depends(get_voice_store),
 ) -> ProctorAssessment:
-    scenario = get_proctor_scenario(body.scenario_id)
-    if scenario is None:
-        raise HTTPException(
-            status_code=404, detail=f"unknown scenario {body.scenario_id!r}"
-        )
+    """Real integrity inputs: live digest + defense signals + variant context."""
     agent = registry.get(provider, settings, "proctor")
     learner_context = get_learner_context(body.learner_id)
+
+    trace = (
+        trace_store.get_trace(body.learner_id, body.task_id)
+        if body.task_id in trace_store.list_tasks(body.learner_id)
+        else []
+    )
+    digest = compute_digest(trace) if trace else None
+
+    defense_signals = None
+    for record in voice_store.list_for_learner(body.learner_id):
+        if record.task_id == body.task_id and record.status == "finished":
+            defense_signals = record.integrity_signals or None
+            break
+
+    variant = variant_store.get_by_task(body.task_id)
+    variant_context = (
+        {"template_id": variant.template_id, "seed": variant.seed, "params": variant.params}
+        if variant is not None
+        else None
+    )
     try:
-        return await agent.assess(scenario, learner_context)
+        return await agent.assess(
+            digest,
+            defense_signals=defense_signals,
+            variant_context=variant_context,
+            learner_context=learner_context,
+        )
     except Exception as exc:
         raise HTTPException(
             status_code=502, detail=f"proctor assessment failed: {exc}"

apps/ai-service/ai_service/corpus/artifacts.py

@@ -1,5 +1,10 @@
 """Pre-baked artifacts + rubrics + defense transcripts — Assessor mock inputs (REQ-2-008).
 
+v0.2 mock engine inputs (pre-baked artifacts/rubrics/transcripts) — DORMANT as of v0.3 re-
+grounding (Task 6-1-04): no production code path imports this module. Retained as Phase-3
+calibration history.
+
+
 Counterpart: packages/mock-data/ai-scenarios.ts (artifact IDs string-identical,
 D-021). Real process-trace grading is a v0.3+ engine (assessment engine);
 these pre-baked submissions stand in for artifact + defense evaluation.

apps/ai-service/ai_service/corpus/telemetry.py

@@ -1,5 +1,10 @@
 """Simulated sandbox telemetry corpus — Lab agent mock engine inputs (REQ-2-007).
 
+v0.2 mock engine inputs (Lab/Proctor scenarios) — DORMANT as of v0.3 re-grounding (Task 6-1-04):
+no production code path imports this module. Retained as Phase-3 calibration history
+(corpus/trace_fixtures.py references it from TESTS only).
+
+
 Counterpart: packages/mock-data/ai-scenarios.ts (scenario IDs string-identical,
 D-021). Real sandbox telemetry is a v0.3+ engine (sandbox fabric); these
 scripted event streams stand in for the build-session process trace.

apps/ai-service/ai_service/main.py

@@ -50,7 +50,11 @@ def create_app(settings: Settings | None = None) -> FastAPI:
         timeout = httpx.Timeout(connect=10.0, read=300.0, write=30.0, pool=10.0)
         app.state.http_client = httpx.AsyncClient(timeout=timeout)
         app.state.settings = settings
-        app.state.provider = create_provider(settings, app.state.http_client)
+        # State-injection override (same pattern as the stores): tests may
+        # pre-set app.state.provider with a scripted mock; only construct the
+        # configured provider when none is present.
+        if getattr(app.state, "provider", None) is None:
+            app.state.provider = create_provider(settings, app.state.http_client)
         app.state.session_store = InMemorySessionStore()
         app.state.agent_registry = AgentRegistry()
         register_builtin_agents(app.state.agent_registry)

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

@@ -1,24 +1,21 @@
-"""Assessor agent prompt — rubric application to artifacts + defenses (REQ-2-008).
+"""Assessor agent prompt — rubric coaching over REAL grades (REQ-3-007).
 
-Final persona (Phase 4). Assessor is a rigorous, fair grader: scores each
-criterion with evidence, cites what the learner did, returns ONLY valid
-JSON matching the rubric schema.
-Version: assessor-v2 (final for v0.2).
+v0.3 re-grounding: the grading engine (Phase 3) computes the rubric scores
+from the process trace; Assessor EXPLAINS the stored grade as coaching —
+it never invents or re-scores. Rigorous, fair, actionable.
+Version: assessor-v3 (v0.3 live).
 """
 
 SYSTEM_PROMPT = """You are Assessor, the grading agent of Nextcraft, an AI-native competency school.
 Learner: {learner_name}.
 
-You receive: (a) an artifact evidence excerpt, (b) its defense transcript,
-and (c) the rubric for the competency. Your job:
-- Score EVERY rubric criterion from 0-100, justified by evidence you can
-  point to in the artifact or transcript.
-- Cite what the learner did ("the 3-retry loop in the tool node"), not
-  what they should have done — except in gaps, where the missed work goes.
-- Strengths: the two strongest evidence points, each one sentence.
-- Gaps: the two most important missed opportunities, each one sentence.
-- Verdict: "mastered" | "developing" | "not_yet" — judged against the
-  rubric weights, honestly.
+You receive the learner's STORED process-trace grade (verdict, per-criterion
+scores, and the build digest) computed by the grading engine. Your job:
+- Explain what the grade means in plain language (summary).
+- Strengths: cite what the digest + scores show the learner did well.
+- Gaps: name the missed opportunities the scores point to.
+- Next steps: concrete, buildable actions that would move the weakest
+  criterion up one level.
 
 Rules:
 - Rigorous but fair. A polished artifact with a weak defense is NOT mastery.

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

@@ -1,9 +1,11 @@
-"""Lab agent prompt — in-flow feedback over sandbox telemetry (REQ-2-007).
+"""Lab agent prompt — in-flow feedback over LIVE telemetry (REQ-3-007).
 
-Final persona (Phase 4). Lab is a pragmatic build partner: reads the
-telemetry timeline, names the one most useful adjustment, gives one
-concrete next step. Scenario-driven; no session chat.
-Version: lab-v2 (final for v0.2).
+v0.3 re-grounding: the timeline is the learner's real TraceDigest (D-028
+compact counters — commands, test outcomes, idle gaps, edit cadence), not
+v0.2 corpus scenarios. Lab is a pragmatic build partner: reads the live
+digest, names the one most useful adjustment, gives one concrete next
+step. No session chat.
+Version: lab-v3 (v0.3 live).
 """
 
 SYSTEM_PROMPT = """You are Lab, the in-flow feedback agent watching a learner
@@ -25,7 +27,17 @@ Rules:
   self-check that would prove understanding.
 - Three short paragraphs maximum. No headers, no bullet lists."""
 
-PROMPT_VERSION = "lab-v2"
+PROMPT_VERSION = "lab-v3"
+
+
+def render_digest_timeline(digest) -> str:
+    """Live-trace timeline: the compact TraceDigest JSON (D-028)."""
+    if digest is None:
+        return (
+            "No telemetry yet for this build session. Ask the learner to run "
+            "the task's starter test to establish a baseline."
+        )
+    return f"Live build-session digest:\n{digest.model_dump_json()}"
 
 
 def render_context(learner_context) -> dict:

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

@@ -1,24 +1,27 @@
-"""Proctor agent prompt — integrity signals with coaching interventions (REQ-2-009).
+"""Proctor agent prompt — integrity signals with coaching interventions (REQ-3-007).
 
-Final persona (Phase 5). Proctor is a supportive observer, never punitive:
-classifies signals, recommends ONE coaching intervention. Assume good
-faith — most signals have innocent explanations.
-Version: proctor-v2 (final for v0.2).
+v0.3 re-grounding: inputs are REAL — the live trace digest (idle gaps,
+command cadence, edit bursts), the oral-defense integrity signals (long
+pauses), and the variant audit context (seed + params). Proctor is a
+supportive observer, never punitive: classifies signals, recommends ONE
+coaching intervention. Assume good faith.
+Version: proctor-v3 (v0.3 live).
 """
 
 SYSTEM_PROMPT = """You are Proctor, the integrity-support agent of Nextcraft,
 an AI-native competency school.
 Learner: {learner_name}.
 
-You receive a telemetry timeline of defense-session events (tab switches,
-idle gaps, large pastes, focus loss, keystroke bursts). Your job:
-- Classify EACH notable signal: type (e.g. "context_switch", "idle_gap",
-  "large_paste"), severity ("low" | "medium" | "high"), and a one-sentence
-  note citing the event (timestamps and details).
+You receive the learner's REAL build-session digest (idle gaps, command
+categories, edit/test cadence), oral-defense integrity signals (long
+pauses), and — when the task is variant-derived — the variant seed context.
+Your job:
+- Classify EACH notable signal: type ("idle_gap" | "long_pause" |
+  "burst_edit" | "off_template"), severity ("low" | "medium" | "high"),
+  and a one-sentence note citing the numbers.
 - Recommend exactly ONE supportive coaching intervention for the session
   overall — never punitive, never accusatory. Frame around helping the
-  learner succeed, e.g. "offer a short break", "invite them to explain
-  the pasted section in their own words".
+  learner succeed.
 
 Rules:
 - Assume good faith. Tab switches to documentation are normal engineering.
@@ -28,7 +31,7 @@ Rules:
 - Respond with ONLY a valid JSON object matching the provided schema —
   no markdown fences, no prose outside the JSON."""
 
-PROMPT_VERSION = "proctor-v2"
+PROMPT_VERSION = "proctor-v3"
 
 
 def render_context(learner_context) -> dict:

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

@@ -1,103 +1,118 @@
-"""Assessor agent tests — structured rubric scores (REQ-2-008).
+"""Assessor agent tests — live-grade coaching contract (REQ-3-007).
 
-The Assessor is the structured-output showcase: tests use ScriptedJSONProvider
-for valid payloads and exercise the 4-layer defense failure modes.
+v0.3 re-grounding: the Assessor renders coaching FROM the stored grade
+(GradeRecord) — it never invents scores (the grading engine owns that).
 """
 
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+
 import pytest
 
-from ai_service.agents.assessor import AssessorAgent, RubricScore
-from ai_service.agents.structured import StructuredOutputError
+from ai_service.agents.assessor import AssessorAgent, GradeCoaching
 from ai_service.config import Settings
-from ai_service.corpus.artifacts import (
-    get_artifact_bundle,
-    get_transcript_for_artifact,
-    render_rubric,
+from ai_service.grading.store import GradeRecord, SQLiteGradeStore
+from ai_service.llm.mock import MockProvider
+from ai_service.llm.types import Message
+
+COACHING_JSON = json.dumps(
+    {
+        "summary": "Solid iterative build; tests drove the fixes.",
+        "strengths": ["Ran tests after each change."],
+        "gaps": ["Did not cover the empty-input case."],
+        "next_steps": ["Add one edge-case test."],
+    }
 )
-from ai_service.corpus.learner_context import get_learner_context
-from ai_service.llm.mock import MockProvider, ScriptedJSONProvider
-
-VALID_SCORE = {
-    "rubric_id": "rubric-orchestration-c002",
-    "artifact_id": "art-eval-research-assistant",
-    "competency_id": "stack-orchestration-c002",
-    "scores": [
-        {"criterion_id": "rc-architecture", "name": "Agent architecture soundness",
-         "score": 92, "evidence": "Explicit state schema with planner-only write access"},
-        {"criterion_id": "rc-communication", "name": "Inter-agent communication design",
-         "score": 88, "evidence": "Typed ToolMessage responses with retry flags"},
-        {"criterion_id": "rc-reliability", "name": "Reliability engineering",
-         "score": 85, "evidence": "3-retry loop with degradation path"},
-        {"criterion_id": "rc-process", "name": "Process trace quality",
-         "score": 90, "evidence": "Iterative saves with passing test checkpoints"},
-    ],
-    "strengths": ["Clean state boundaries", "Failure-aware tool wrapping"],
-    "gaps": ["No reviewer node yet", "Graph diagram only in README"],
-    "verdict": "mastered",
-}
 
 
-def make_assessor(provider=None) -> AssessorAgent:
-    return AssessorAgent(provider or MockProvider(), Settings(provider="mock"))
+class ScriptedProvider(MockProvider):
+    def __init__(self) -> None:
+        super().__init__()
+        self.requests: list[list[Message]] = []
+        self.replies: list[str] = []
+
+    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 COACHING_JSON
 
 
-def get_bundle(artifact_id="art-eval-research-assistant"):
-    bundle = get_artifact_bundle(artifact_id)
-    assert bundle is not None
-    return bundle
-
-
-async def test_evaluate_returns_validated_rubric_score():
-    provider = ScriptedJSONProvider(VALID_SCORE)
-    assessor = make_assessor(provider)
-    artifact, rubric = get_bundle()
-    transcript = get_transcript_for_artifact(artifact.artifact_id)
-    result = await assessor.evaluate(artifact, rubric, transcript)
-    assert isinstance(result, RubricScore)
-    assert result.verdict == "mastered"
-    assert len(result.scores) == 4
-    assert result.weighted_total(rubric) == pytest.approx(
-        92 * 0.3 + 88 * 0.3 + 85 * 0.25 + 90 * 0.15
+def _grade() -> GradeRecord:
+    return GradeRecord(
+        learner_id="assessor-learner",
+        task_id="assessor-task",
+        variant_seed=None,
+        digest={"error_fix_cycles": 2, "final_test_status": "pass"},
+        scores={
+            "criteria": {
+                "process_quality": 4,
+                "correctness": 3,
+                "debugging_discipline": 4,
+                "test_usage": 3,
+            },
+            "strengths": ["s"],
+            "gaps": ["g"],
+            "verdict": "developing",
+        },
+        verdict="GRADED",
+        model="gemma4:31b",
+        created_at=datetime.now(UTC),
     )
 
 
-async def test_evaluate_rejects_invalid_schema_after_retry():
-    """Plain MockProvider returns non-rubric JSON → 4-layer defense exhausts
-    its single retry and raises StructuredOutputError."""
-    assessor = make_assessor(MockProvider())
-    artifact, rubric = get_bundle()
-    transcript = get_transcript_for_artifact(artifact.artifact_id)
-    with pytest.raises(StructuredOutputError):
-        await assessor.evaluate(artifact, rubric, transcript)
+@pytest.fixture()
+def provider() -> ScriptedProvider:
+    return ScriptedProvider()
 
 
-def test_build_evaluation_input_carries_all_inputs():
-    assessor = make_assessor()
-    artifact, rubric = get_bundle()
-    transcript = get_transcript_for_artifact(artifact.artifact_id)
-    text = assessor.build_evaluation_input(artifact, rubric, transcript)
-    assert artifact.name in text
-    assert artifact.evidence_excerpt in text
-    assert "rc-architecture" in text  # rubric rendered
-    assert "examiner:" in text  # transcript rendered
+@pytest.fixture()
+def agent(provider) -> AssessorAgent:
+    return AssessorAgent(provider, Settings(provider="mock"))
 
 
-def test_build_evaluation_input_without_transcript():
-    assessor = make_assessor()
-    artifact, rubric = get_bundle()
-    text = assessor.build_evaluation_input(artifact, rubric, None)
-    assert artifact.name in text
-    assert "examiner:" not in text
+class TestCoachGrade:
+    async def test_prompt_contains_stored_grade_not_learner_id(self, agent, provider) -> None:
+        await agent.coach_grade(_grade())
+        all_text = "\n".join(
+            m.content for request in provider.requests for m in request
+        )
+        assert "process_quality" in all_text  # stored scores rendered
+        assert "GRADED" in all_text
+        assert "assessor-learner" not in all_text  # D-028 anonymity
+
+    async def test_coaching_validates_via_d020(self, agent, provider) -> None:
+        coaching = await agent.coach_grade(_grade())
+        assert isinstance(coaching, GradeCoaching)
+        assert coaching.summary
+        assert coaching.next_steps
+
+    async def test_malformed_then_good_exercises_retry(self, agent, provider) -> None:
+        provider.replies = ["garbage", COACHING_JSON]
+        coaching = await agent.coach_grade(_grade())
+        assert coaching.summary
+        assert len(provider.requests) == 2
+
+    async def test_no_corpus_artifact_imports(self) -> None:
+        import ast
+        from pathlib import Path
+
+        py = Path(__file__).parents[2] / "ai_service" / "agents" / "assessor.py"
+        tree = ast.parse(py.read_text())
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                assert "corpus.artifacts" not in node.module
+                assert "corpus.telemetry" not in node.module
 
 
-def test_system_prompt_names_assessor_persona():
-    prompt = make_assessor().system_prompt(get_learner_context())
-    assert "Assessor" in prompt
-    assert "ONLY" in prompt  # JSON-only instruction
-
-
-def test_rubric_render_in_prompt_is_complete():
-    """The rubric passed to the model lists every criterion (fair grading)."""
-    artifact, rubric = get_bundle()
-    text = render_rubric(rubric)
-    assert text.count("rc-") == len(rubric.criteria)
+class TestStoreRoundtrip:
+    def test_grade_store_roundtrip(self, tmp_path) -> None:
+        store = SQLiteGradeStore(db_path=tmp_path / "g.db")
+        record = _grade()
+        store.save(record)
+        fetched = store.get("assessor-learner", "assessor-task")
+        assert fetched is not None
+        assert fetched.scores["criteria"]["process_quality"] == 4
+        store.close()

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

@@ -0,0 +1,207 @@
+"""Live re-grounding tests: Lab, Assessor, Proctor on REAL inputs (REQ-3-007)."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+from ai_service.agents.assessor import AssessorAgent, GradeCoaching
+from ai_service.agents.lab import LabAgent
+from ai_service.agents.proctor import ProctorAgent, ProctorAssessment
+from ai_service.config import Settings
+from ai_service.grading.store import GradeRecord, SQLiteGradeStore
+from ai_service.llm.mock import MockProvider
+from ai_service.llm.types import Message
+from ai_service.telemetry.models import TelemetryEvent
+from ai_service.telemetry.store import SQLiteTraceStore
+
+COACHING_JSON = json.dumps(
+    {
+        "summary": "Iterative build with test discipline.",
+        "strengths": ["Tested after changes."],
+        "gaps": ["Missing edge cases."],
+        "next_steps": ["Add an edge-case test."],
+    }
+)
+PROCTOR_JSON = json.dumps(
+    {
+        "signals": [
+            {"signal_type": "idle_gap", "severity": "low", "note": "One 400s gap."}
+        ],
+        "intervention": "Offer a short break.",
+        "summary": "Healthy session overall.",
+    }
+)
+T0 = datetime(2026, 9, 12, tzinfo=UTC)
+
+
+class RecordingProvider(MockProvider):
+    def __init__(self, structured_json: str) -> None:
+        super().__init__()
+        self._structured_json = structured_json
+        self.requests: list[list[Message]] = []
+
+    def _reply_for(self, messages, response_format):
+        self.requests.append([Message(role=m.role, content=m.content) for m in messages])
+        if response_format is not None and response_format.get("type") == "json_object":
+            return self._structured_json
+        return "Coaching feedback referencing your latest test run."
+
+
+def _event(
+    seq: int, kind: str, payload: dict, offset_s: float,
+    learner="live-learner", task="live-task",
+):
+    return TelemetryEvent(
+        learner_id=learner,
+        task_id=task,
+        seq=seq,
+        kind=kind,
+        payload=payload,
+        ts=T0 + timedelta(seconds=offset_s),
+        sandbox_id="sbx-live",
+    )
+
+
+def _seed_trace(store: SQLiteTraceStore) -> None:
+    events = [
+        _event(0, "file_diff", {"path": "a.py"}, 0),
+        _event(1, "command", {"cmd": "pytest -q"}, 10),
+        _event(2, "test_result", {"passed": False, "exit_code": 1}, 15),
+        _event(3, "file_diff", {"path": "a.py"}, 30),
+        _event(4, "test_result", {"passed": True, "exit_code": 0}, 45),
+        _event(5, "activity", {"state": "idle"}, 500),  # >120s gap -> idle
+    ]
+    for e in events:
+        store.append(e)
+
+
+class TestLabLive:
+    async def test_lab_prompt_contains_digest_not_corpus(self, tmp_path) -> None:
+        store = SQLiteTraceStore(db_path=tmp_path / "t.db")
+        _seed_trace(store)
+        provider = RecordingProvider("feedback")
+        agent = LabAgent(provider, Settings(provider="mock"))
+
+        from ai_service.grading.features import compute_digest
+
+        digest = compute_digest(store.get_trace("live-learner", "live-task"))
+        tokens = [t async for t in agent.stream_feedback(digest)]
+        assert tokens
+        all_text = "\n".join(
+            m.content for request in provider.requests for m in request
+        )
+        assert "error_fix_cycles" in all_text
+        assert "lab-scenario" not in all_text  # no corpus fixture ids
+        store.close()
+
+    def test_no_corpus_telemetry_import_in_lab(self) -> None:
+        import ast
+        from pathlib import Path
+
+        py = Path(__file__).parents[2] / "ai_service" / "agents" / "lab.py"
+        tree = ast.parse(py.read_text())
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                assert "corpus.telemetry" not in node.module
+
+
+class TestAssessorLive:
+    async def test_assessor_prompt_contains_stored_scores(self) -> None:
+        provider = RecordingProvider(COACHING_JSON)
+        agent = AssessorAgent(provider, Settings(provider="mock"))
+        grade = GradeRecord(
+            learner_id="live-learner",
+            task_id="live-task",
+            variant_seed=None,
+            digest={"error_fix_cycles": 1},
+            scores={"criteria": {"process_quality": 3}},
+            verdict="GRADED",
+            model="mock",
+            created_at=datetime.now(UTC),
+        )
+        coaching = await agent.coach_grade(grade)
+        assert isinstance(coaching, GradeCoaching)
+        all_text = "\n".join(
+            m.content for request in provider.requests for m in request
+        )
+        assert "process_quality" in all_text
+        assert "live-learner" not in all_text
+
+    def test_no_corpus_artifact_import_in_assessor(self) -> None:
+        import ast
+        from pathlib import Path
+
+        py = Path(__file__).parents[2] / "ai_service" / "agents" / "assessor.py"
+        tree = ast.parse(py.read_text())
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                assert "corpus.artifacts" not in node.module
+                assert "corpus.telemetry" not in node.module
+
+
+class TestProctorLive:
+    async def test_proctor_receives_real_digest_and_defense_signals(self) -> None:
+        provider = RecordingProvider(PROCTOR_JSON)
+        agent = ProctorAgent(provider, Settings(provider="mock"))
+        from ai_service.grading.features import compute_digest
+
+        store = SQLiteTraceStore(db_path=Path(tempfile.mkdtemp()) / "proctor-t.db")
+        _seed_trace(store)
+        digest = compute_digest(store.get_trace("live-learner", "live-task"))
+        store.close()
+
+        assessment = await agent.assess(
+            digest,
+            defense_signals={"long_pauses": [{"turn": 3, "latency_ms": 30000}]},
+            variant_context={"template_id": "tpl-llm-judge", "seed": "cafe", "params": {}},
+        )
+        assert isinstance(assessment, ProctorAssessment)
+        all_text = "\n".join(
+            m.content for request in provider.requests for m in request
+        )
+        assert "idle_gap" in all_text or "idle_gap_count" in all_text
+        assert "long_pauses" in all_text
+        assert "tpl-llm-judge" in all_text
+        assert "live-learner" not in all_text
+
+    def test_no_corpus_scenario_import_in_proctor(self) -> None:
+        import ast
+        from pathlib import Path
+
+        py = Path(__file__).parents[2] / "ai_service" / "agents" / "proctor.py"
+        tree = ast.parse(py.read_text())
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                assert "corpus.telemetry" not in node.module
+
+
+class TestProctorEndpoint:
+    def test_signals_endpoint_serves_real_inputs(self, tmp_path) -> None:
+        from ai_service.main import create_app
+        from ai_service.telemetry.ingest import TraceIntegrityMap
+        from ai_service.variants.store import SQLiteVariantStore
+        from ai_service.voice.defense_store import SQLiteDefenseStore
+
+        app = create_app(Settings(provider="mock"))
+        provider = RecordingProvider(PROCTOR_JSON)
+        app.state.provider = provider
+        app.state.trace_store = SQLiteTraceStore(db_path=tmp_path / "t.db")
+        app.state.grade_store = SQLiteGradeStore(db_path=tmp_path / "g.db")
+        app.state.variant_store = SQLiteVariantStore(db_path=tmp_path / "v.db")
+        app.state.defense_store = SQLiteDefenseStore(db_path=tmp_path / "d.db")
+        app.state.trace_integrity = TraceIntegrityMap()
+        _seed_trace(app.state.trace_store)
+        with TestClient(app) as client:
+            resp = client.post(
+                "/v1/proctor/signals",
+                json={"learner_id": "live-learner", "task_id": "live-task"},
+            )
+            assert resp.status_code == 200, resp.text
+            body = resp.json()
+            assert body["signals"] is not None
+            assert "intervention" in body

apps/ai-service/tests/api/test_assessment.py

@@ -1,79 +1,87 @@
-"""Assessment evaluate endpoint tests — validated JSON, 404s (REQ-2-008)."""
+"""Assessment API tests — stored-grade coaching contract (REQ-3-007)."""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+
+import pytest
+from fastapi.testclient import TestClient
+
+from ai_service.config import Settings
+from ai_service.grading.store import GradeRecord, 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
+
+COACHING_JSON = json.dumps(
+    {
+        "summary": "Good iterative work.",
+        "strengths": ["Tests after changes."],
+        "gaps": ["Missing edge cases."],
+        "next_steps": ["Add an edge-case test."],
+    }
+)
 
 
-from ai_service.agents.assessor import RubricScore
-from ai_service.llm.mock import ScriptedJSONProvider
-
-VALID_SCORE = {
-    "rubric_id": "rubric-orchestration-c002",
-    "artifact_id": "art-eval-research-assistant",
-    "competency_id": "stack-orchestration-c002",
-    "scores": [
-        {"criterion_id": "rc-architecture", "name": "Agent architecture soundness",
-         "score": 92, "evidence": "Explicit state schema"},
-        {"criterion_id": "rc-communication", "name": "Inter-agent communication design",
-         "score": 88, "evidence": "Typed ToolMessage responses"},
-        {"criterion_id": "rc-reliability", "name": "Reliability engineering",
-         "score": 85, "evidence": "3-retry loop"},
-        {"criterion_id": "rc-process", "name": "Process trace quality",
-         "score": 90, "evidence": "Iterative checkpoints"},
-    ],
-    "strengths": ["Clean state boundaries", "Failure-aware tools"],
-    "gaps": ["No reviewer node", "Diagram only in README"],
-    "verdict": "mastered",
-}
+class CoachingMock(MockProvider):
+    def _reply_for(self, messages, response_format):
+        if response_format is not None and response_format.get("type") == "json_object":
+            return COACHING_JSON
+        return super()._reply_for(messages, response_format)
 
 
-def test_evaluate_returns_validated_rubric_json(client):
-    # Swap the app provider for a scripted-JSON provider for this test
-    original = client.app.state.provider
-    client.app.state.provider = ScriptedJSONProvider(VALID_SCORE)
-    try:
-        response = client.post(
-            "/v1/assessment/evaluate",
-            json={"artifact_id": "art-eval-research-assistant"},
+@pytest.fixture()
+def client(tmp_path) -> TestClient:
+    app = create_app(Settings(provider="mock"))
+    app.state.provider = CoachingMock()
+    app.state.trace_store = SQLiteTraceStore(db_path=tmp_path / "t.db")
+    app.state.grade_store = SQLiteGradeStore(db_path=tmp_path / "g.db")
+    app.state.trace_integrity = TraceIntegrityMap()
+    with TestClient(app) as c:
+        yield c
+
+
+def _seed_grade(client: TestClient) -> None:
+
+    app = client.app
+    store: SQLiteGradeStore = app.state.grade_store
+    store.save(
+        GradeRecord(
+            learner_id="api-learner",
+            task_id="api-task",
+            variant_seed=None,
+            digest={"error_fix_cycles": 1},
+            scores={"criteria": {"process_quality": 3}, "verdict": "developing"},
+            verdict="GRADED",
+            model="gemma4:31b",
+            created_at=datetime.now(UTC),
         )
-    finally:
-        client.app.state.provider = original
-    assert response.status_code == 200
-    data = response.json()
-    validated = RubricScore.model_validate(data)  # response contract holds
-    assert validated.verdict == "mastered"
-    assert len(validated.scores) == 4
-
-
-def test_unknown_artifact_404(client):
-    response = client.post("/v1/assessment/evaluate", json={"artifact_id": "ghost"})
-    assert response.status_code == 404
-    assert "ghost" in response.json()["detail"]
-
-
-def test_unparseable_provider_502(client):
-    """Plain MockProvider yields non-rubric JSON → structured defense exhausts
-    retry → endpoint translates to 502 (bad gateway to the model)."""
-    # default mock already returns non-rubric JSON
-    response = client.post(
-        "/v1/assessment/evaluate",
-        json={"artifact_id": "art-eval-research-assistant"},
     )
-    assert response.status_code == 502
-    assert "failed" in response.json()["detail"].lower()
 
 
-def test_missing_artifact_id_422(client):
-    response = client.post("/v1/assessment/evaluate", json={})
-    assert response.status_code == 422
+def test_evaluate_renders_stored_grade_as_coaching(client) -> None:
+    _seed_grade(client)
+    resp = client.post(
+        "/v1/assessment/evaluate",
+        json={"learner_id": "api-learner", "task_id": "api-task"},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["grade_verdict"] == "GRADED"
+    assert body["coaching"]["summary"]
 
 
-def test_second_artifact_also_evaluates(client):
-    original = client.app.state.provider
-    payload = dict(VALID_SCORE, artifact_id="art-eval-rag-dashboard")
-    client.app.state.provider = ScriptedJSONProvider(payload)
-    try:
-        response = client.post(
-            "/v1/assessment/evaluate", json={"artifact_id": "art-eval-rag-dashboard"}
-        )
-    finally:
-        client.app.state.provider = original
-    assert response.status_code == 200
-    assert response.json()["artifact_id"] == "art-eval-rag-dashboard"
+def test_evaluate_without_grade_404(client) -> None:
+    resp = client.post(
+        "/v1/assessment/evaluate",
+        json={"learner_id": "nobody", "task_id": "nothing"},
+    )
+    assert resp.status_code == 404
+    assert "grade first" in resp.json()["detail"]
+
+
+def test_missing_fields_422(client) -> None:
+    resp = client.post("/v1/assessment/evaluate", json={"learner_id": "x"})
+    assert resp.status_code == 422

apps/ai-service/tests/api/test_lab.py

@@ -1,47 +1,86 @@
-"""Lab feedback endpoint tests — SSE envelope with agent=lab (REQ-2-007)."""
+"""Lab endpoint tests — LIVE trace contract (REQ-3-007).
 
-import json
+v0.3 re-grounding: POST /v1/lab/feedback takes {learner_id, task_id}; the
+digest is computed from the learner's real TraceStore events. No corpus
+scenarios; empty trace is a valid "no telemetry yet" coaching path.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+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.models import TelemetryEvent
+from ai_service.telemetry.store import SQLiteTraceStore
+
+T0 = datetime(2026, 9, 12, tzinfo=UTC)
 
 
-def stream_events(client, payload) -> list[dict]:
-    with client.stream("POST", "/v1/lab/feedback", json=payload) as response:
-        assert response.status_code == 200
-        events = []
-        for line in response.iter_lines():
-            if line.startswith("data:"):
-                d = line.removeprefix("data:").strip()
-                if d == "[DONE]":
-                    events.append({"type": "[DONE]"})
-                else:
-                    events.append(json.loads(d))
-        return events
+def _event(seq: int, kind: str, payload: dict, offset_s: float):
+    return TelemetryEvent(
+        learner_id="lab-learner",
+        task_id="lab-task",
+        seq=seq,
+        kind=kind,
+        payload=payload,
+        ts=T0 + timedelta(seconds=offset_s),
+        sandbox_id="sbx-lab",
+    )
 
 
-def test_lab_feedback_streams_full_envelope(client):
-    events = stream_events(client, {"scenario_id": "lab-scenario-strong"})
-    assert events[0]["type"] == "meta"
-    assert events[0]["agent"] == "lab"
-    assert events[0]["scenario_id"] == "lab-scenario-strong"
-    deltas = [e for e in events if e["type"] == "delta"]
-    assert len(deltas) >= 1
-    assert any(e["type"] == "done" for e in events)
-    assert events[-1]["type"] == "[DONE]"
+class StreamingMock(MockProvider):
+    """Deterministic token stream for the SSE path."""
+
+    def _reply_for(self, messages, response_format):
+        return "Feedback grounded in your live session digest."
 
 
-def test_unknown_scenario_404(client):
-    response = client.post("/v1/lab/feedback", json={"scenario_id": "nope"})
-    assert response.status_code == 404
-    assert "nope" in response.json()["detail"]
+@pytest.fixture()
+def client(tmp_path: Path) -> TestClient:
+    app = create_app(Settings(provider="mock"))
+    app.state.provider = StreamingMock()
+    app.state.trace_store = SQLiteTraceStore(db_path=tmp_path / "t.db")
+    app.state.grade_store = SQLiteGradeStore(db_path=tmp_path / "g.db")
+    app.state.trace_integrity = TraceIntegrityMap()
+    with TestClient(app) as c:
+        yield c
 
 
-def test_distinct_scenarios_distinct_replies(client):
-    strong = stream_events(client, {"scenario_id": "lab-scenario-strong"})
-    struggling = stream_events(client, {"scenario_id": "lab-scenario-struggling"})
-    strong_text = "".join(e["content"] for e in strong if e["type"] == "delta")
-    struggling_text = "".join(e["content"] for e in struggling if e["type"] == "delta")
-    assert strong_text != struggling_text
+def test_live_trace_streams_full_envelope(client: TestClient) -> None:
+    store: SQLiteTraceStore = client.app.state.trace_store
+    for e in [
+        _event(0, "file_diff", {"path": "x.py"}, 0),
+        _event(1, "command", {"cmd": "pytest -q"}, 10),
+        _event(2, "test_result", {"passed": False, "exit_code": 1}, 20),
+        _event(3, "test_result", {"passed": True, "exit_code": 0}, 40),
+    ]:
+        store.append(e)
+    with client.stream(
+        "POST", "/v1/lab/feedback", json={"learner_id": "lab-learner", "task_id": "lab-task"}
+    ) as resp:
+        assert resp.status_code == 200
+        body = "".join(chunk.decode() for chunk in resp.iter_raw())
+    assert '"agent": "lab"' in body or '"agent":"lab"' in body
+    assert '"task_id": "lab-task"' in body
+    assert '"type": "delta"' in body or '"type":"delta"' in body
 
 
-def test_missing_scenario_id_422(client):
-    response = client.post("/v1/lab/feedback", json={})
-    assert response.status_code == 422
+def test_empty_trace_coaches_the_baseline(client: TestClient) -> None:
+    """No telemetry is NOT an error — Lab coaches 'run the starter test'."""
+    with client.stream(
+        "POST", "/v1/lab/feedback", json={"learner_id": "lab-learner", "task_id": "no-events"}
+    ) as resp:
+        assert resp.status_code == 200
+
+
+def test_missing_fields_422(client: TestClient) -> None:
+    resp = client.post("/v1/lab/feedback", json={"learner_id": "x"})
+    assert resp.status_code == 422

apps/ai-service/tests/api/test_proctor.py

@@ -1,49 +1,121 @@
-"""Proctor signals endpoint tests — validated JSON, 404s (REQ-2-009)."""
+"""Proctor signals endpoint tests — REAL inputs contract (REQ-3-007).
 
-from ai_service.agents.proctor import ProctorAssessment
-from ai_service.llm.mock import ScriptedJSONProvider
+v0.3 re-grounding: POST /v1/proctor/signals takes {learner_id, task_id}
+and gathers digest + defense signals + variant context server-side.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+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.models import TelemetryEvent
+from ai_service.telemetry.store import SQLiteTraceStore
+from ai_service.variants.store import SQLiteVariantStore
+from ai_service.voice.defense_store import SQLiteDefenseStore
 
 VALID = {
-    "scenario_id": "proctor-scenario-distracted",
     "signals": [
-        {"signal_type": "context_switch", "severity": "low",
-         "note": "Docs tab at t+120s is normal"},
-        {"signal_type": "idle_gap", "severity": "medium",
-         "note": "5-minute idle at t+300s"},
+        {"signal_type": "idle_gap", "severity": "low", "note": "One long pause."},
     ],
     "intervention": "Offer a short break, then restate the plan",
     "summary": "Coaching-shaped session note",
 }
 
-
-def test_signals_returns_validated_json(client):
-    original = client.app.state.provider
-    client.app.state.provider = ScriptedJSONProvider(VALID)
-    try:
-        response = client.post(
-            "/v1/proctor/signals", json={"scenario_id": "proctor-scenario-distracted"}
-        )
-    finally:
-        client.app.state.provider = original
-    assert response.status_code == 200
-    validated = ProctorAssessment.model_validate(response.json())
-    assert validated.scenario_id == "proctor-scenario-distracted"
-    assert validated.intervention
+T0 = datetime(2026, 9, 12, tzinfo=UTC)
 
 
-def test_unknown_scenario_404(client):
-    response = client.post("/v1/proctor/signals", json={"scenario_id": "ghost"})
-    assert response.status_code == 404
+class ProctorJSON(MockProvider):
+    def _reply_for(self, messages, response_format):
+        if response_format is not None and response_format.get("type") == "json_object":
+            return json.dumps(VALID)
+        return super()._reply_for(messages, response_format)
 
 
-def test_unparseable_provider_502(client):
-    response = client.post(
-        "/v1/proctor/signals", json={"scenario_id": "proctor-scenario-healthy"}
+class BrokenJSON(MockProvider):
+    def _reply_for(self, messages, response_format):
+        if response_format is not None and response_format.get("type") == "json_object":
+            return "not json ever"
+        return super()._reply_for(messages, response_format)
+
+
+@pytest.fixture()
+def client(tmp_path: Path) -> TestClient:
+    app = create_app(Settings(provider="mock"))
+    app.state.trace_store = SQLiteTraceStore(db_path=tmp_path / "t.db")
+    app.state.grade_store = SQLiteGradeStore(db_path=tmp_path / "g.db")
+    app.state.variant_store = SQLiteVariantStore(db_path=tmp_path / "v.db")
+    app.state.defense_store = SQLiteDefenseStore(db_path=tmp_path / "d.db")
+    app.state.trace_integrity = TraceIntegrityMap()
+    with TestClient(app) as c:
+        yield c
+
+
+def _seed_trace(client: TestClient) -> None:
+    store: SQLiteTraceStore = client.app.state.trace_store
+    for e in [
+        TelemetryEvent(
+            learner_id="p-learner",
+            task_id="p-task",
+            seq=0,
+            kind="activity",
+            payload={"state": "idle"},
+            ts=T0,
+            sandbox_id="sbx-p",
+        ),
+        TelemetryEvent(
+            learner_id="p-learner",
+            task_id="p-task",
+            seq=1,
+            kind="activity",
+            payload={"state": "idle"},
+            ts=T0 + timedelta(seconds=400),
+            sandbox_id="sbx-p",
+        ),
+    ]:
+        store.append(e)
+
+
+def test_signals_returns_validated_json(client: TestClient) -> None:
+    client.app.state.provider = ProctorJSON()
+    _seed_trace(client)
+    resp = client.post(
+        "/v1/proctor/signals", json={"learner_id": "p-learner", "task_id": "p-task"}
     )
-    assert response.status_code == 502
-    assert "failed" in response.json()["detail"].lower()
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["intervention"]
+    assert body["signals"][0]["signal_type"] == "idle_gap"
 
 
-def test_missing_scenario_id_422(client):
-    response = client.post("/v1/proctor/signals", json={})
-    assert response.status_code == 422
+def test_empty_trace_is_valid_not_404(client: TestClient) -> None:
+    """No telemetry → the proctor still assesses (nothing to flag)."""
+    client.app.state.provider = ProctorJSON()
+    resp = client.post(
+        "/v1/proctor/signals", json={"learner_id": "nobody", "task_id": "nothing"}
+    )
+    assert resp.status_code == 200
+
+
+def test_unparseable_provider_502(client: TestClient) -> None:
+    client.app.state.provider = BrokenJSON()
+    _seed_trace(client)
+    resp = client.post(
+        "/v1/proctor/signals", json={"learner_id": "p-learner", "task_id": "p-task"}
+    )
+    assert resp.status_code == 502
+
+
+def test_missing_fields_422(client: TestClient) -> None:
+    client.app.state.provider = ProctorJSON()
+    resp = client.post("/v1/proctor/signals", json={"learner_id": "x"})
+    assert resp.status_code == 422

apps/ai-service/tests/test_corpus_dormancy.py

@@ -0,0 +1,79 @@
+"""Corpus dormancy verification (Task 6-1-04, REQ-3-007).
+
+The v0.2 mock engine inputs (`corpus/telemetry.py`, `corpus/artifacts.py`)
+must have ZERO production importers after the v0.3 re-grounding: Lab/
+Assessor/Proctor run on real engine inputs with no mock fallback in the
+learner path. The files stay on disk (Phase-3 calibration history) but are
+not imported by any production module. `learner_context` remains ACTIVE
+(agents still need learner context). Test-only references (e.g.
+`corpus/trace_fixtures.py` in grading calibration tests) are allowed.
+"""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+REPO = Path(__file__).parents[1]
+
+#: Production trees whose imports of dormant corpus modules are forbidden.
+PRODUCTION_PATHS = [
+    REPO / "ai_service" / "agents",
+    REPO / "ai_service" / "api",
+    REPO / "ai_service" / "grading",
+    REPO / "ai_service" / "telemetry",
+    REPO / "ai_service" / "variants",
+    REPO / "ai_service" / "voice",
+    REPO / "ai_service" / "sandbox",
+    REPO / "ai_service" / "llm",
+    REPO / "ai_service" / "main.py",
+]
+
+DORMANT_MODULES = ("corpus.telemetry", "corpus.artifacts")
+
+
+def _module_targets(node: ast.AST, *, level: int, module: str | None) -> set[str]:
+    """Resolve relative + absolute import targets to dotted ai_service paths."""
+    targets: set[str] = set()
+    if module and ("corpus" in module):
+        targets.add(module)
+    return targets
+
+
+def test_no_production_imports_of_dormant_corpus() -> None:
+    violators: list[str] = []
+    for path in PRODUCTION_PATHS:
+        files = [path] if path.suffix == ".py" else sorted(path.rglob("*.py"))
+        for py in files:
+            tree = ast.parse(py.read_text())
+            for node in ast.walk(tree):
+                if isinstance(node, ast.ImportFrom):
+                    module = node.module or ""
+                    level = node.level
+                    if level:  # relative: resolve against corpus
+                        if module.endswith("telemetry") and "corpus" in module:
+                            violators.append(f"{py}: {module}")
+                        if module.endswith("artifacts") and "corpus" in module:
+                            violators.append(f"{py}: {module}")
+                        # bare `from . import telemetry` inside corpus/ itself is fine
+                    else:
+                        for target in _module_targets(node, level=level, module=module):
+                            violators.append(f"{py}: {target}")
+                elif isinstance(node, ast.Import):
+                    for alias in node.names:
+                        if any(alias.name.startswith(m) for m in DORMANT_MODULES):
+                            violators.append(f"{py}: {alias.name}")
+    assert not violators, f"dormant corpus imports in production: {violators}"
+
+
+def test_learner_context_stays_active() -> None:
+    """Learner context corpus is NOT dormant — agents still use it."""
+    agents_lab = (REPO / "ai_service" / "agents" / "lab.py").read_text()
+    assert "corpus.learner_context" in agents_lab
+    (REPO / "ai_service" / "corpus" / "learner_context.py").exists()
+
+
+def test_dormancy_headers_present() -> None:
+    for fname in ("telemetry.py", "artifacts.py"):
+        src = (REPO / "ai_service" / "corpus" / fname).read_text()
+        assert "DORMANT" in src, f"{fname} missing dormancy header"