fix(P04): ship variant anchors to the grader prompt (MH#4 second clause — verifier P1)

GradingEngine takes an optional VariantStore (constructor DI); when the graded
task_id joins to a stored variant: the template's difficulty anchors render into
the grader user turn ("Expected effort envelope" — same bar for every variant of
the template, a-5) and the variant seed is stamped on the GradeRecord (D-029).
Lifespan reordered: VariantStore builds before the engine and is passed in.
Anchors context carries only template id + anchor numbers — D-028 learner-anonymity
preserved (leak tests keep holding). Plain engine (no store) stays variant-blind;
non-variant tasks grade without the envelope.

3 new tests: variant task -> anchors + seed present in prompt/record;
non-variant task -> no envelope; plain engine -> variant_seed None.
Suite 327 green; ruff clean.

---ci---
phase: 4
milestone: v0.3
status: verify
requirements: {covered: [REQ-3-005], partial: []}
---/ci---

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

@@ -47,7 +47,7 @@ style; engine files hold typed contracts.
 
 import logging
 from datetime import UTC, datetime
-from typing import Final
+from typing import TYPE_CHECKING, Final
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
@@ -65,6 +65,9 @@ from ..telemetry.store import TraceStore
 from .features import compute_digest
 from .store import GradeRecord, GradeStore
 
+if TYPE_CHECKING:  # pragma: no cover - protocol-only import for the optional
+    from ..variants.store import VariantStore  # noqa: TC001 (variant-blind without it)
+
 logger = logging.getLogger(__name__)
 
 #: First-class gate verdicts (G-4). GradeRecord.verdict values; non-empty
@@ -78,6 +81,28 @@ VERDICT_GRADED: Final = "GRADED"
 
 #: Gate-detail keys surfaced in GradeRecord.scores (tests assert on these).
 _INTEGRITY_FLAG_KEY: Final = "integrity_flag"
+
+
+def _anchors_context(variant) -> str:  # noqa: ANN001 - VariantRecord (duck-typed)
+    """Render the variant template's difficulty anchors for the grader prompt.
+
+    Contains only the template id + anchor numbers — no learner-identifying
+    material (D-028 anonymity preserved; the digest-leak tests keep holding).
+    Lazy template import: grading must not import variants/ at module load
+    (variants/prompts import-cycle safety mirrors llm/ rules).
+    """
+    from ..variants.templates import get_template
+
+    template = get_template(variant.template_id)
+    if template is None:
+        return f"template={variant.template_id} (anchors unavailable)"
+    a = template.rubric_anchors
+    return (
+        f"template={template.id}; "
+        f"expected_edit_count_band={list(a.expected_edit_count_band)}; "
+        f"expected_min_test_runs={a.expected_min_test_runs}; "
+        f"expected_error_fix_cycles_band={list(a.expected_error_fix_cycles_band)}"
+    )
 _MISSING_SEQS_KEY: Final = "missing_seqs"
 
 
@@ -133,12 +158,19 @@ class GradingEngine:
         provider: LLMProvider,
         *,
         model: str = "gemma4:31b",
+        variant_store: "VariantStore | None" = None,
     ) -> None:
         self._trace_store = trace_store
         self._grade_store = grade_store
         self._integrity = integrity
         self._provider = provider
         self._model = model
+        # Phase 4 (MH#4): optional variant lookup — when the graded task
+        # derives from a generated variant, its template's difficulty anchors
+        # ship to the grader prompt (same bar for every variant of the
+        # template, a-5) and the variant seed is stamped on the record.
+        # Optional so engine tests stay decoupled; main.py lifespan wires it.
+        self._variant_store = variant_store
 
     async def grade(self, learner_id: str, task_id: str) -> GradeRecord:
         """Grade one trace; persist latest-state (GradeStore upserts); return it.
@@ -201,9 +233,16 @@ class GradingEngine:
 
         # -- Guarded path: digest (D-028) → prompt → D-020 4-layer defense.
         digest = compute_digest(trace)
+        variant = self._lookup_variant(task_id)
+        anchors_context = (
+            _anchors_context(variant) if variant is not None else None
+        )
         messages = [
             Message(role="system", content=SYSTEM_PROMPT),
-            Message(role="user", content=render_trace_digest(digest)),
+            Message(
+                role="user",
+                content=render_trace_digest(digest, anchors_context=anchors_context),
+            ),
         ]
         try:
             rubric = await structured_completion(
@@ -232,7 +271,7 @@ class GradingEngine:
         return GradeRecord(
             learner_id=learner_id,
             task_id=task_id,
-            variant_seed=None,  # null until P4 (D-029)
+            variant_seed=variant.seed if variant is not None else None,  # D-029
             digest=digest.model_dump(),
             scores=rubric.model_dump(),
             verdict=VERDICT_GRADED,
@@ -240,6 +279,12 @@ class GradingEngine:
             created_at=datetime.now(tz=UTC),
         )
 
+    def _lookup_variant(self, task_id: str):  # noqa: ANN202 - VariantRecord | None
+        """MH#4: resolve the graded task's variant (None when not variant-derived)."""
+        if self._variant_store is None:
+            return None
+        return self._variant_store.get_by_task(task_id)
+
     # ------------------------------------------------------------ gate record
 
     @staticmethod

apps/ai-service/ai_service/main.py

@@ -76,6 +76,30 @@ def create_app(settings: Settings | None = None) -> FastAPI:
         if getattr(app.state, "trace_integrity", None) is None:
             app.state.trace_integrity = TraceIntegrityMap()
 
+        # Variant generation (REQ-3-005): VariantStore from the same
+        # SQLite file as traces/grades (D-027), one VariantGenerator singleton
+        # wired through app.state — the generator receives store + provider
+        # via constructor DI and knows nothing of FastAPI (api/ composes it,
+        # same pattern as GradingEngine). Tests may pre-set
+        # app.state.variant_store / app.state.variant_generator (the same
+        # state-injection override); the lifespan adopts a pre-set store but
+        # NEVER rebuilds a pre-set generator (its provider binding is part
+        # of the test fixture).
+        # ORDER NOTE: built BEFORE the grading engine — the engine takes the
+        # variant store (Phase 4 MH#4: variant anchors ship to the grader
+        # prompt; variant_seed stamped on graded records).
+        variant_store = getattr(app.state, "variant_store", None)
+        if variant_store is None:
+            settings.db_path.parent.mkdir(parents=True, exist_ok=True)
+            variant_store = SQLiteVariantStore(db_path=settings.db_path)
+            app.state.variant_store = variant_store
+        if getattr(app.state, "variant_generator", None) is None:
+            app.state.variant_generator = VariantGenerator(
+                variant_store,
+                app.state.provider,
+                model=settings.model,
+            )
+
         # Grading persistence + engine (REQ-3-004): GradeStore from the same
         # SQLite file as traces (D-027), one GradingEngine singleton wired
         # through app.state — the engine receives its stores via constructor
@@ -95,27 +119,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
                 app.state.trace_integrity,
                 app.state.provider,
                 model=settings.model,
-            )
+                variant_store=variant_store,  # MH#4: anchors + seed (D-029)
-
-        # Variant generation (REQ-3-005, Wave 3): VariantStore from the same
-        # SQLite file as traces/grades (D-027), one VariantGenerator singleton
-        # wired through app.state — the generator receives store + provider
-        # via constructor DI and knows nothing of FastAPI (api/ composes it,
-        # same pattern as GradingEngine). Tests may pre-set
-        # app.state.variant_store / app.state.variant_generator (the same
-        # state-injection override); the lifespan adopts a pre-set store but
-        # NEVER rebuilds a pre-set generator (its provider binding is part
-        # of the test fixture).
-        variant_store = getattr(app.state, "variant_store", None)
-        if variant_store is None:
-            settings.db_path.parent.mkdir(parents=True, exist_ok=True)
-            variant_store = SQLiteVariantStore(db_path=settings.db_path)
-            app.state.variant_store = variant_store
-        if getattr(app.state, "variant_generator", None) is None:
-            app.state.variant_generator = VariantGenerator(
-                variant_store,
-                app.state.provider,
-                model=settings.model,
             )
 
         async def _reaper_loop() -> None:

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

@@ -113,7 +113,10 @@ RUBRIC_SCORE_SCHEMA_HINT = (
 )
 
 
-def render_trace_digest(digest: TraceDigest) -> str:
+def render_trace_digest(
+    digest: TraceDigest,
+    anchors_context: str | None = None,
+) -> 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):
@@ -121,8 +124,17 @@ def render_trace_digest(digest: TraceDigest) -> str:
     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.
+
+    `anchors_context` (Phase 4, MH#4): when the graded task derives from a
+    variant, the engine passes the template's difficulty-normalization
+    anchors (the expected effort envelope) so the rubric is applied against
+    the SAME bar for every variant of that template (a-5). It contains only
+    the anchor numbers + the template id — no learner-identifying material.
     """
-    return (
+    base = (
         "Score this build session against the rubric.\n"
         f"{DIGEST_MARKER}\n{digest.model_dump_json()}"
     )
+    if anchors_context:
+        base = f"{base}\n\nExpected effort envelope for this task variant:\n{anchors_context}"
+    return base

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

@@ -31,6 +31,7 @@ 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
+from ai_service.variants.store import SQLiteVariantStore
 
 T0 = datetime(2026, 9, 12, 1, 0, 0, tzinfo=UTC)
 RAW_MARKER = "SECRET-COMMAND-MARKER-7f3a"
@@ -447,3 +448,81 @@ class TestModelValidationSmoke:
             "test_usage",
         }
         assert isinstance(score, BaseModel)
+
+
+class TestVariantAnchorsShipment:
+    """Phase 4 MH#4: variant anchors + seed ship into grading (a-5 same bar)."""
+
+    @pytest.fixture
+    def variant_engine(self, trace_store, grade_store, integrity, provider, tmp_path):
+        store = SQLiteVariantStore(db_path=tmp_path / "variants.db")
+        yield GradingEngine(
+            trace_store,
+            grade_store,
+            integrity,
+            provider,
+            model="gemma4:31b",
+            variant_store=store,
+        ), store
+        store.close()
+
+    @pytest.fixture
+    def plain_engine(self, trace_store, grade_store, integrity, provider):
+        return GradingEngine(
+            trace_store, grade_store, integrity, provider, model="gemma4:31b"
+        )
+
+    async def test_variant_task_grades_with_anchors_and_seed(
+        self, variant_engine, trace_store, provider
+    ):
+        engine, vstore = variant_engine
+        _ingest(trace_store, _complete_trace())
+
+        # A stored variant whose task_id matches the graded trace.
+        from datetime import UTC
+        from datetime import datetime as dt
+
+        from ai_service.variants.store import VariantRecord
+
+        vstore.save(
+            VariantRecord(
+                learner_id="engine-learner",
+                task_id="engine-task",
+                template_id="tpl-llm-judge",
+                seed="cafe" * 16,
+                params={"domain": "tutoring"},
+                statement="Scripted statement long enough to be legal.",
+                starter_files={"README.md": "x"},
+                created_at=dt.now(UTC),
+            )
+        )
+
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert record.verdict == "GRADED"
+        assert record.variant_seed == "cafe" * 16  # D-029 stamped
+        user_msgs = [m.content for req in provider.requests for m in req if m.role == "user"]
+        assert any("Expected effort envelope" in u for u in user_msgs)
+        assert any("tpl-llm-judge" in u for u in user_msgs)
+        assert any("expected_min_test_runs" in u for u in user_msgs)
+
+    async def test_non_variant_task_has_no_anchors(
+        self, variant_engine, trace_store, provider
+    ):
+        engine, _ = variant_engine
+        _ingest(trace_store, _complete_trace())
+
+        record = await engine.grade("engine-learner", "engine-task")
+
+        assert record.verdict == "GRADED"
+        assert record.variant_seed is None
+        user_msgs = [m.content for req in provider.requests for m in req if m.role == "user"]
+        assert not any("Expected effort envelope" in u for u in user_msgs)
+
+    async def test_plain_engine_stays_variant_blind(
+        self, plain_engine, trace_store
+    ):
+        _ingest(trace_store, _complete_trace())
+        record = await plain_engine.grade("engine-learner", "engine-task")
+        assert record.verdict == "GRADED"
+        assert record.variant_seed is None