feat(P03): rubric scoring engine + calibration (Wave 2)

Task 3-2-01: prompts/grading.py (grader-v1 rubric, 4 criteria x 0-4 anchors, a-4 churn
advisory) + grading/engine.py — GradingEngine with the BINDING G-4 gate-first ordering
(INCOMPLETE_FLOODED -> gaps -> empty; LLM unreachable for gated traces; first-class
UNGRADABLE_* GradeRecords, model="none" provenance), digest-only prompts (D-028; planted
marker proven absent from all provider messages), D-020 reused via one module-direct
import of agents/structured (grep-auditable). RubricScore validated per-criterion.
Task 3-2-02: corpus/trace_fixtures.py (D-021-aligned archetype IDs) + ordering-contract
calibration test (strong>=lazy on process; strong>struggling on correctness; digest
feature separation asserted deterministically).

54 grading tests green; suite 272 green; ruff clean.

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

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

@@ -0,0 +1,230 @@
+"""Synthetic trace fixtures for grading calibration (Task 3-2-02, REQ-3-004).
+
+Three builder archetypes as REAL `TelemetryEvent` traces (the grading
+engine's native input — these are NOT the v0.2 corpus's simplified
+`{timestamp, kind, detail}` event shapes):
+
+    strong builder     — iterative debugging: small edits, tests early,
+                          failed runs closed by targeted fixes, eventual pass.
+    lazy builder       — one large paste, a single late test run, pass.
+                          (v0.2 alignment: the `lab-scenario-flagged` paste-
+                          and-run archetype; D-021.)
+    struggling builder — many edit/test cycles, failures never close,
+                          never reaches a pass.
+
+ID convention (D-021 alignment, documented in each fixture):
+v0.2 corpus scenario IDs are `<domain>-scenario-<slug>` (`lab-scenario-strong`,
+`lab-scenario-struggling`, `lab-scenario-flagged`, `proctor-scenario-*` — see
+corpus/telemetry.py). Grading fixtures carry ids string-aligned to that
+convention:
+
+    fixture id = "<scenario id>::<archetype>-trace"
+    learner id = "learner-003" (a member of the corpus learner-00N id space;
+                                 learner-001/002 exist in learner_context.py)
+
+so a fixture is greppable against its v0.2 scenario counterpart while staying
+a distinct id space (a grading trace is a real event stream, not the v0.2
+mock scenario timeline — same convention, richer event kind set).
+
+Instance hygiene: fixtures store event SPECS (plain tuples) and materialize
+FRESH `TelemetryEvent` instances on every `fixture.events` access. SQLModel
+rows carry SQLAlchemy instance state once a session has flushed them —
+re-adding the SAME instance to another store is a silent no-op, which would
+poison sequential test runs (a fixture ingested by test N would vanish for
+test N+1). Materializing per access keeps every consumer independent.
+
+These fixtures exist for the CALIBRATION CONTRACT (tests/grading/
+test_calibration.py): the mock provider scripts archetype-mapped scores and
+the test asserts the ORDERING the rubric must eventually enforce. They are
+NOT an LLM quality benchmark — see the test module docstring for the honest
+scope statement.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Final
+
+from pydantic import BaseModel, Field
+
+from ..telemetry.models import TelemetryEvent
+
+_T0: Final = datetime(2026, 9, 12, 0, 0, 0, tzinfo=UTC)
+
+#: Event spec: (seq, kind, payload, seconds-since-session-start).
+EventSpec = tuple[int, str, dict, float]
+
+
+class TraceFixture(BaseModel):
+    """One named synthetic trace + its v0.2 scenario alignment (D-021).
+
+    `event_specs` is the durable, session-state-free description; `events`
+    materializes fresh TelemetryEvent rows from it on every access.
+    """
+
+    model_config = {"frozen": True}
+
+    fixture_id: str  # "<v0.2 scenario id>::<archetype>-trace"
+    archetype: str  # "strong" | "lazy" | "struggling"
+    aligned_scenario_id: str  # the v0.2 corpus scenario this fixture mirrors
+    competency_id: str  # corpus competency id space (stack-orchestration-c00N)
+    task_id: str  # trace task id (grading operates on (learner, task))
+    learner_id: str
+    event_specs: tuple[EventSpec, ...] = Field(default=())
+
+    @property
+    def events(self) -> list[TelemetryEvent]:
+        """FRESH TelemetryEvent instances — safe to ingest into any store.
+
+        Never cache these: an instance flushed by one SQLite session
+        carries persistent identity, and re-appending it elsewhere no-ops.
+        """
+        return [
+            TelemetryEvent(
+                learner_id=self.learner_id,
+                task_id=self.task_id,
+                seq=seq,
+                kind=kind,
+                payload=payload,
+                ts=_T0 + timedelta(seconds=offset_s),
+                sandbox_id="sbx-calibration",
+            )
+            for seq, kind, payload, offset_s in self.event_specs
+        ]
+
+    def description(self) -> str:
+        return (
+            f"{self.fixture_id} (archetype={self.archetype}, aligned="
+            f"{self.aligned_scenario_id}, competency={self.competency_id})"
+        )
+
+
+class _Builder:
+    """Seq-accurate event-spec builder for one (learner, task) pair."""
+
+    def __init__(self) -> None:
+        self.specs: list[EventSpec] = []
+        self._seq = 0
+
+    def add(self, kind: str, payload: dict | None, at: float) -> None:
+        self.specs.append((self._seq, kind, payload or {}, at))
+        self._seq += 1
+
+
+def _strong_builder_specs() -> list[EventSpec]:
+    """Iterative debugging: tests early, tight edit→test loops, eventual pass.
+
+    Mirrors `lab-scenario-strong` (v0.2: keystrokes → file_save → run_tests →
+    test_pass, c002) at full telemetry fidelity — every failed cycle is
+    closed by a targeted edit followed by a re-run that passes.
+    """
+    b = _Builder()
+    b.add("activity", {"state": "starting"}, 0)
+    b.add("file_diff", {"path": "planner.py", "added": 14}, 95)        # small edit
+    b.add("command", {"cmd": "pytest -q tests/test_planner.py"}, 120)  # tests EARLY
+    b.add("test_result", {"passed": True, "exit_code": 0}, 126)
+    b.add("file_diff", {"path": "tool_node.py", "added": 22}, 180)
+    b.add("command", {"cmd": "pytest -q"}, 275)
+    b.add("test_result", {"passed": False, "exit_code": 1}, 281)      # honest failure
+    b.add("file_diff", {"path": "tool_node.py", "added": 6, "removed": 2}, 340)  # targeted fix
+    b.add("command", {"cmd": "pytest -q"}, 430)
+    b.add("test_result", {"passed": True, "exit_code": 0}, 436)       # cycle CLOSED
+    b.add("command", {"cmd": "git commit -m 'tool node with retries'"}, 500)
+    return b.specs
+
+
+def _lazy_builder_specs() -> list[EventSpec]:
+    """Paste-and-run: one large paste, a single LATE test run, instant pass.
+
+    Mirrors `lab-scenario-flagged` (v0.2: paste of 2,400 chars → run_tests →
+    instant 6/6 pass, c003) at full telemetry fidelity — zero iteration, zero
+    verification during construction, one terminal test run only.
+    """
+    b = _Builder()
+    b.add("activity", {"state": "starting"}, 0)
+    b.add("file_diff", {"path": "eval.py", "added": 240, "removed": 0}, 30)  # one bulk paste
+    b.add("file_diff", {"path": "README.md", "added": 12}, 40)
+    b.add("command", {"cmd": "npm run build"}, 45)
+    b.add("run_result", {"exit_code": 0, "ok": True}, 60)
+    b.add("command", {"cmd": "pytest -q"}, 520)                       # single LATE test run
+    b.add("test_result", {"passed": True, "exit_code": 0}, 540)        # instant pass
+    return b.specs
+
+
+def _struggling_builder_specs() -> list[EventSpec]:
+    """Many cycles, none close: repeated failures, no eventual pass.
+
+    Mirrors `lab-scenario-struggling` (v0.2: repeated identical ImportErrors,
+    idle gaps, no checkpoint, c002) at full telemetry fidelity — edits happen
+    between failures, but the same failure recurs; no pass is ever reached.
+    """
+    b = _Builder()
+    b.add("activity", {"state": "starting"}, 0)
+    b.add("file_diff", {"path": "main.py", "added": 40}, 20)
+    b.add("command", {"cmd": "pytest -q"}, 210)
+    b.add("test_result", {"passed": False, "exit_code": 1}, 215)     # ImportError
+    b.add("file_diff", {"path": "main.py", "added": 8, "removed": 3}, 300)
+    b.add("command", {"cmd": "pytest -q"}, 520)
+    b.add("test_result", {"passed": False, "exit_code": 1}, 525)      # SAME error
+    b.add("file_diff", {"path": "main.py", "added": 5}, 610)
+    b.add("command", {"cmd": "pytest -q"}, 960)
+    b.add("test_result", {"passed": False, "exit_code": 1}, 965)     # STILL failing
+    b.add("activity", {"state": "idle"}, 1500)                       # long idle
+    b.add("activity", {"state": "idle"}, 2200)
+    return b.specs
+
+
+#: Calibration learner — a member of the corpus learner-00N id space (D-021;
+#: learner-001/002 live in corpus/learner_context.py; grading fixtures use
+#: a third id so calibration traces never collide with mock-context reads).
+CALIBRATION_LEARNER_ID: Final = "learner-003"
+
+
+STRONG_BUILDER: Final = TraceFixture(
+    fixture_id="lab-scenario-strong::strong-trace",
+    archetype="strong",
+    aligned_scenario_id="lab-scenario-strong",
+    competency_id="stack-orchestration-c002",
+    task_id="task-calibration-strong",
+    learner_id=CALIBRATION_LEARNER_ID,
+    event_specs=tuple(_strong_builder_specs()),
+)
+
+LAZY_BUILDER: Final = TraceFixture(
+    # v0.2's paste-and-run archetype is the "flagged" lab scenario (D-021):
+    # large paste → instant test pass. "lazy builder" is that behavior
+    # without the proctor flag; the alignment is behavioral, documented here.
+    fixture_id="lab-scenario-flagged::lazy-trace",
+    archetype="lazy",
+    aligned_scenario_id="lab-scenario-flagged",
+    competency_id="stack-orchestration-c003",
+    task_id="task-calibration-lazy",
+    learner_id=CALIBRATION_LEARNER_ID,
+    event_specs=tuple(_lazy_builder_specs()),
+)
+
+STRUGGLING_BUILDER: Final = TraceFixture(
+    fixture_id="lab-scenario-struggling::struggling-trace",
+    archetype="struggling",
+    aligned_scenario_id="lab-scenario-struggling",
+    competency_id="stack-orchestration-c002",
+    task_id="task-calibration-struggling",
+    learner_id=CALIBRATION_LEARNER_ID,
+    event_specs=tuple(_struggling_builder_specs()),
+)
+
+TRACE_FIXTURES: Final[dict[str, TraceFixture]] = {
+    f.fixture_id: f
+    for f in (STRONG_BUILDER, LAZY_BUILDER, STRUGGLING_BUILDER)
+}
+
+
+def get_trace_fixture(fixture_id: str) -> TraceFixture | None:
+    return TRACE_FIXTURES.get(fixture_id)
+
+
+def digest_of(fixture: TraceFixture):
+    """Compute the digest for a fixture (pure compute; test-side helper)."""
+    from ..grading.features import compute_digest
+
+    return compute_digest(fixture.events)

apps/ai-service/ai_service/grading/__init__.py

@@ -1,19 +1,26 @@
 """Process-trace grading — trace digest, rubric engine, GradeStore (REQ-3-004).
 
 Boundary rule (D-027): grading/ is an engine module — it never imports
-agents/ or api/ (api/ composes the engine and stores via DI). features.py
+api/; the single sanctioned agents/ dependency is the shared D-020
-imports telemetry/; store.py imports config only.
+structured defense (agents/structured.py), imported module-direct in
+engine.py (see its docstring for why). features.py imports telemetry/;
+store.py imports config only; engine.py composes llm/ + telemetry/ +
+prompts/ + agents.structured.
 
-features.py (TraceDigest, compute_digest) and store.py (GradeRecord,
+Wave status: features.py (TraceDigest, compute_digest) + store.py
-GradeStore, SQLiteGradeStore) are both landed (tasks 3-1-01 + 3-1-02).
+(GradeRecord, GradeStore, SQLiteGradeStore) landed in Wave 1 (3-1-01 +
+3-1-02); engine.py (GradingEngine, RubricScore) is Wave 2 (3-2-01).
 """
 
+from .engine import GradingEngine, RubricScore
 from .features import TraceDigest, compute_digest
 from .store import GradeRecord, GradeStore, SQLiteGradeStore
 
 __all__ = [
     "GradeRecord",
     "GradeStore",
+    "GradingEngine",
+    "RubricScore",
     "SQLiteGradeStore",
     "TraceDigest",
     "compute_digest",

apps/ai-service/ai_service/grading/engine.py

@@ -0,0 +1,268 @@
+"""GradingEngine — rubric scoring over real process traces (REQ-3-004).
+
+The Wave-2 composition of the grading stack:
+
+    trace completeness gate (G-4, FIRST — nothing is sent to any LLM
+    when the gate trips) → TraceStore.get_trace → compute_digest (D-028)
+    → prompts.grading.render_trace_digest → D-020 4-layer structured
+    defense (agents/structured.py — REUSED, composed, never duplicated)
+    → RubricScore validation → GradeStore persistence → GradeRecord.
+
+Gate verdicts are FIRST-CLASS RESULTS, not exceptions (G-4 is binding):
+    UNGRADABLE_TRACE_INCOMPLETE — seq gaps in the store OR the trace is
+        flagged by TraceIntegrityMap (INCOMPLETE_FLOODED). The gap list /
+        flag reason is surfaced in `scores` for API rendering, and the
+        record is PERSISTED like any grade so a learner sees why no
+        credential can be issued for this trace — the gate outcome is
+        durable and auditable, not a transient error string.
+    UNGRADABLE_EMPTY_TRACE      — no events stored for the pair.
+On either verdict `scores.criteria` is empty and the LLM is never called.
+
+DI (D-027/D-032 house pattern): the engine receives trace_store,
+grade_store, integrity and provider through the constructor and knows
+NOTHING of FastAPI — api/ composes it (Task 3-3-01). `model` is injected
+alongside the provider so tests script the mock against the production
+wiring without touching Settings.
+
+Boundary (D-027): grading/ imports llm/ (provider protocol + Message
+type), telemetry/ (store + integrity map), prompts/ (rubric text) and
+ONLY agents.structured — the sanctioned shared D-020 defense. We import
+the MODULE directly (`from ..agents.structured import structured_completion`)
+rather than the `agents` package, mirroring how agents/base.py consumes
+it (same direct-module import): that keeps the dependency surface to
+exactly the two names the engine needs (structured_completion,
+StructuredOutputError) and avoids executing agents/__init__ re-exports
+(BaseAgent, registry, session store) that grading has no business
+loading — a side-effect-hygiene choice that keeps this import line
+grep-auditable as "the one agents dependency". grading/ never imports api/.
+
+RubricScore placement (documented decision): the validated output model
+lives HERE, not in prompts/. The pydantic model is the engine's return
+CONTRACT (the shape GradeStore.scores must hold), while prompts/grading.py
+is pure prompt text + its mirror schema HINT string — the same split as
+agents/assessor.py (model + hint) but with the model owned by the engine
+module that validates it. Prompt files hold text, per the prompts/ house
+style; engine files hold typed contracts.
+"""
+
+import logging
+from datetime import UTC, datetime
+from typing import Final
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from ..agents.structured import StructuredOutputError, structured_completion
+from ..llm.base import LLMProvider
+from ..llm.types import Message
+from ..prompts.grading import (
+    RUBRIC_CRITERIA,
+    RUBRIC_SCORE_SCHEMA_HINT,
+    SYSTEM_PROMPT,
+    render_trace_digest,
+)
+from ..telemetry.ingest import TraceIntegrityMap
+from ..telemetry.store import TraceStore
+from .features import compute_digest
+from .store import GradeRecord, GradeStore
+
+logger = logging.getLogger(__name__)
+
+#: First-class gate verdicts (G-4). GradeRecord.verdict values; non-empty
+#: by store contract. Rubric verdicts (mastered/developing/not_yet) ride in
+#: `scores.verdict` — `record.verdict` stays the machine-readable outcome.
+VERDICT_UNGRADABLE_INCOMPLETE: Final = "UNGRADABLE_TRACE_INCOMPLETE"
+VERDICT_UNGRADABLE_EMPTY: Final = "UNGRADABLE_EMPTY_TRACE"
+#: record.verdict for a successfully LLM-graded trace (the rubric verdict
+#: travels inside scores); keeps verdict non-empty for every persisted row.
+VERDICT_GRADED: Final = "GRADED"
+
+#: Gate-detail keys surfaced in GradeRecord.scores (tests assert on these).
+_INTEGRITY_FLAG_KEY: Final = "integrity_flag"
+_MISSING_SEQS_KEY: Final = "missing_seqs"
+
+
+class RubricScore(BaseModel):
+    """Validated LLM output: per-criterion 0-4 scores + strengths + gaps + verdict.
+
+    The D-020 schema for the grader: `structured_completion` parses the
+    model reply into THIS shape (layer 3), retrying once with the
+    validation error fed back (layer 4). Exact criteria set + 0-4 ranges
+    are enforced here, so the scores dict persisted to GradeStore is
+    always rubric-shaped no matter what the model produced.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    criteria: dict[str, int]
+    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"
+
+    @field_validator("criteria")
+    @classmethod
+    def _criteria_rubric_shaped(cls, value: dict[str, int]) -> dict[str, int]:
+        """Exact criteria keys (no extras, no omissions) and 0-4 scores."""
+        expected = set(RUBRIC_CRITERIA)
+        got = set(value)
+        if got != expected:
+            raise ValueError(
+                f"criteria keys must be exactly {sorted(expected)}, got {sorted(got)}"
+            )
+        for key, score in value.items():
+            if not 0 <= score <= 4:
+                raise ValueError(f"criterion {key!r} must be within 0-4, got {score}")
+        return value
+
+    @field_validator("verdict")
+    @classmethod
+    def _verdict_known(cls, value: str) -> str:
+        allowed = {"mastered", "developing", "not_yet"}
+        if value not in allowed:
+            raise ValueError(f"verdict must be one of {sorted(allowed)}, got {value!r}")
+        return value
+
+
+class GradingEngine:
+    """Scores a (learner_id, task_id) trace into a persisted GradeRecord."""
+
+    def __init__(
+        self,
+        trace_store: TraceStore,
+        grade_store: GradeStore,
+        integrity: TraceIntegrityMap,
+        provider: LLMProvider,
+        *,
+        model: str = "gemma4:31b",
+    ) -> None:
+        self._trace_store = trace_store
+        self._grade_store = grade_store
+        self._integrity = integrity
+        self._provider = provider
+        self._model = model
+
+    async def grade(self, learner_id: str, task_id: str) -> GradeRecord:
+        """Grade one trace; persist latest-state (GradeStore upserts); return it.
+
+        Gate FIRST (G-4): the LLM is only ever reached from the fully
+        guarded path — no gate state can be masked by an LLM error.
+        """
+        record = await self._grade(learner_id, task_id)
+        self._grade_store.save(record)
+        return record
+
+    # ------------------------------------------------------------------ core
+
+    async def _grade(self, learner_id: str, task_id: str) -> GradeRecord:
+        # -- G-4 gate FIRST: integrity flag OR seq gaps. Ordering matters:
+        # gaps() returns [] for an EMPTY trace, so the empty check below is
+        # reachable only when no rows exist at all; a gapped or flooded
+        # trace can never fall through to the LLM path.
+        if self._integrity.is_incomplete(learner_id, task_id):
+            reason = self._integrity.reason(learner_id, task_id) or "unknown"
+            gaps = self._trace_store.gaps(learner_id, task_id)
+            logger.info(
+                "grade gate (G-4): %s/%s integrity-flagged (%s) — ungradable",
+                learner_id,
+                task_id,
+                reason,
+            )
+            return self._ungradable(
+                learner_id,
+                task_id,
+                detail={_INTEGRITY_FLAG_KEY: reason, _MISSING_SEQS_KEY: gaps},
+                verdict=VERDICT_UNGRADABLE_INCOMPLETE,
+            )
+        gaps = self._trace_store.gaps(learner_id, task_id)
+        if gaps:
+            logger.info(
+                "grade gate (G-4): %s/%s seq gaps %s — ungradable",
+                learner_id,
+                task_id,
+                gaps,
+            )
+            return self._ungradable(
+                learner_id,
+                task_id,
+                detail={_INTEGRITY_FLAG_KEY: None, _MISSING_SEQS_KEY: gaps},
+                verdict=VERDICT_UNGRADABLE_INCOMPLETE,
+            )
+
+        trace = self._trace_store.get_trace(learner_id, task_id)
+        if not trace:
+            logger.info(
+                "grade gate: %s/%s empty trace — ungradable", learner_id, task_id
+            )
+            return self._ungradable(
+                learner_id,
+                task_id,
+                detail={_INTEGRITY_FLAG_KEY: None, _MISSING_SEQS_KEY: []},
+                verdict=VERDICT_UNGRADABLE_EMPTY,
+            )
+
+        # -- Guarded path: digest (D-028) → prompt → D-020 4-layer defense.
+        digest = compute_digest(trace)
+        messages = [
+            Message(role="system", content=SYSTEM_PROMPT),
+            Message(role="user", content=render_trace_digest(digest)),
+        ]
+        try:
+            rubric = await structured_completion(
+                self._provider,
+                messages,
+                model=self._model,
+                schema=RubricScore,
+                schema_hint=RUBRIC_SCORE_SCHEMA_HINT,
+            )
+        except StructuredOutputError as exc:
+            # The trace was gradable but the model failed to produce valid
+            # JSON within the D-020 budget (two attempts). Raise — the API
+            # layer maps this to a 502 (assessor precedent). Persisting a
+            # fabricated or partial grade here would violate the no-silent-
+            # fallback rule: no credential-worthy record without a validated
+            # RubricScore.
+            raise StructuredOutputError(f"grading LLM failed for {task_id}: {exc}") from exc
+
+        logger.debug(
+            "graded %s/%s: %s (model=%s)",
+            learner_id,
+            task_id,
+            rubric.verdict,
+            self._model,
+        )
+        return GradeRecord(
+            learner_id=learner_id,
+            task_id=task_id,
+            variant_seed=None,  # null until P4 (D-029)
+            digest=digest.model_dump(),
+            scores=rubric.model_dump(),
+            verdict=VERDICT_GRADED,
+            model=self._model,
+            created_at=datetime.now(tz=UTC),
+        )
+
+    # ------------------------------------------------------------ gate record
+
+    @staticmethod
+    def _ungradable(
+        learner_id: str,
+        task_id: str,
+        *,
+        detail: dict,
+        verdict: str,
+    ) -> GradeRecord:
+        """Build a gate record: no digest (nothing was graded), gate detail
+        surfaced in `scores` (the store allows an empty scores dict, but
+        G-4 requires the gap list / flag reason surfaced — the detail IS the
+        verdict's payload), model="none" (no LLM was involved; provenance
+        stays honest).
+        """
+        return GradeRecord(
+            learner_id=learner_id,
+            task_id=task_id,
+            variant_seed=None,
+            digest={},
+            scores=detail,
+            verdict=verdict,
+            model="none",
+            created_at=datetime.now(tz=UTC),
+        )

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

@@ -0,0 +1,128 @@
+"""Grading rubric prompt — criteria, level anchors, digest render (REQ-3-004).
+
+The grading prompt is deliberately learner-anonymous and trace-bare: the
+model receives ONLY the fixed rubric text and the compact numeric digest
+(TraceDigest JSON, D-028) — never a raw command, file path, payload
+string, learner id, or task id. Everything variable the LLM sees is
+deterministic counters, which both bounds the prompt-injection surface
+and makes "no raw trace reaches the prompt" assert-able in tests (plant
+a distinctive marker in a command payload; assert it absent from every
+message the provider received).
+
+Rubric (four criteria, each scored 0-4 — the ids are the validated
+RubricScore keys enforced by grading/engine.py):
+  process_quality       — iterative building in small, verified steps.
+  correctness          — where the session ended (test/run outcomes).
+  debugging_discipline — how failures were handled.
+  test_usage           — when and how often tests were run.
+
+Advisory a-4 (embedded in the process_quality anchors): high edit/command
+churn with NO test progress is a process-quality NEGATIVE — churn is not
+work. A session with many edits/commands whose test state never moves is
+thrashing, not iterating, and must score low on process quality.
+
+House-style deviation, documented: unlike the tutor prompts, this module
+has no SYSTEM_PROMPT placeholders and no render_context(learner_context)
+— grading is context-free by design (learner anonymity; the digest is the
+only variable input). Runtime imports are TYPE_CHECKING-only so this
+module stays pure text and can never import-cycle with grading/engine.py
+(engine imports this module; if this module imported grading.* at runtime
+while grading/__init__ pulls engine, the package init would deadlock on a
+partially-initialized module).
+
+Version: grader-v1.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Final
+
+if TYPE_CHECKING:  # pragma: no cover - typing only; keeps this module pure text
+    from ..grading.features import TraceDigest
+
+PROMPT_VERSION = "grader-v1"
+
+#: Canonical criterion ids. The engine validates RubricScore criteria keys
+#: against this tuple; the schema hint and anchors below speak the same ids.
+RUBRIC_CRITERIA: Final[tuple[str, ...]] = (
+    "process_quality",
+    "correctness",
+    "debugging_discipline",
+    "test_usage",
+)
+
+#: Sentinel line the engine's user turn is rendered around. Tests (and the
+#: calibration mock) split on it to locate the digest JSON in the prompt.
+DIGEST_MARKER: Final = "PROCESS TRACE DIGEST (JSON):"
+
+SYSTEM_PROMPT = """You are the Grader of Nextcraft, an AI-native competency school.
+You score a learner's build session from a compact numeric digest of their
+process trace. You NEVER see the raw trace — commands, file contents, and
+payloads do not exist on your side; every number you need is in the digest.
+
+Rubric — score each criterion 0-4:
+
+process_quality — iterative building in small, verified steps.
+  4: tight edit→test loops throughout; small verified increments; healthy pacing.
+  3: steady small edits with regular runs; progress mostly verified.
+  2: some iteration, but large unverified leaps or long idle stretches.
+  1: a single bulk change (e.g. one large paste) then a single run; no iteration.
+  0: no meaningful work visible.
+  ADVISORY: high edit/command churn with NO test progress (no runs, no
+  movement in pass counts) is a process-quality NEGATIVE — churn is not
+  work. Cap such a session at 1 on this criterion no matter how many
+  edits or commands were counted.
+
+correctness — where the session ended up.
+  4: final test status pass, with tests passing early and consistently.
+  3: final pass, reached through fail→fix→pass cycles that closed.
+  2: final pass, but preceded by a long unresolved failure streak.
+  1: final fail, but partial passes observed along the way.
+  0: final fail, or no test/run evidence at all.
+
+debugging_discipline — how failures were handled.
+  4: every failure cycle closes; targeted fixes with low mean fix latency.
+  3: most fail→edit→re-run cycles close with a pass.
+  2: failures followed by edits, but cycles rarely close.
+  1: repeated failures with no targeted edits between runs (flailing).
+  0: failures with no fix attempts at all.
+
+test_usage — when and how often tests were run.
+  4: tests run early (small first-pass offset) and throughout the session.
+  3: regular test runs interleaved with edits.
+  2: sparse tests; long stretches of unverified edits.
+  1: a single late test run only.
+  0: no test or run evidence.
+
+Rules:
+- Judge STRICTLY from the digest numbers; cite the fields you used.
+- Strengths: the two strongest digest observations, one sentence each.
+- Gaps: the two most important missed opportunities, one sentence each
+  (a clean session names its next-level improvement instead).
+- Be rigorous but fair: a session that ends green was not necessarily
+  well built, and a struggling session that never passed may still show
+  real debugging discipline.
+- Respond with ONLY a valid JSON object matching the provided schema —
+  no markdown fences, no prose outside the JSON."""
+
+RUBRIC_SCORE_SCHEMA_HINT = (
+    '{"criteria": {"process_quality": <0-4 int>, "correctness": <0-4 int>, '
+    '"debugging_discipline": <0-4 int>, "test_usage": <0-4 int>}, '
+    '"strengths": ["<one sentence>"], "gaps": ["<one sentence>"], '
+    '"verdict": "mastered" | "developing" | "not_yet"}'
+)
+
+
+def render_trace_digest(digest: TraceDigest) -> str:
+    """Render the grader's user turn: a marker line + the digest JSON — nothing else.
+
+    This is the ONLY per-session content that ever reaches the LLM (D-028):
+    the engine composes [system: SYSTEM_PROMPT, user: render_trace_digest(digest)]
+    and the D-020 defense appends its generic schema instruction to this
+    user turn at request time. No learner id, task id, or raw trace material
+    is injected — assert-able by tests.
+    """
+    return (
+        "Score this build session against the rubric.\n"
+        f"{DIGEST_MARKER}\n{digest.model_dump_json()}"
+    )

apps/ai-service/tests/grading/test_calibration.py

@@ -0,0 +1,361 @@
+"""Grading calibration CONTRACT test (Task 3-2-02, REQ-3-004, D-021).
+
+HONEST SCOPE — what this test is and is NOT:
+
+This is a CALIBRATION CONTRACT test, not an LLM quality test. The mock
+provider scripts archetype-mapped rubric scores for each fixture; the
+assertions verify that:
+
+  1. the ENGINE pipeline (gate → digest → prompt → D-020 → validate →
+     persist) runs each archetype end-to-end and preserves the score
+     ordering the scripts impose;
+  2. the fixtures' digest FEATURES actually separate the archetypes
+     deterministically (the non-LLM half of the calibration: paste-and-run
+     vs iterative debugging vs never-passing MUST produce observably
+     different digests — otherwise no rubric could ever separate them);
+  3. fixture IDs are valid corpus-aligned strings (D-021).
+
+It does NOT prove the real LLM scores archetypes in this order — that
+requires live-model evaluation, which is out of scope for v0.3 tests
+(tests NEVER call the cloud; conftest enforces). What the contract pins
+is the ORDERING the rubric + digest must eventually enforce:
+
+    strong.process_quality >= lazy.process_quality
+    strong.correctness     >  struggling.correctness
+
+If a future prompt/digest change makes the scripted mapping unachievable
+by ANY model (e.g. the digest stops separating the archetypes), test 2
+fails — that is the calibration signal.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from ai_service.corpus.telemetry import LAB_SCENARIOS
+from ai_service.corpus.trace_fixtures import (
+    CALIBRATION_LEARNER_ID,
+    LAZY_BUILDER,
+    STRONG_BUILDER,
+    STRUGGLING_BUILDER,
+    TRACE_FIXTURES,
+    digest_of,
+)
+from ai_service.grading.engine import GradingEngine
+from ai_service.grading.store import SQLiteGradeStore
+from ai_service.telemetry.ingest import TraceIntegrityMap
+from ai_service.telemetry.store import SQLiteTraceStore
+
+#: Archetype-mapped rubric scripts (the ORDERING CONTRACT in numbers).
+#: strong: iterative verified building — highest process quality.
+#: lazy: ends pass with zero iteration — the a-4 negative (churn with no
+#:        test progress caps process_quality at 1; here it's paste-churn +
+#:        one late test, so 1). correctness lands high (final pass, fast).
+#: struggling: never passes — lowest correctness.
+ARCHETYPE_SCORES: dict[str, dict] = {
+    "strong": {
+        "criteria": {
+            "process_quality": 4,
+            "correctness": 4,
+            "debugging_discipline": 4,
+            "test_usage": 4,
+        },
+        "strengths": ["tight edit-test loops; every failure cycle closed"],
+        "gaps": ["session ended without a final full-suite run"],
+        "verdict": "mastered",
+    },
+    "lazy": {
+        "criteria": {
+            "process_quality": 1,  # a-4: bulk paste + churn, one late test
+            "correctness": 3,
+            "debugging_discipline": 1,
+            "test_usage": 1,
+        },
+        "strengths": ["final tests pass on the first run"],
+        "gaps": ["one bulk paste with zero verified increments"],
+        "verdict": "developing",
+    },
+    "struggling": {
+        "criteria": {
+            "process_quality": 2,
+            "correctness": 0,  # never passes
+            "debugging_discipline": 1,  # cycles never close
+            "test_usage": 3,
+        },
+        "strengths": ["persisted through repeated failures"],
+        "gaps": ["the same failure recurred three times"],
+        "verdict": "not_yet",
+    },
+}
+
+
+class ArchetypeProvider:
+    """Mock provider scripting archetype-mapped scores keyed by fixture.
+
+    Keying on the digest embedded in the user turn (not on call order)
+    keeps the mapping robust to gate-path noise and asserts the prompt
+    really carries the fixture's digest. Records the last user turn so
+    tests can re-assert D-028 (digest in prompt; raw events out).
+    """
+
+    def __init__(self) -> None:
+        self.digest_to_archetype: dict[str, str] = {}
+        self.calls = 0
+        self.last_user_prompt: str = ""
+
+    def register(self, fixture) -> None:
+        self.digest_to_archetype[digest_of(fixture).model_dump_json()] = (
+            fixture.archetype
+        )
+
+    async def chat(self, messages, *, model, temperature=0.7, response_format=None):
+        self.calls += 1
+        user = next((m for m in reversed(messages) if m.role == "user"), None)
+        assert user is not None
+        self.last_user_prompt = user.content
+        archetype = None
+        for digest_json, arch in self.digest_to_archetype.items():
+            if digest_json in user.content:
+                archetype = arch
+                break
+        assert archetype is not None, "fixture digest not found in prompt"
+        return json.dumps(ARCHETYPE_SCORES[archetype])
+
+    async def stream_chat(self, messages, *, model, temperature=0.7, response_format=None):
+        yield await self.chat(
+            messages, model=model, temperature=temperature, response_format=response_format
+        )
+
+
+@pytest.fixture
+def trace_store(tmp_path: Path) -> SQLiteTraceStore:
+    store = SQLiteTraceStore(db_path=tmp_path / "traces.db")
+    yield store
+    store.close()
+
+
+@pytest.fixture
+def grade_store(tmp_path: Path) -> SQLiteGradeStore:
+    store = SQLiteGradeStore(db_path=tmp_path / "grades.db")
+    yield store
+    store.close()
+
+
+@pytest.fixture
+def provider() -> ArchetypeProvider:
+    p = ArchetypeProvider()
+    for fixture in (STRONG_BUILDER, LAZY_BUILDER, STRUGGLING_BUILDER):
+        p.register(fixture)
+    return p
+
+
+@pytest.fixture
+def engine(trace_store, grade_store, provider) -> GradingEngine:
+    return GradingEngine(
+        trace_store, grade_store, TraceIntegrityMap(), provider, model="gemma4:31b"
+    )
+
+
+async def _grade_fixture(engine, trace_store, fixture) -> dict:
+    for event in fixture.events:
+        trace_store.append(event)
+    record = await engine.grade(fixture.learner_id, fixture.task_id)
+    assert record.verdict == "GRADED", (
+        f"fixture {fixture.fixture_id} did not grade cleanly: {record.verdict}"
+    )
+    return record.scores
+
+
+class TestFixtureAlignment:
+    """D-021: fixture IDs align with the v0.2 corpus scenario IDs."""
+
+    def test_every_fixture_id_is_corpus_scenario_scoped(self):
+        for fixture in TRACE_FIXTURES.values():
+            assert fixture.fixture_id.count("::") == 1
+            scenario_part, trace_part = fixture.fixture_id.split("::")
+            assert scenario_part in LAB_SCENARIOS, (
+                f"{scenario_part!r} is not a v0.2 lab scenario id"
+            )
+            assert trace_part.endswith("-trace")
+            # double-brace-free single slug
+            assert "-" in trace_part
+
+    def test_every_fixture_aligns_to_an_existing_scenario(self):
+        assert set(f.aligned_scenario_id for f in TRACE_FIXTURES.values()) <= set(
+            LAB_SCENARIOS
+        )
+
+    def test_fixture_ids_are_valid_strings(self):
+        for fixture in TRACE_FIXTURES.values():
+            assert fixture.fixture_id and isinstance(fixture.fixture_id, str)
+            assert fixture.learner_id == CALIBRATION_LEARNER_ID
+            assert fixture.learner_id.startswith("learner-00")
+
+    def test_lazy_alignment_is_the_flagged_scenario(self):
+        # v0.2's paste-and-run archetype is the flagged lab scenario (D-021).
+        assert LAZY_BUILDER.aligned_scenario_id == "lab-scenario-flagged"
+        assert LAZY_BUILDER.archetype == "lazy"
+
+    def test_all_three_archetypes_present(self):
+        assert {f.archetype for f in TRACE_FIXTURES.values()} == {
+            "strong",
+            "lazy",
+            "struggling",
+        }
+
+    def test_events_are_valid_telemetry_events(self):
+        from ai_service.telemetry.models import TelemetryEvent
+
+        for fixture in TRACE_FIXTURES.values():
+            assert len(fixture.events) > 0
+            for e in fixture.events:
+                assert isinstance(e, TelemetryEvent)
+                assert e.seq >= 0
+            # contiguous seqs: a grading trace must clear the G-4 gate
+            seqs = [e.seq for e in fixture.events]
+            assert seqs == list(range(len(seqs))), (
+                f"{fixture.fixture_id} has seq gaps — it would trip the gate"
+            )
+
+
+class TestDigestSeparatesArchetypes:
+    """The non-LLM half of calibration: digests differ in the RIGHT direction.
+
+    If these deterministic assertions ever fail, no rubric — however good —
+    could separate the archetypes: the digest would be lossy in the wrong
+    places. These are the features the a-4 advisory and the rubric anchors
+    speak about, computed over REAL trace fixtures.
+    """
+
+    def test_strong_shows_iterative_cycles_lazy_shows_none(self):
+        strong = digest_of(STRONG_BUILDER)
+        lazy = digest_of(LAZY_BUILDER)
+        assert strong.error_fix_cycles >= 1
+        assert lazy.error_fix_cycles == 0
+
+    def test_strong_tests_early_lazy_tests_late(self):
+        strong = digest_of(STRONG_BUILDER)
+        lazy = digest_of(LAZY_BUILDER)
+        assert strong.first_test_pass_offset_s is not None
+        assert lazy.first_test_pass_offset_s is not None
+        assert strong.first_test_pass_offset_s < lazy.first_test_pass_offset_s
+
+    def test_struggling_never_passes(self):
+        struggling = digest_of(STRUGGLING_BUILDER)
+        assert struggling.final_test_status == "fail"
+        assert struggling.test_pass_count == 0
+        assert struggling.test_fail_count >= 3
+
+    def test_strong_and_struggling_both_end(self):
+        strong = digest_of(STRONG_BUILDER)
+        assert strong.final_test_status == "pass"
+        assert strong.test_fail_count >= 1  # honest failure, then closed
+
+    def test_lazy_churns_without_test_progress(self):
+        """a-4 anchor: bulk edit + command churn with (almost) no test signal."""
+        lazy = digest_of(LAZY_BUILDER)
+        assert lazy.edit_count >= 2
+        assert lazy.test_fail_count == 0
+        assert lazy.command_count >= 2
+        assert lazy.final_test_status == "pass"  # ...and still one late test
+
+
+class TestOrderingContract:
+    """The scripted ORDERING CONTRACT through the full engine pipeline."""
+
+    async def test_strong_at_least_lazy_on_process_quality(
+        self, engine, trace_store
+    ):
+        strong = await _grade_fixture(engine, trace_store, STRONG_BUILDER)
+        lazy = await _grade_fixture(engine, trace_store, LAZY_BUILDER)
+        assert (
+            strong["criteria"]["process_quality"]
+            >= lazy["criteria"]["process_quality"]
+        )
+
+    async def test_strong_above_struggling_on_correctness(self, engine, trace_store):
+        strong = await _grade_fixture(engine, trace_store, STRONG_BUILDER)
+        struggling = await _grade_fixture(engine, trace_store, STRUGGLING_BUILDER)
+        assert strong["criteria"]["correctness"] > struggling["criteria"]["correctness"]
+
+    async def test_full_ordering_across_all_three_archetypes(
+        self, engine, trace_store, grade_store
+    ):
+        scores = {
+            arch: await _grade_fixture(engine, trace_store, f)
+            for arch, f in (
+                ("strong", STRONG_BUILDER),
+                ("lazy", LAZY_BUILDER),
+                ("struggling", STRUGGLING_BUILDER),
+            )
+        }
+        # the required ordering contract (task 3-2-02):
+        assert (
+            scores["strong"]["criteria"]["process_quality"]
+            >= scores["lazy"]["criteria"]["process_quality"]
+        )
+        assert (
+            scores["strong"]["criteria"]["correctness"]
+            > scores["struggling"]["criteria"]["correctness"]
+        )
+        # Deliberate NON-ordering, asserted to document the calibration
+        # intent: struggling > lazy on process quality. Process quality is
+        # not outcome — the struggling builder iterated (edits between
+        # runs, three test runs), the lazy builder only pasted once and
+        # verified once (a-4 caps that at 1). A rubric that scores a
+        # never-passing iterator BELOW a paste-and-runner on process is
+        # miscalibrated on its own anchors.
+        assert (
+            scores["struggling"]["criteria"]["process_quality"]
+            > scores["lazy"]["criteria"]["process_quality"]
+        )
+        # verdicts also ordered: mastered >= developing > not_yet
+        assert scores["strong"]["verdict"] == "mastered"
+        assert scores["lazy"]["verdict"] == "developing"
+        assert scores["struggling"]["verdict"] == "not_yet"
+
+        # all three persisted under the calibration learner, latest-state
+        grades = grade_store.list_for_learner(CALIBRATION_LEARNER_ID)
+        assert len(grades) == 3
+        assert {g.task_id for g in grades} == {
+            "task-calibration-strong",
+            "task-calibration-lazy",
+            "task-calibration-struggling",
+        }
+
+    async def test_prompt_carries_fixture_digest_not_raw_events(
+        self, engine, trace_store, provider
+    ):
+        """D-028 on the calibration path: digest in the prompt, raw events
+        never — the lazy fixture plants distinctive payload strings
+        ("eval.py", the bulk 240-line paste) that must NOT appear."""
+        for event in LAZY_BUILDER.events:
+            trace_store.append(event)
+        await engine.grade(LAZY_BUILDER.learner_id, LAZY_BUILDER.task_id)
+
+        assert provider.calls == 1
+        prompt = provider.last_user_prompt
+        # the fixture's digest reached the LLM...
+        assert "PROCESS TRACE DIGEST (JSON):" in prompt
+        assert '"error_fix_cycles"' in prompt
+        # ...but no raw trace material or identity did (D-028)
+        assert "eval.py" not in prompt
+        assert LAZY_BUILDER.learner_id not in prompt
+        assert LAZY_BUILDER.task_id not in prompt
+
+    async def test_struggling_digests_reach_engine_without_gate(
+        self, engine, trace_store, provider
+    ):
+        for event in STRUGGLING_BUILDER.events:
+            trace_store.append(event)
+        record = await engine.grade(
+            STRUGGLING_BUILDER.learner_id, STRUGGLING_BUILDER.task_id
+        )
+        assert record.verdict == "GRADED"  # honest struggle is gradable
+        assert record.scores["criteria"]["correctness"] == 0
+        # a never-passing session still yields a real verdict — the grade
+        # is low, not missing
+        assert record.scores["verdict"] == "not_yet"

apps/ai-service/tests/grading/test_engine.py

@@ -0,0 +1,449 @@
+"""GradingEngine tests (Task 3-2-01, REQ-3-004, G-4 binding).
+
+Contract under test:
+    - G-4 gate FIRST: seq gaps OR integrity-flagged trace →
+      verdict=UNGRADABLE_TRACE_INCOMPLETE with the gap list surfaced in
+      scores; the LLM is NEVER called on those paths (asserted). Empty
+      trace → UNGRADABLE_EMPTY_TRACE. All are first-class GradeRecords
+      (returned + persisted), not exceptions.
+    - Complete trace: prompt carries the digest JSON but NO raw trace
+      material (D-028) — a distinctive marker planted in a command payload
+      must be absent from every message the provider received.
+    - Malformed LLM JSON → the D-020 retry path recovers (scripted bad
+      first, good second) → graded.
+    - Re-grade overwrites the stored record (GradeStore upsert).
+
+Zero network: everything runs against the mock provider (conftest rule).
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+import pytest
+from pydantic import BaseModel, ValidationError
+
+from ai_service.grading.engine import GradingEngine, RubricScore
+from ai_service.grading.store import SQLiteGradeStore
+from ai_service.llm.types import Message
+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, 1, 0, 0, tzinfo=UTC)
+RAW_MARKER = "SECRET-COMMAND-MARKER-7f3a"
+
+
+class RubricJSON:
+    """Canonical well-formed rubric payload (matches engine.RubricScore)."""
+
+    PAYLOAD: dict = {
+        "criteria": {
+            "process_quality": 3,
+            "correctness": 4,
+            "debugging_discipline": 3,
+            "test_usage": 4,
+        },
+        "strengths": ["tight edit-test loops throughout"],
+        "gaps": ["final commit discipline loose"],
+        "verdict": "mastered",
+    }
+
+
+class RecordingProvider:
+    """Mock LLMProvider: scripted replies + captured requests (no network).
+
+    Subclass-composes ai_service.llm.mock.MockProvider so the conftest
+    cloud-free guard philosophy holds — but records every message list
+    and counts calls, which the stock mocks don't do.
+    """
+
+    def __init__(self) -> None:
+        from ai_service.llm.mock import MockProvider
+
+        self._mock = MockProvider()
+        self.replies: list[str] = []  # consumed left-to-right; last repeats
+        self.requests: list[list[Message]] = []
+        self.calls = 0
+
+    async def chat(self, messages, *, model, temperature=0.7, response_format=None):
+        self.calls += 1
+        self.requests.append([Message(role=m.role, content=m.content) for m in messages])
+        if self.replies:
+            reply = self.replies.pop(0)
+        else:
+            reply = json.dumps(RubricJSON.PAYLOAD)
+        return reply
+
+    async def stream_chat(self, messages, *, model, temperature=0.7, response_format=None):
+        yield await self.chat(
+            messages, model=model, temperature=temperature, response_format=response_format
+        )
+
+
+def _event(
+    seq: int,
+    kind: str,
+    payload: dict | None = None,
+    offset_s: float = 0.0,
+    *,
+    learner: str = "engine-learner",
+    task: str = "engine-task",
+) -> TelemetryEvent:
+    return TelemetryEvent(
+        learner_id=learner,
+        task_id=task,
+        seq=seq,
+        kind=kind,
+        payload=payload or {},
+        ts=T0 + timedelta(seconds=offset_s),
+        sandbox_id="sbx-engine",
+    )
+
+
+def _complete_trace(marker_command: bool = True) -> list[TelemetryEvent]:
+    """A healthy iterative trace; seq 0..N contiguous (no gaps).
+
+    `marker_command` plants a distinctive string inside a command payload
+    — the D-028 assertion target: raw trace material must never reach the
+    provider, so the marker must not appear in any captured request.
+    """
+    if marker_command:
+        cmd_payload = {"cmd": f"echo {RAW_MARKER} && pytest -q"}
+    else:
+        cmd_payload = {"cmd": "pytest -q"}
+    return [
+        _event(0, "activity", {"state": "starting"}, 0.0),
+        _event(1, "file_diff", {"path": "a.py", "added": 12}, 10.0),
+        _event(2, "command", cmd_payload, 20.0),
+        _event(3, "test_result", {"passed": False, "exit_code": 1}, 25.0),
+        _event(4, "file_diff", {"path": "a.py", "added": 4, "removed": 2}, 40.0),
+        _event(5, "command", {"cmd": "pytest -q"}, 60.0),
+        _event(6, "test_result", {"passed": True, "exit_code": 0}, 65.0),
+    ]
+
+
+@pytest.fixture
+def trace_store(tmp_path: Path) -> SQLiteTraceStore:
+    store = SQLiteTraceStore(db_path=tmp_path / "traces.db")
+    yield store
+    store.close()
+
+
+@pytest.fixture
+def grade_store(tmp_path: Path) -> SQLiteGradeStore:
+    store = SQLiteGradeStore(db_path=tmp_path / "grades.db")
+    yield store
+    store.close()
+
+
+@pytest.fixture
+def integrity() -> TraceIntegrityMap:
+    return TraceIntegrityMap()
+
+
+@pytest.fixture
+def provider() -> RecordingProvider:
+    return RecordingProvider()
+
+
+@pytest.fixture
+def engine(trace_store, grade_store, integrity, provider) -> GradingEngine:
+    return GradingEngine(
+        trace_store, grade_store, integrity, provider, model="gemma4:31b"
+    )
+
+
+def _ingest(store: SQLiteTraceStore, events: list[TelemetryEvent]) -> None:
+    for e in events:
+        store.append(e)
+
+
+class TestRubricScoreModel:
+    """The D-020 schema contract — engine-side validation rules."""
+
+    def _rubric(self, **overrides) -> dict:
+        payload = json.loads(json.dumps(RubricJSON.PAYLOAD))  # deep copy
+        payload.update(overrides)
+        return payload
+
+    def test_well_formed_payload_validates(self):
+        score = RubricScore.model_validate(RubricJSON.PAYLOAD)
+        assert score.criteria["process_quality"] == 3
+        assert score.verdict == "mastered"
+
+    def test_unknown_criterion_rejected(self):
+        bad = self._rubric()
+        bad["criteria"]["extra_criterion"] = 2
+        with pytest.raises(ValidationError, match="criteria keys must be exactly"):
+            RubricScore.model_validate(bad)
+
+    def test_missing_criterion_rejected(self):
+        bad = self._rubric()
+        del bad["criteria"]["test_usage"]
+        with pytest.raises(ValidationError, match="criteria keys must be exactly"):
+            RubricScore.model_validate(bad)
+
+    def test_out_of_range_score_rejected(self):
+        bad = self._rubric()
+        bad["criteria"]["process_quality"] = 5
+        with pytest.raises(ValidationError, match="must be within 0-4"):
+            RubricScore.model_validate(bad)
+
+    def test_unknown_verdict_rejected(self):
+        bad = self._rubric()
+        bad["verdict"] = "excellent"
+        with pytest.raises(ValidationError, match="verdict must be one of"):
+            RubricScore.model_validate(bad)
+
+
+class TestCompleteTraceGrading:
+    async def test_graded_returns_validated_rubric_persisted_and_returned(
+        self, engine, trace_store, grade_store, provider
+    ):
+        _ingest(trace_store, _complete_trace())
+
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert record.verdict == "GRADED"
+        assert record.model == "gemma4:31b"
+        assert record.scores == RubricJSON.PAYLOAD
+        assert record.scores["criteria"]["process_quality"] == 3
+        assert record.digest.get("error_fix_cycles") == 1
+        assert record.digest.get("final_test_status") == "pass"
+        # digest persisted for auditability (D-028 reproducible input)
+        assert record.digest.get("event_count") == 7
+
+        stored = grade_store.get("engine-learner", "engine-task")
+        assert stored is not None
+        assert stored.scores == record.scores
+        assert stored.verdict == "GRADED"
+
+    async def test_prompt_contains_digest_but_no_raw_trace(self, engine, trace_store, provider):
+        """D-028: the provider saw the digest JSON but NEVER the raw trace.
+
+        The marker was planted inside a command payload (seq 2); it must be
+        absent from every message of every request, while the digest marker
+        line + a JSON object with the digest's field names must be present.
+        """
+        _ingest(trace_store, _complete_trace(marker_command=True))
+
+        await engine.grade("engine-learner", "engine-task")
+
+        assert provider.calls == 1  # happy path: exactly one LLM call
+        all_messages = [m for req in provider.requests for m in req]
+        prompt_text = "\n".join(m.content for m in all_messages)
+
+        assert RAW_MARKER not in prompt_text, (
+            "raw trace material reached the LLM prompt (D-028 violation)"
+        )
+        # digest presence: marker line + a JSON object carrying digest fields
+        assert "PROCESS TRACE DIGEST (JSON):" in prompt_text
+        assert '"error_fix_cycles"' in prompt_text
+        assert '"final_test_status"' in prompt_text
+        # learner anonymity: identity strings never reach the LLM either
+        assert "engine-learner" not in prompt_text
+        assert "engine-task" not in prompt_text
+
+    async def test_system_prompt_has_rubric_and_a4_advisory(
+        self, engine, trace_store, provider
+    ):
+        _ingest(trace_store, _complete_trace())
+
+        await engine.grade("engine-learner", "engine-task")
+
+        system = provider.requests[0][0]
+        assert system.role == "system"
+        assert "process_quality" in system.content
+        assert "correctness" in system.content
+        assert "debugging_discipline" in system.content
+        assert "test_usage" in system.content
+        # a-4: churn-without-test-progress is a process-quality NEGATIVE
+        assert "churn is not" in system.content
+        # 0-4 level anchors present for each criterion
+        assert "ADVISORY" in system.content
+
+
+class TestGateFirst:
+    """G-4 binding: the gate fires BEFORE any grading work."""
+
+    async def test_gapped_trace_ungradable_llm_not_called(
+        self, engine, trace_store, provider, grade_store
+    ):
+        # seqs 0,1,3 stored — seq 2 missing (mirrors the task spec example)
+        events = _complete_trace()
+        gapped = [e for e in events if e.seq != 2]
+        _ingest(trace_store, gapped)
+
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert record.verdict == "UNGRADABLE_TRACE_INCOMPLETE"
+        assert provider.calls == 0, "LLM was called despite a gapped trace (G-4)"
+        # gap list surfaced in scores
+        assert record.scores["missing_seqs"] == [2]
+        assert record.scores["integrity_flag"] is None
+        # first-class record: persisted, retrievable, empty digest
+        stored = grade_store.get("engine-learner", "engine-task")
+        assert stored is not None
+        assert stored.verdict == "UNGRADABLE_TRACE_INCOMPLETE"
+        assert stored.scores["missing_seqs"] == [2]
+        assert stored.digest == {}
+        assert stored.model == "none"
+
+    async def test_integrity_flagged_trace_ungradable_llm_not_called(
+        self, engine, trace_store, integrity, provider
+    ):
+        _ingest(trace_store, _complete_trace())
+        integrity.mark("engine-learner", "engine-task", "INCOMPLETE_FLOODED")
+
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert record.verdict == "UNGRADABLE_TRACE_INCOMPLETE"
+        assert provider.calls == 0, "LLM was called despite integrity flag (G-4)"
+        assert record.scores["integrity_flag"] == "INCOMPLETE_FLOODED"
+        assert record.scores["missing_seqs"] == []  # complete rows, but untrusted
+
+    async def test_flag_wins_even_with_gaps(self, engine, trace_store, integrity, provider):
+        """Both gate signals set: the flag reason is surfaced (both listed)."""
+        events = _complete_trace()
+        _ingest(trace_store, [e for e in events if e.seq != 2])
+        integrity.mark("engine-learner", "engine-task", "INCOMPLETE_FLOODED")
+
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert record.verdict == "UNGRADABLE_TRACE_INCOMPLETE"
+        assert provider.calls == 0
+        assert record.scores["integrity_flag"] == "INCOMPLETE_FLOODED"
+        assert record.scores["missing_seqs"] == [2]
+
+    async def test_empty_trace_ungradable(self, engine, provider, grade_store):
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert record.verdict == "UNGRADABLE_EMPTY_TRACE"
+        assert provider.calls == 0, "LLM was called for an empty trace"
+        assert record.scores == {"integrity_flag": None, "missing_seqs": []}
+        assert record.digest == {}
+        stored = grade_store.get("engine-learner", "engine-task")
+        assert stored is not None
+        assert stored.verdict == "UNGRADABLE_EMPTY_TRACE"
+
+
+class TestD020RetryPath:
+    async def test_malformed_first_response_recovers_via_retry(
+        self, engine, trace_store, provider
+    ):
+        """First reply invalid (wrong shape, fenced), second valid → graded.
+
+        This exercises the D-020 layer-4 path INSIDE the engine through the
+        real agents/structured.py — the engine does not reimplement it.
+        """
+        _ingest(trace_store, _complete_trace())
+        provider.replies = [
+            '```json\n{"summary": "wrong shape"}\n```',  # layer 3 rejects
+            json.dumps(RubricJSON.PAYLOAD),               # retry recovers
+        ]
+
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert provider.calls == 2  # bounded to exactly one retry (layer 4)
+        assert record.verdict == "GRADED"
+        assert record.scores == RubricJSON.PAYLOAD
+        # the retry feedback must carry the validation error back
+        retry_request = provider.requests[1]
+        retry_text = " ".join(m.content for m in retry_request)
+        assert "previous response was invalid" in retry_text
+
+    async def test_persistently_malformed_raises_after_bounded_retry(
+        self, engine, trace_store, provider
+    ):
+        from ai_service.agents.structured import StructuredOutputError
+
+        _ingest(trace_store, _complete_trace())
+        provider.replies = ["not json at all", "still not json"]
+
+        with pytest.raises(StructuredOutputError):
+            await engine.grade("engine-learner", "engine-task")
+        assert provider.calls == 2  # bounded: never more than one retry
+
+
+class TestRegrade:
+    async def test_regrade_overwrites_stored_record(
+        self, engine, trace_store, grade_store, provider
+    ):
+        _ingest(trace_store, _complete_trace())
+
+        first = await engine.grade("engine-learner", "engine-task")
+        assert first.verdict == "GRADED"
+
+        # second grade: the provider now scripts a DIFFERENT rubric outcome
+        updated = json.loads(json.dumps(RubricJSON.PAYLOAD))
+        updated["criteria"]["process_quality"] = 1
+        updated["verdict"] = "not_yet"
+        provider.replies = [json.dumps(updated)]
+
+        second = await engine.grade("engine-learner", "engine-task")
+
+        assert second.scores == updated
+        assert second.created_at >= first.created_at
+        # upsert, not append: exactly one row, holding the LATEST grade
+        grades = grade_store.list_for_learner("engine-learner")
+        assert len(grades) == 1
+        assert grades[0].scores == updated
+        assert grades[0].verdict == "GRADED"
+
+    async def test_regrade_after_gate_outcome_replaces_it(
+        self, engine, trace_store, grade_store, provider, integrity
+    ):
+        """A later successful regrade wholesale-replaces a gate record —
+        the GradeStore upsert contract applied across verdict kinds."""
+        # first: gapped → gate record persisted
+        events = _complete_trace()
+        _ingest(trace_store, [e for e in events if e.seq != 2])
+        gated = await engine.grade("engine-learner", "engine-task")
+        assert gated.verdict == "UNGRADABLE_TRACE_INCOMPLETE"
+
+        # gap healed (late delivery): same pair regrades cleanly
+        trace_store.append(next(e for e in events if e.seq == 2))
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert record.verdict == "GRADED"
+        stored = grade_store.get("engine-learner", "engine-task")
+        assert stored is not None
+        assert stored.verdict == "GRADED"
+        assert stored.scores == RubricJSON.PAYLOAD
+        assert len(grade_store.list_for_learner("engine-learner")) == 1
+
+
+class TestBoundary:
+    def test_engine_module_imports_no_fastapi(self):
+        """D-027: the engine module must not import fastapi (transitively
+        beyond the sanctioned agents/structured + llm + telemetry + prompts)."""
+        import sys
+
+        import ai_service.grading.engine as engine_mod
+
+        code = open(engine_mod.__file__).read()
+        assert "fastapi" not in code
+        assert "from ..api" not in code and "from ..api." not in code
+        # the one sanctioned agents import is the module-direct structured defense
+        assert "from ..agents.structured import" in code
+        assert "from ..agents import" not in code
+        # sanity: the module really is loaded (no import cycle surprises)
+        assert engine_mod.__name__ in sys.modules
+
+
+class TestModelValidationSmoke:
+    """RubricScore as a plain pydantic model (D-020 layer-3 target type)."""
+
+    def test_rubric_score_is_frozen_shape_for_scores_dict(self):
+        score = RubricScore.model_validate(RubricJSON.PAYLOAD)
+        dumped = score.model_dump()
+        assert set(dumped["criteria"]) == {
+            "process_quality",
+            "correctness",
+            "debugging_discipline",
+            "test_usage",
+        }
+        assert isinstance(score, BaseModel)