merge(P04): phase/04 variant generation → milestone/v0.3-credential-engines

---ci---
phase: 4
milestone: v0.3
status: ship
---/ci---

.ciagent/CHECKPOINT.json

@@ -1,8 +1,8 @@
 {
-  "phase": 3,
+  "phase": 4,
-  "stage": "complete",
+  "stage": "verify",
   "milestone": "v0.3",
   "phase_role": "execution",
   "attempts": 0,
-  "updated_at": "2026-09-12T02:49:33Z"
+  "updated_at": "2026-09-12T03:51:30Z"
 }

.ciagent/REQUIREMENTS.md

@@ -15,7 +15,7 @@
 | ID | Description | Priority | Phase | Status |
 |----|-------------|----------|-------|--------|
 | REQ-3-004 | Process-trace grading engine: grade artifacts from their full process traces; rubric-aligned structured scores; feeds Assessor real inputs | critical | 3 | complete |
-| REQ-3-005 | Variant task generation: per-learner task variants (no two learners get identical prompts); variant seed registry; difficulty normalization | high | 4 | pending |
+| REQ-3-005 | Variant task generation: per-learner task variants (no two learners get identical prompts); variant seed registry; difficulty normalization | high | 4 | complete |
 | REQ-3-006 | Oral/voice defense: AI examiner conducts spoken defense (STT → dialogue → TTS); transcript + integrity signals captured; feeds Proctor/Mentor | high | 5 | pending |
 
 ### Agent Re-grounding & Integration
@@ -178,7 +178,7 @@
 | REQ-3-002 | 1 | complete |
 | REQ-3-003 | 2 | complete |
 | REQ-3-004 | 3 | complete |
-| REQ-3-005 | 4 | pending |
+| REQ-3-005 | 4 | complete |
 | REQ-3-006 | 5 | pending |
 | REQ-3-007 | 6 | pending |
 | REQ-3-008 | 6 | pending |

.ciagent/ROADMAP.md

@@ -22,7 +22,7 @@
 | 1 | Sandbox fabric | complete | 0 | REQ-3-001, REQ-3-002 | Isolated per-learner sandbox environments provisioned (IDE / design / simulation); lifecycle API (create/destroy/snapshot); resource limits enforced; no cross-tenant access |
 | 2 | Live build telemetry | complete | 1 | REQ-3-003 | In-environment capture of process events (commands, file diffs, run/test results, keystroke-level activity) streamed to ai-service; reliable transport; per-learner trace persistence |
 | 3 | Process-trace grading engine | complete | 2 | REQ-3-004 | Grades artifacts from their full process traces (not just final output); emits structured rubric-aligned scores; feeds Assessor real inputs |
-| 4 | Variant task generation | pending | 1 | REQ-3-005 | Per-learner task variants generated so no two learners receive identical prompts; variant seed recorded for grading fairness |
+| 4 | Variant task generation | complete | 1 | REQ-3-005 | Per-learner task variants generated so no two learners receive identical prompts; variant seed recorded for grading fairness |
 | 5 | Oral / voice defense | pending | 3 | REQ-3-006 | AI examiner conducts spoken defense of submitted work; STT → dialogue → TTS; transcript + integrity signals captured; feeds Proctor/Mentor |
 | 6 | Agent re-grounding + learner surface integration | pending | 2,3,4,5 | REQ-3-007, REQ-3-008 | Lab/Assessor/Proctor consume real engine inputs; v0.1 sandbox + assessment mockups wired to real engines (in-browser build/run, live telemetry, live defense) |
 | 7 | Final review + ship | pending | 6 | — | Code review clean; audit passes; milestone tagged (v0.2.x final patch); release created on Gitea |

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

@@ -10,6 +10,7 @@ from .mentor import router as mentor_router
 from .proctor import router as proctor_router
 from .sandboxes import router as sandboxes_router
 from .telemetry import router as telemetry_router
+from .variants import router as variants_router
 
 __all__ = [
     "assessment_router",
@@ -19,4 +20,5 @@ __all__ = [
     "proctor_router",
     "sandboxes_router",
     "telemetry_router",
+    "variants_router",
 ]

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

@@ -12,6 +12,8 @@ from ..sandbox.manager import SandboxManager
 from ..sandbox.workdir import SandboxDir
 from ..telemetry.ingest import TraceIntegrityMap
 from ..telemetry.store import TraceStore
+from ..variants.generator import VariantGenerator
+from ..variants.store import VariantStore
 
 
 def get_settings(request: Request) -> Settings:
@@ -53,3 +55,11 @@ def get_grade_store(request: Request) -> GradeStore:
 
 def get_grading_engine(request: Request) -> GradingEngine:
     return request.app.state.grading_engine
+
+
+def get_variant_generator(request: Request) -> VariantGenerator:
+    return request.app.state.variant_generator
+
+
+def get_variant_store(request: Request) -> VariantStore:
+    return request.app.state.variant_store

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

@@ -0,0 +1,187 @@
+"""/v1/variants — seeded per-learner task variant endpoints (REQ-3-005, D-029).
+
+Three faces over the variant engine (the generator + store stay FastAPI-free;
+this module owns all HTTP wiring — the D-027/D-032 house pattern):
+
+    POST /v1/variants {learner_id, template_id | competency_id}
+        → 200 the learner's variant — GENERATED on the first request,
+          CACHED (store read, zero LLM calls) on every repeat: D-029
+          reproducibility means one (learner_id, template_id) is ONE
+          variant forever, so a regenerate is always a 200 of the SAME
+          variant, never a second render.
+        → 404 unknown template_id, or competency_id with no bound template.
+        → 422 neither template_id nor competency_id given.
+    GET  /v1/variants/{task_id}
+        → 200 the stored variant owning the task key (the grading and
+          telemetry join path); 404 when no variant was ever generated
+          for the task.
+    GET  /v1/variants?learner_id=...
+        → 200 the learner's variants, chronological; [] when none.
+
+Template resolution: an explicit `template_id` wins; without it the FIRST
+template bound to `competency_id` is used (`template_for_competency`,
+D-021 corpus alignment). The response carries `competency_id` resolved
+from the template library at read time — an enrichment, not a persisted
+column (the seed re-derives the whole variant, D-029) — so the learner
+surface can bind a variant to its competency without a library round-trip.
+
+Distinctness (REQ-3-005): different learners on the same template draw
+different seeded params and receive distinct statements and task_ids;
+tests/api/test_variants.py asserts this end-to-end through the API.
+"""
+
+from datetime import datetime
+from typing import Self
+
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel, Field, model_validator
+
+from ..variants.generator import VariantGenerator
+from ..variants.store import VariantRecord, VariantStore
+from ..variants.templates import TaskTemplate, get_template, template_for_competency
+from .deps import get_variant_generator, get_variant_store
+
+router = APIRouter(prefix="/v1/variants", tags=["variants"])
+
+
+# -- contracts ------------------------------------------------------------------
+
+
+class VariantGenerateRequest(BaseModel):
+    """One variant identity: an explicit `template_id`, or the first
+    template bound to a `competency_id` (D-021). `template_id` wins when
+    both are given (explicit identity beats derived); at least one is
+    required — 422 otherwise.
+    """
+
+    learner_id: str = Field(min_length=1)
+    template_id: str | None = None
+    competency_id: str | None = None
+
+    @model_validator(mode="after")
+    def _require_template_or_competency(self) -> Self:
+        if self.template_id is None and self.competency_id is None:
+            raise ValueError("template_id or competency_id is required")
+        return self
+
+
+class VariantResponse(BaseModel):
+    """VariantRecord over HTTP, plus the `competency_id` enrichment.
+
+    Every field except `competency_id` mirrors `VariantRecord` exactly
+    (snake_case; `created_at` is an ISO 8601 UTC datetime) — the wire shape
+    typed as `TaskVariant` in packages/types/variants.ts.
+    """
+
+    learner_id: str
+    task_id: str
+    template_id: str
+    competency_id: str
+    seed: str
+    params: dict[str, str | int]
+    statement: str
+    starter_files: dict[str, str]
+    created_at: datetime
+
+
+class VariantListResponse(BaseModel):
+    variants: list[VariantResponse]
+
+
+# -- resolution + rendering -----------------------------------------------------
+
+
+def _resolve_template(body: VariantGenerateRequest) -> TaskTemplate:
+    """Template for the request: the explicit id, else the first template
+    bound to the competency; 404 when neither resolves."""
+    if body.template_id is not None:
+        template = get_template(body.template_id)
+        if template is None:
+            raise HTTPException(
+                status_code=404,
+                detail=f"no task template with id {body.template_id!r}",
+            )
+        return template
+    # The request validator guarantees the disjunction, so reaching here
+    # means a competency_id was given (never None).
+    assert body.competency_id is not None
+    templates = template_for_competency(body.competency_id)
+    if not templates:
+        raise HTTPException(
+            status_code=404,
+            detail=f"no task template for competency {body.competency_id!r}",
+        )
+    return templates[0]
+
+
+def _competency_for(template_id: str) -> str:
+    """competency_id enrichment for stored records (read paths)."""
+    template = get_template(template_id)
+    if template is None:
+        # Integrity guard: a stored variant referencing a template that is
+        # no longer in the library cannot be enriched; fail loudly rather
+        # than fabricate a competency binding.
+        raise HTTPException(
+            status_code=500,
+            detail=f"stored variant references unknown template {template_id!r}",
+        )
+    return template.competency_id
+
+
+def _to_response(record: VariantRecord, competency_id: str) -> VariantResponse:
+    return VariantResponse(
+        learner_id=record.learner_id,
+        task_id=record.task_id,
+        template_id=record.template_id,
+        competency_id=competency_id,
+        seed=record.seed,
+        params=dict(record.params),
+        statement=record.statement,
+        starter_files=dict(record.starter_files),
+        created_at=record.created_at,
+    )
+
+
+# -- endpoints ------------------------------------------------------------------
+
+
+@router.post("", response_model=VariantResponse)
+async def generate_variant(
+    body: VariantGenerateRequest,
+    generator: VariantGenerator = Depends(get_variant_generator),
+) -> VariantResponse:
+    """The learner's variant for the resolved template — generated on the
+    first request, cached (no LLM call) on every repeat: D-029 makes a
+    regenerate a 200 of the SAME stored variant.
+    """
+    template = _resolve_template(body)
+    record = await generator.generate(body.learner_id, template.id)
+    return _to_response(record, competency_id=template.competency_id)
+
+
+@router.get("", response_model=VariantListResponse)
+async def list_variants(
+    learner_id: str,
+    store: VariantStore = Depends(get_variant_store),
+) -> VariantListResponse:
+    """All stored variants for the learner, chronological; [] when none."""
+    variants = [
+        _to_response(record, competency_id=_competency_for(record.template_id))
+        for record in store.list_for_learner(learner_id)
+    ]
+    return VariantListResponse(variants=variants)
+
+
+@router.get("/{task_id}", response_model=VariantResponse)
+async def get_variant(
+    task_id: str,
+    store: VariantStore = Depends(get_variant_store),
+) -> VariantResponse:
+    """The stored variant owning the task key — the grading and telemetry
+    join path; 404 when no variant was ever generated for the task."""
+    record = store.get_by_task(task_id)
+    if record is None:
+        raise HTTPException(
+            status_code=404, detail=f"no stored variant for task {task_id!r}"
+        )
+    return _to_response(record, competency_id=_competency_for(record.template_id))

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

@@ -19,6 +19,7 @@ from .api import (
     proctor_router,
     sandboxes_router,
     telemetry_router,
+    variants_router,
 )
 from .config import Settings
 from .grading.engine import GradingEngine
@@ -27,6 +28,8 @@ from .llm import create_provider
 from .sandbox import SandboxManager, UnshareBackend
 from .telemetry.ingest import TraceIntegrityMap
 from .telemetry.store import SQLiteTraceStore
+from .variants.generator import VariantGenerator
+from .variants.store import SQLiteVariantStore
 
 logger = logging.getLogger(__name__)
 
@@ -73,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
@@ -92,6 +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)
             )
 
         async def _reaper_loop() -> None:
@@ -115,6 +143,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
             await manager.destroy_all()
             trace_store.close()
             grade_store.close()
+            variant_store.close()
             await app.state.http_client.aclose()
 
     app = FastAPI(title="Nextcraft AI Service", version="0.3.0", lifespan=lifespan)
@@ -143,6 +172,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     app.include_router(proctor_router)
     app.include_router(sandboxes_router)
     app.include_router(telemetry_router)
+    app.include_router(variants_router)
     return app
 
 

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/ai_service/prompts/variant.py

@@ -0,0 +1,46 @@
+"""Variant instantiation prompt (D-029, REQ-3-005).
+
+The model's ONLY job is to render already-sampled slot values into a task
+statement — it never invents parameters (the seeded sampler is pure code)
+and never changes difficulty. Prompt-injection surface is bounded: the
+variable inputs are the skeleton text, the seeded slot values, and the
+template title — nothing from the learner's environment.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from ..llm.types import Message
+
+if TYPE_CHECKING:  # pragma: no cover - keeps this module pure text
+    from ..variants.templates import TaskTemplate
+
+VARIANT_SYSTEM_PROMPT = (
+    "You instantiate per-learner task variants for a competency-based AI school. "
+    "You receive a task statement skeleton and ALREADY-SAMPLED slot values. "
+    "Render the slot values into the skeleton, producing a complete, unambiguous "
+    "task statement a learner can build against. Rules:\n"
+    "- Use EXACTLY the given slot values; do not invent, rename, or add parameters.\n"
+    "- Keep the engineering depth IDENTICAL across draws: slot values change the "
+    "scenario, never the difficulty or scope.\n"
+    "- Keep the statement in the same language and register as the skeleton.\n"
+    "- Output STRICT JSON only: {\"statement\": \"<rendered statement>\"}.\n"
+)
+
+VARIANT_SCHEMA_HINT = '{"statement": "<complete rendered task statement string>"}'
+
+
+def render_variant_prompt(template: TaskTemplate, params: dict[str, str | int]) -> list[Message]:
+    """Messages for one seeded instantiation (D-020 defense drives the call)."""
+    slot_lines = "\n".join(f"  {{{slot.name}}} = {params[slot.name]!r}" for slot in template.slots)
+    user = (
+        f"Template: {template.title} (id={template.id})\n"
+        f"Statement skeleton:\n{template.statement_skeleton}\n\n"
+        f"Seeded slot values (use EXACTLY these):\n{slot_lines}\n\n"
+        "Render the complete task statement now."
+    )
+    return [
+        Message(role="system", content=VARIANT_SYSTEM_PROMPT),
+        Message(role="user", content=user),
+    ]

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

@@ -0,0 +1,31 @@
+"""Per-learner variant task generation — templates, generator, VariantStore (REQ-3-005).
+
+Boundary rule (D-027): variants/ is an engine module — it never imports
+api/; its ONLY agents/ dependency is the module-direct
+agents.structured import in generator.py (the sanctioned shared D-020
+structured defense, same exception as grading/engine.py). api/ composes
+the generator and store via DI; store.py imports config only.
+
+CO-ORDINATION NOTE (ADD, don't REMOVE — same convention as grading/):
+This __init__.py is a minimal placeholder created by the VariantStore
+task (4-1-02). The templates task (4-1-01) owns this file's final shape
+— when templates.py lands, ADD its exports alongside these; do not
+remove the store exports below.
+
+Wave status: store.py (VariantRecord, VariantStore, SQLiteVariantStore)
+landed in Wave 1 (task 4-1-02); templates.py is Wave 1 task 4-1-01;
+generator.py is Wave 2 (4-2-01).
+"""
+
+from .store import SQLiteVariantStore, VariantRecord, VariantStore
+from .templates import TEMPLATES, TaskTemplate, get_template, template_for_competency
+
+__all__ = [
+    "SQLiteVariantStore",
+    "TEMPLATES",
+    "TaskTemplate",
+    "VariantRecord",
+    "VariantStore",
+    "get_template",
+    "template_for_competency",
+]

apps/ai-service/ai_service/variants/generator.py

@@ -0,0 +1,136 @@
+"""Seeded per-learner variant generator (D-029, REQ-3-005).
+
+Contract (binding, from GRILL + PLAN Must-Haves):
+- REPRODUCIBLE: seed = sha256(template_id|learner_id|milestone); the same
+  (template, learner) re-derives the same seed, params, task_id — and the
+  second generate() call is a cache hit with NO LLM call.
+- DISTINCT: different learners on the same template draw different params
+  (the sampler is seeded per-learner) and receive distinct statements.
+- NEVER BLOCKS ON THE LLM: the deterministic skeleton render
+  (`template.render(params)`) is a complete, valid statement; if the D-020
+  LLM render fails after its bounded retry, the fallback is used — and
+  because the fallback is exactly `template.render(seed-params)`, it is
+  auditable from the persisted seed + params without a provenance column.
+- AUDITABLE: seed + params + statement persist via VariantStore
+  (insert-only first-wins) — the proctoring cross-check path.
+- FAIR (a-5): slot draws change the scenario, never the difficulty; the
+  template's rubric anchors bound the expected effort envelope, so every
+  variant of one template is held to the same bar.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from ..agents.structured import StructuredOutputError, structured_completion
+from ..llm.types import Message
+from ..prompts.variant import VARIANT_SCHEMA_HINT, render_variant_prompt
+from .store import VariantRecord
+from .templates import TaskTemplate, get_template
+
+if TYPE_CHECKING:  # pragma: no cover
+    from ..llm.base import LLMProvider
+    from .store import VariantStore
+
+MILESTONE = "v0.3"
+
+
+class RenderedVariant(BaseModel):
+    """D-20-validated LLM render output (statement only — files come from the template)."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    statement: str = Field(min_length=20)
+
+
+class UnknownTemplateError(ValueError):
+    """Raised when generate() is asked for a template id not in the library."""
+
+
+def derive_seed(template_id: str, learner_id: str, milestone: str = MILESTONE) -> str:
+    """Reproducible per-(template, learner, milestone) seed (D-029)."""
+    return hashlib.sha256(f"{template_id}|{learner_id}|{milestone}".encode()).hexdigest()
+
+
+def derive_task_id(seed: str) -> str:
+    """Deterministic grading/telemetry task key from the seed (16 hex chars)."""
+    return f"task-{seed[:16]}"
+
+
+class VariantGenerator:
+    """Seeded instantiation over the template library. DI: store + provider."""
+
+    def __init__(self, store: VariantStore, provider: LLMProvider, model: str) -> None:
+        self._store = store
+        self._provider = provider
+        self._model = model
+
+    async def generate(self, learner_id: str, template_id: str) -> VariantRecord:
+        template = get_template(template_id)
+        if template is None:
+            raise UnknownTemplateError(f"no task template with id {template_id!r}")
+
+        # Cache: D-029 reproducibility — same (learner, template) is served
+        # from the store with no LLM call.
+        cached = self._store.get(learner_id, template_id)
+        if cached is not None:
+            return cached
+
+        seed_hex = derive_seed(template_id, learner_id)
+        task_id = derive_task_id(seed_hex)
+        params = template.sample_params(_seed_int(seed_hex))
+        _validate_params(template, params)
+
+        statement = await self._render(template, params)
+        record = VariantRecord(
+            learner_id=learner_id,
+            task_id=task_id,
+            template_id=template_id,
+            seed=seed_hex,
+            params=dict(params),
+            statement=statement,
+            starter_files=dict(template.starter_files),
+            created_at=datetime.now(UTC),
+        )
+        self._store.save(record)
+        return record
+
+    async def _render(self, template: TaskTemplate, params: dict[str, str | int]) -> str:
+        """LLM render via D-020; deterministic fallback never blocks task work.
+
+        Provenance note: unlike grades, variants carry no `model` column —
+        the deterministic fallback is exactly `template.render(params)`,
+        re-derivable from the persisted seed + params, so a fallback render is
+        auditable without storing provenance (the seed IS the provenance).
+        """
+        messages: list[Message] = render_variant_prompt(template, params)
+        try:
+            rendered = await structured_completion(
+                self._provider,
+                messages,
+                model=self._model,
+                schema=RenderedVariant,
+                schema_hint=VARIANT_SCHEMA_HINT,
+            )
+        except StructuredOutputError:
+            # Deterministic fallback: the skeleton + seeded slots is already a
+            # complete statement, re-derivable from the persisted seed.
+            return template.render(params)
+        return rendered.statement
+
+
+def _seed_int(seed_hex: str) -> int:
+    """Stable int for random.Random from the hex seed."""
+    return int(seed_hex[:16], 16)
+
+
+def _validate_params(template: TaskTemplate, params: dict[str, str | int]) -> None:
+    """Defense in depth: every sampled value must be schema-valid (a-5)."""
+    for slot in template.slots:
+        value = params.get(slot.name)
+        if value is None or not slot.validate_value(value):
+            raise ValueError(f"sampled params invalid for slot {slot.name!r}: {value!r}")

apps/ai-service/ai_service/variants/store.py

@@ -0,0 +1,311 @@
+"""VariantStore — variant persistence protocol + SQLite implementation (REQ-3-005, D-027).
+
+Postgres-migration-ready (D-027): the protocol is the only surface the
+variant generator and API layers touch; swapping SQLiteVariantStore for a
+Postgres-backed implementation must not change call sites. The
+`variant_record` table uses only portable column types (str / JSON /
+datetime), so the same SQLModel schema stands up unchanged on Postgres.
+
+Insert-only, NOT upsert: (learner_id, template_id) is the variant identity
+and the FIRST generation is authoritative — reproducibility (D-029) means
+the seed re-derives the same variant, so the generator's cache path serves
+`get` instead of saving again. `save` is a plain INSERT; a duplicate pair
+raises sqlalchemy.exc.IntegrityError to the caller (documented behavior).
+`task_id` is unique too — it is the grading/telemetry trace key, so a
+trace or grade can never silently join to a different variant. Both
+rejections are deliberate: overwriting a stored variant would swap a
+learner's graded task underneath its trace and grade (audit corruption).
+Contrast TraceStore.append (dedup-keep-first, swallowed — at-least-once
+ingest) and GradeStore.save (upsert-latest-wins — a regrade is
+latest-state); this store is the third contract of the D-027 family.
+
+Concurrency (a-3): the store enables WAL + synchronous=NORMAL and a busy
+timeout at connection time, so a generation writer and API readers do not
+hit `database is locked` on the single-box pilot.
+
+`created_at` contract: callers stamp UTC (datetime.now(UTC)); SQLite
+stores it naive and the read paths re-label it tz-aware UTC (same
+boundary normalization as TelemetryEvent.ts / GradeRecord.created_at, so
+the contract holds on any backend).
+
+Boundary (D-027): `variants/` never imports `agents/` / `api/`; this
+module imports config only.
+"""
+
+import logging
+import sqlite3
+from collections.abc import Iterator
+from contextlib import contextmanager
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any, Protocol
+
+import sqlalchemy as sa
+from sqlalchemy import JSON, Index, UniqueConstraint
+from sqlalchemy.orm import validates
+from sqlmodel import Field, Session, SQLModel, create_engine, select
+
+from ..config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+class VariantRecord(SQLModel, table=True):
+    """A persisted task variant; (learner_id, template_id) is the PK — first wins.
+
+    Written once by the variant generator (Task 4-2-01), read by the API
+    layer and proctoring cross-checks through the VariantStore protocol.
+    Constraint enforcement mirrors TelemetryEvent / GradeRecord: sqlmodel
+    0.0.42's metaclass drops pydantic constraints on table models, so
+    SQLAlchemy `@validates` hooks enforce instead and the column types
+    stay Postgres-ready (D-027).
+
+    Field contract:
+        learner_id     — non-empty learner identifier (same id space as
+                         traces and grades).
+        template_id    — non-empty task template identifier; variant
+                         identity is the (learner_id, template_id) pair —
+                         the pair the generator caches on (exactly one
+                         variant per learner per template).
+        task_id        — non-empty, GLOBALLY unique task identifier; the
+                         grading/telemetry trace key (the (learner_id,
+                         task_id) pair TraceStore / GradeStore key on),
+                         stamped at generation so a variant's trace and
+                         grade join back to it exactly once.
+        seed           — non-empty variant seed (D-029); derived from
+                         (template_id, learner_id, milestone) so the
+                         variant is reproducible and auditable.
+        params         — typed parameter-slot values the generator filled;
+                         JSON dict. An empty dict is legal (a slotless
+                         template).
+        statement      — non-empty rendered task statement shown to the
+                         learner (distinct per learner by construction,
+                         REQ-3-005).
+        starter_files  — workspace scaffold: filename -> file content;
+                         JSON dict. An empty dict is legal (no scaffold).
+        created_at     — UTC generation timestamp.
+    """
+
+    __tablename__ = "variant_record"
+    # The composite PK covers (learner_id, template_id) point lookups; the
+    # unique task_id covers get_by_task (the grading/telemetry join path);
+    # the two secondary indexes cover list_for_learner / list_by_template
+    # ordered by created_at without a sort step (Postgres target D-027).
+    __table_args__ = (
+        UniqueConstraint("task_id", name="uq_variant_record_task_id"),
+        Index("ix_variant_record_learner_created", "learner_id", "created_at"),
+        Index("ix_variant_record_template_created", "template_id", "created_at"),
+    )
+
+    learner_id: str = Field(primary_key=True)
+    template_id: str = Field(primary_key=True)
+    task_id: str
+    seed: str
+    # JSON columns: stored as TEXT on SQLite, native JSONB on Postgres (D-027).
+    params: dict[str, Any] = Field(default_factory=dict, sa_type=JSON)
+    statement: str
+    starter_files: dict[str, Any] = Field(default_factory=dict, sa_type=JSON)
+    created_at: datetime
+
+    @validates("learner_id", "template_id", "task_id")
+    def _ids_non_empty(self, key: str, value: str) -> str:
+        if not value:
+            raise ValueError(f"{key} must be a non-empty identifier")
+        return value
+
+    @validates("seed")
+    def _seed_non_empty(self, key: str, value: str) -> str:
+        if not value:
+            raise ValueError(f"{key} must be a non-empty seed string")
+        return value
+
+    @validates("statement")
+    def _statement_non_empty(self, key: str, value: str) -> str:
+        if not value:
+            raise ValueError(f"{key} must be a non-empty statement string")
+        return value
+
+
+class VariantStore(Protocol):
+    """Persistence contract for reproducible per-learner task variants.
+
+    Implemented by SQLiteVariantStore (v0.3, D-027); a Postgres
+    implementation must satisfy the same surface.
+    """
+
+    def save(self, variant: VariantRecord) -> None:
+        """Persist a new variant. INSERT-ONLY on (learner_id, template_id):
+        the FIRST generated variant is authoritative (reproducibility,
+        D-029); a duplicate pair raises sqlalchemy.exc.IntegrityError to
+        the caller — the generator serves cached variants via `get`
+        instead of saving again. `task_id` is unique too: claiming an
+        existing trace key for a different variant is equally rejected.
+        NOT upsert; contrast GradeStore.save (latest-wins) and
+        TraceStore.append (dedup-keep-first, swallowed).
+        """
+        ...
+
+    def get(self, learner_id: str, template_id: str) -> VariantRecord | None:
+        """The learner's stored variant for the template; None when none
+        exists. Detached from any DB session — safe to pass across layers.
+        """
+        ...
+
+    def get_by_task(self, task_id: str) -> VariantRecord | None:
+        """The variant owning the task key (the grading/telemetry join
+        path); None when none exists. Detached from any DB session.
+        """
+        ...
+
+    def list_for_learner(self, learner_id: str) -> list[VariantRecord]:
+        """All stored variants for the learner, ordered by created_at
+        ascending (chronological; task_id breaks same-instant ties).
+        Empty list when the learner has none.
+        """
+        ...
+
+    def list_by_template(self, template_id: str) -> list[VariantRecord]:
+        """All stored variants generated from the template — one row per
+        learner — ordered by created_at ascending (chronological;
+        learner_id breaks same-instant ties). Empty list when the
+        template has none. The proctoring cross-check path (seed params
+        per learner) reads through this.
+        """
+        ...
+
+    def close(self) -> None:
+        """Release DB connections. Store must not be used after close."""
+        ...
+
+
+def _sqlite_connect(dbapi_connection: sqlite3.Connection, _: object) -> None:
+    """Per-connection pragma setup (a-3). Mirrors telemetry/grading stores.
+
+    journal_mode=WAL  — readers never block the single writer.
+    synchronous=NORMAL — safe in WAL mode, avoids full fsync-per-commit.
+    busy_timeout=5000  — retry briefly under contention instead of
+                         `OperationalError: database is locked`.
+    """
+    cursor = dbapi_connection.cursor()
+    cursor.execute("PRAGMA journal_mode=WAL")
+    cursor.execute("PRAGMA synchronous=NORMAL")
+    cursor.execute("PRAGMA busy_timeout=5000")
+    cursor.close()
+
+
+def _as_utc(ts: datetime) -> datetime:
+    """Timestamps travel as naive datetime on SQLite, tz-aware elsewhere.
+
+    SQLite (via SQLModel) drops tzinfo; Postgres TIMESTAMP WITH TIME ZONE
+    keeps it. Normalizing on the read path makes the store's contract
+    tz-aware UTC regardless of the backend (D-027).
+    """
+    if ts.tzinfo is None:
+        return ts.replace(tzinfo=UTC)  # naive UTC read-side: label as UTC
+    return ts.astimezone(UTC)
+
+
+class SQLiteVariantStore:
+    """SQLite-backed VariantStore (SQLModel). Third protocol-wrapped store
+    of the D-027 family (first: SQLiteTraceStore, second: SQLiteGradeStore).
+    """
+
+    def __init__(self, db_path: Path | None = None) -> None:
+        self._db_path: Path = db_path if db_path is not None else Settings().db_path
+        self._engine = create_engine(f"sqlite:///{self._db_path}")
+        sa.event.listen(self._engine, "connect", _sqlite_connect)
+        SQLModel.metadata.create_all(self._engine)
+
+    @contextmanager
+    def _session(self) -> Iterator[Session]:
+        # expire_on_commit=False: identical session behavior to the other
+        # D-027 stores. save() never commits on the error path and the read
+        # paths never commit, but a uniform flag across the family keeps
+        # their detachment guarantees from diverging.
+        with Session(self._engine, expire_on_commit=False) as session:
+            yield session
+
+    def save(self, variant: VariantRecord) -> None:
+        # Plain INSERT, no merge: overwriting a stored variant would swap a
+        # learner's graded task underneath its trace and grade (audit
+        # corruption), so a duplicate identity is a race or bug to SURFACE,
+        # not paper over. The generator's cache path (get before generate)
+        # makes duplicate saves a programming error, not a normal flow.
+        # The trace store swallows its IntegrityError (dedup is the
+        # contract there); the grade store merges (latest-wins is the
+        # contract there); this store re-raises (first-wins is the
+        # contract here).
+        with self._session() as session:
+            try:
+                session.add(variant)
+                session.commit()
+            except sa.exc.IntegrityError:
+                session.rollback()
+                logger.debug(
+                    "variant insert rejected (identity already stored): "
+                    "learner=%s template=%s task=%s",
+                    variant.learner_id,
+                    variant.template_id,
+                    variant.task_id,
+                )
+                raise
+        logger.debug(
+            "variant saved: %s/%s task=%s seed=%s",
+            variant.learner_id,
+            variant.template_id,
+            variant.task_id,
+            variant.seed,
+        )
+
+    def get(self, learner_id: str, template_id: str) -> VariantRecord | None:
+        with self._session() as session:
+            record = session.get(VariantRecord, (learner_id, template_id))
+            if record is None:
+                return None
+            record.created_at = _as_utc(record.created_at)
+            # Detach from the session: callers must not depend on
+            # open-session ORM magic (lazy loads fail once it closes).
+            session.expunge(record)
+            return record
+
+    def get_by_task(self, task_id: str) -> VariantRecord | None:
+        with self._session() as session:
+            stmt = select(VariantRecord).where(VariantRecord.task_id == task_id)
+            record = session.exec(stmt).first()
+            if record is None:
+                return None
+            record.created_at = _as_utc(record.created_at)
+            session.expunge(record)
+            return record
+
+    def list_for_learner(self, learner_id: str) -> list[VariantRecord]:
+        with self._session() as session:
+            stmt = (
+                select(VariantRecord)
+                .where(VariantRecord.learner_id == learner_id)
+                # Chronological; task_id is a deterministic tie-break for
+                # variants stamped within the same instant.
+                .order_by(VariantRecord.created_at, VariantRecord.task_id)
+            )
+            results = session.exec(stmt).all()
+            for row in results:
+                row.created_at = _as_utc(row.created_at)
+                session.expunge(row)
+            return list(results)
+
+    def list_by_template(self, template_id: str) -> list[VariantRecord]:
+        with self._session() as session:
+            stmt = (
+                select(VariantRecord)
+                .where(VariantRecord.template_id == template_id)
+                # Chronological; learner_id is a deterministic tie-break.
+                .order_by(VariantRecord.created_at, VariantRecord.learner_id)
+            )
+            results = session.exec(stmt).all()
+            for row in results:
+                row.created_at = _as_utc(row.created_at)
+                session.expunge(row)
+            return list(results)
+
+    def close(self) -> None:
+        self._engine.dispose()

apps/ai-service/ai_service/variants/templates.py

@@ -0,0 +1,305 @@
+"""Task template library for seeded variant generation (D-029, REQ-3-005).
+
+A `TaskTemplate` binds a competency (D-021-aligned corpus ID), a statement
+skeleton with `{slot}` placeholders, typed `ParameterSlot`s, difficulty-
+normalization rubric anchors (the expected feature envelope that bounds
+variant fairness in the a-5 envelope test — grader-prompt shipment is the
+tracked P4 follow-up; grading is variant-blind today), and starter-file
+scaffolds served into the sandbox workdir (wired in P6).
+
+Slot sampling is PURE CODE: `random.Random(seed)` over typed slots — fully
+reproducible for a given seed, independent of the LLM. The LLM only renders
+the seeded slot values into the statement skeleton (D-020 defense).
+"""
+
+from __future__ import annotations
+
+import random
+import re
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+SlotType = Literal["enum", "int_range", "string_set"]
+
+
+class ParameterSlot(BaseModel):
+    """One typed fill-in for a statement skeleton."""
+
+    model_config = ConfigDict(frozen=True)
+
+    name: str = Field(min_length=1)
+    type: SlotType
+    values: list[str] = Field(default_factory=list)  # enum/string_set options
+    lo: int | None = None  # int_range bounds
+    hi: int | None = None
+
+    @field_validator("values")
+    @classmethod
+    def _values_nonempty_for_enums(cls, v: list[str], info) -> list[str]:
+        if info.data.get("type") in ("enum", "string_set") and not v:
+            raise ValueError(f"slot {info.data.get('name')!r} needs values")
+        return v
+
+    def sample(self, rng: random.Random) -> str | int:
+        """Deterministic sample from the seeded RNG. Validated after sampling."""
+        if self.type == "enum" or self.type == "string_set":
+            return rng.choice(self.values)
+        if self.type == "int_range":
+            lo = self.lo if self.lo is not None else 0
+            hi = self.hi if self.hi is not None else lo
+            if hi < lo:
+                raise ValueError(f"slot {self.name!r}: hi < lo")
+            return rng.randint(lo, hi)
+        raise ValueError(f"unsupported slot type: {self.type!r}")
+
+    def validate_value(self, value: str | int) -> bool:
+        """Is `value` schema-valid for this slot? (params JSON gate, a-5.)"""
+        if self.type in ("enum", "string_set"):
+            return isinstance(value, str) and value in self.values
+        if self.type == "int_range":
+            lo = self.lo if self.lo is not None else 0
+            hi = self.hi if self.hi is not None else lo
+            return isinstance(value, int) and lo <= value <= hi
+        return False
+
+
+class RubricAnchors(BaseModel):
+    """Difficulty-normalization anchors for the grader (a-5).
+
+    Expected FEATURE ENVELOPE (digest-space): the expected effort band
+    for this template, so two variants of one template are held to the
+    same bar regardless of which slot values a learner drew. The a-5
+    envelope test (tests/variants/test_generator.py) binds variants to
+    these bands in code. Shipping them into the grader prompt context is
+    the P4 must-have follow-up tracked for final review: grading is
+    variant-blind in the current wiring (engine.py stamps
+    variant_seed=None), so today the anchors gate variant fairness in
+    tests only — not yet in the LLM prompt.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    expected_edit_count_band: tuple[int, int]
+    expected_min_test_runs: int
+    expected_error_fix_cycles_band: tuple[int, int]
+    notes: str = ""
+
+
+class TaskTemplate(BaseModel):
+    """A reusable task shape; variants instantiate it per learner."""
+
+    model_config = ConfigDict(frozen=True)
+
+    id: str = Field(min_length=1)
+    competency_id: str = Field(min_length=1)  # D-021 corpus alignment
+    title: str
+    statement_skeleton: str = Field(min_length=1)  # {slot} placeholders
+    slots: list[ParameterSlot] = Field(min_length=1)
+    rubric_anchors: RubricAnchors
+    starter_files: dict[str, str] = Field(default_factory=dict)  # path -> content
+    test_command: str
+
+    @field_validator("statement_skeleton")
+    @classmethod
+    def _skeleton_placeholders(cls, v: str) -> str:
+        if "{" not in v or "}" not in v:
+            raise ValueError("statement_skeleton needs at least one {slot}")
+        return v
+
+    def render(self, params: dict[str, str | int]) -> str:
+        """Fill the skeleton with validated params."""
+        for slot in self.slots:
+            if slot.name not in params:
+                raise ValueError(f"missing param for slot {slot.name!r}")
+            if not slot.validate_value(params[slot.name]):
+                raise ValueError(f"invalid value for slot {slot.name!r}: {params[slot.name]!r}")
+        return self.statement_skeleton.format(**params)
+
+    def sample_params(self, seed: int) -> dict[str, str | int]:
+        """Seeded, reproducible, schema-valid slot values (pure code)."""
+        rng = random.Random(seed)
+        return {slot.name: slot.sample(rng) for slot in self.slots}
+
+
+# --- Template library (v0.3 initial set) --------------------------------------
+# Competency IDs are D-021-aligned with the Python corpus
+# (ai_service/corpus/learner_context.py) and the TS mock-data layer
+# (packages/mock-data/competency-stacks.ts: deterministic cid() scheme).
+
+TEMPLATES: dict[str, TaskTemplate] = {
+    "tpl-llm-judge": TaskTemplate(
+        id="tpl-llm-judge",
+        competency_id="stack-orchestration-c007",
+        title="Build an LLM-as-Judge Evaluator",
+        statement_skeleton=(
+            "Build a small LLM-as-judge evaluator for {domain} answers. "
+            "The judge must score each answer on {criterion} using a 0-4 scale, "
+            "return structured JSON, and handle at least {edge_cases} edge-case "
+            "answer classes (empty, off-topic, adversarial). Include a tiny "
+            "repro test set of at least {test_size} examples and print a summary "
+            "table of scores."
+        ),
+        slots=[
+            ParameterSlot(
+                name="domain",
+                type="enum",
+                values=["customer-support", "code-review", "summarization", "tutoring"],
+            ),
+            ParameterSlot(
+                name="criterion",
+                type="enum",
+                values=["factual-accuracy", "helpfulness", "safety", "completeness"],
+            ),
+            ParameterSlot(name="edge_cases", type="int_range", lo=2, hi=4),
+            ParameterSlot(name="test_size", type="int_range", lo=3, hi=8),
+        ],
+        rubric_anchors=RubricAnchors(
+            expected_edit_count_band=(3, 25),
+            expected_min_test_runs=2,
+            expected_error_fix_cycles_band=(0, 4),
+            notes="Slot draw changes the SCENARIO, not the engineering depth.",
+        ),
+        starter_files={
+            "README.md": (
+                "# LLM-as-Judge Evaluator\n\n"
+                "Implement `judge.py`:\n"
+                "- `score(answer: str) -> dict` — 0-4 on the named criterion\n"
+                "- structured JSON output (schema below)\n"
+                "- edge-case classes handled explicitly\n"
+                "- `pytest` must pass\n"
+            ),
+            "judge.py": "def score(answer: str) -> dict:\n    raise NotImplementedError\n",
+            "test_judge.py": "def test_placeholder():\n    assert True\n",
+        },
+        test_command="pytest -q",
+    ),
+    "tpl-guardrail-schema": TaskTemplate(
+        id="tpl-guardrail-schema",
+        competency_id="stack-orchestration-c008",
+        title="Schema Guardrail Pipeline",
+        statement_skeleton=(
+            "Implement an output-validation guardrail for a model returning "
+            "{entity} records. Validate against a typed schema with {field_count} "
+            "required fields, coerce or reject {failure_mode} failures, and emit "
+            "a fallback response for invalid payloads. Cover with at least "
+            "{test_size} unit tests including malformed JSON."
+        ),
+        slots=[
+            ParameterSlot(
+                name="entity",
+                type="enum",
+                values=["user-profile", "job-posting", "candidate", "invoice"],
+            ),
+            ParameterSlot(
+                name="failure_mode",
+                type="enum",
+                values=["strict-reject", "coerce-when-safe"],
+            ),
+            ParameterSlot(name="field_count", type="int_range", lo=4, hi=8),
+            ParameterSlot(name="test_size", type="int_range", lo=4, hi=10),
+        ],
+        rubric_anchors=RubricAnchors(
+            expected_edit_count_band=(3, 30),
+            expected_min_test_runs=2,
+            expected_error_fix_cycles_band=(0, 5),
+            notes="All slot draws land in the same engineering band.",
+        ),
+        starter_files={
+            "README.md": (
+                "# Schema Guardrail\n\nImplement `guardrail.py`:\n"
+                "- `validate(payload: dict) -> dict | Fallback`\n"
+                "- required-field checks, failure policy, fallback emission\n"
+            ),
+            "guardrail.py": "def validate(payload: dict):\n    raise NotImplementedError\n",
+            "test_guardrail.py": "def test_placeholder():\n    assert True\n",
+        },
+        test_command="pytest -q",
+    ),
+    "tpl-rag-chunker": TaskTemplate(
+        id="tpl-rag-chunker",
+        competency_id="stack-orchestration-c005",
+        title="RAG Chunking Strategy",
+        statement_skeleton=(
+            "Implement a document chunker for {doc_type} retrieval. Support "
+            "{strategy} chunking with a target size of ~{chunk_size} tokens, "
+            "preserve {invariant} across chunk boundaries, and evaluate overlap "
+            "quality with at least {test_size} fixture documents."
+        ),
+        slots=[
+            ParameterSlot(
+                name="doc_type",
+                type="enum",
+                values=["technical-docs", "legal-contracts", "transcripts"],
+            ),
+            ParameterSlot(
+                name="strategy",
+                type="enum",
+                values=["fixed-window", "semantic-boundary", "hybrid"],
+            ),
+            ParameterSlot(
+                name="invariant",
+                type="enum",
+                values=["code-block-integrity", "section-headers", "sentence-completeness"],
+            ),
+            ParameterSlot(name="chunk_size", type="int_range", lo=200, hi=800),
+            ParameterSlot(name="test_size", type="int_range", lo=3, hi=6),
+        ],
+        rubric_anchors=RubricAnchors(
+            expected_edit_count_band=(4, 35),
+            expected_min_test_runs=2,
+            expected_error_fix_cycles_band=(0, 6),
+            notes="Strategy draw changes implementation shape, not depth.",
+        ),
+        starter_files={
+            "README.md": (
+                "# RAG Chunker\n\nImplement `chunker.py`:\n"
+                "- `chunk(text: str) -> list[str]`\n- invariant preserved\n- tests green\n"
+            ),
+            "chunker.py": "def chunk(text: str) -> list[str]:\n    raise NotImplementedError\n",
+            "test_chunker.py": "def test_placeholder():\n    assert True\n",
+        },
+        test_command="pytest -q",
+    ),
+}
+
+
+_KNOWN_COMPETENCY_IDS: set[str] = {
+    # D-021: mirrored from ai_service/corpus/learner_context.py — the Python
+    # source of truth for stack-orchestration competencies used by v0.2 agents.
+    "stack-orchestration-c001",
+    "stack-orchestration-c002",
+    "stack-orchestration-c003",
+    "stack-orchestration-c004",
+    "stack-orchestration-c005",
+    "stack-orchestration-c007",
+    "stack-orchestration-c008",
+    "stack-orchestration-c011",
+    "stack-designer-c001",
+    "stack-designer-c002",
+    "stack-safety-c021",
+}
+
+
+def get_template(template_id: str) -> TaskTemplate | None:
+    return TEMPLATES.get(template_id)
+
+
+def template_for_competency(competency_id: str) -> list[TaskTemplate]:
+    return [t for t in TEMPLATES.values() if t.competency_id == competency_id]
+
+
+def validate_competency_binding() -> None:
+    """All templates must bind to known D-021 corpus competency IDs."""
+    for t in TEMPLATES.values():
+        if t.competency_id not in _KNOWN_COMPETENCY_IDS:
+            raise ValueError(
+                f"template {t.id!r} binds unknown competency {t.competency_id!r}"
+            )
+
+
+def slots_pattern_ok(skeleton: str, slots: list[ParameterSlot]) -> bool:
+    """Every {placeholder} in the skeleton has a matching slot and vice versa."""
+    placeholders = set(re.findall(r"\{([a-z_][a-z0-9_]*)\}", skeleton))
+    slot_names = {s.name for s in slots}
+    return placeholders == slot_names

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

@@ -0,0 +1,277 @@
+"""Variant API tests — generation, cache, distinctness over HTTP (Task 4-3-01).
+
+Contract under test (api/variants.py, REQ-3-005):
+
+    POST /v1/variants {learner_id, template_id}
+        first request       → 200 generated variant (task_id, seed, params,
+                              statement, starter_files, competency_id)
+        repeat request      → 200 the SAME cached variant with ZERO LLM
+                              calls (D-029 reproducibility through the
+                              whole HTTP stack)
+        unknown template    → 404
+    POST /v1/variants {learner_id, competency_id}
+        bound competency    → 200 the first template for that competency
+        unknown competency  → 404
+        neither id given    → 422
+    GET  /v1/variants/{task_id}
+        generated earlier  → 200 the stored variant (generate→get roundtrip)
+        unknown task       → 404
+    GET  /v1/variants?learner_id=...
+        → 200 that learner's variants only (scoping); [] for a learner
+          with none.
+
+Distinctness (REQ-3-005, API level): two learners POSTing the same
+template receive distinct statements, seeds and task_ids, and GET by
+task_id hands each back their own variant.
+
+Wiring: per-test tmp-path SQLiteVariantStore + a pre-set VariantGenerator
+(state-injection override — the lifespan adopts variant_store /
+variant_generator from app.state instead of constructing them; same
+pattern as test_grading.py / test_telemetry_ingest.py). The generator
+binds a ScriptedRenderProvider so each test controls the render LLM
+exactly, and counts calls so the cache test can assert the LLM was never
+reached on the second POST.
+
+Zero network: providers are MockProvider family members only (conftest
+rule, enforced in _make_client).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from ai_service.config import Settings
+from ai_service.llm.mock import MockProvider
+from ai_service.main import create_app
+from ai_service.variants.generator import VariantGenerator
+from ai_service.variants.store import SQLiteVariantStore
+from ai_service.variants.templates import get_template
+
+LEARNER_A = "variant-learner-a"
+LEARNER_B = "variant-learner-b"
+TEMPLATE = "tpl-llm-judge"
+COMPETENCY = "stack-orchestration-c007"  # tpl-llm-judge's D-021 binding
+
+
+class ScriptedRenderProvider(MockProvider):
+    """Deterministic render whose statement embeds the params (distinct per
+    draw); counts calls so the cache test asserts zero LLM calls on the
+    second POST. Mirrors tests/variants/test_generator.py."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.calls = 0
+
+    async def chat(self, messages, *, model, temperature=0.7, response_format=None):  # noqa: ANN001
+        self.calls += 1
+        import json
+
+        user = next(m.content for m in reversed(messages) if m.role == "user")
+        # Distinct per distinct params: hash the seeded slot lines.
+        fingerprint = abs(hash(user)) % 10_000
+        return json.dumps({"statement": f"Scripted variant #{fingerprint} — build it."})
+
+
+@pytest.fixture()
+def store(tmp_path: Path) -> Iterator[SQLiteVariantStore]:
+    s = SQLiteVariantStore(db_path=tmp_path / "variants.db")
+    yield s
+    s.close()
+
+
+@pytest.fixture()
+def provider() -> ScriptedRenderProvider:
+    return ScriptedRenderProvider()
+
+
+def _make_client(
+    tmp_path: Path, store: SQLiteVariantStore, provider: MockProvider
+) -> TestClient:
+    """App + TestClient with a pre-set store + generator.
+
+    The lifespan adopts both (state-injection override); the generator
+    binds OUR provider, so the fixture — not Settings — scripts the LLM.
+    Cloud-free guard: the provider must be a MockProvider family member
+    (conftest rule, enforced here because this module builds its own
+    client rather than consuming the conftest one).
+    """
+    assert isinstance(provider, MockProvider)
+    settings = Settings(
+        provider="mock",
+        db_path=tmp_path / "variant-test.db",
+        sandbox_dir=tmp_path / "sandboxes",
+    )
+    app = create_app(settings)
+    app.state.variant_store = store
+    if getattr(app.state, "variant_generator", None) is None:
+        app.state.variant_generator = VariantGenerator(
+            store, provider, model="gemma4:31b"
+        )
+    return TestClient(app)
+
+
+def _post(
+    client: TestClient,
+    learner_id: str,
+    template_id: str | None = TEMPLATE,
+    competency_id: str | None = None,
+):
+    return client.post(
+        "/v1/variants",
+        json={
+            "learner_id": learner_id,
+            **({"template_id": template_id} if template_id is not None else {}),
+            **({"competency_id": competency_id} if competency_id is not None else {}),
+        },
+    )
+
+
+# -- POST: generation + response shape --------------------------------------------
+
+
+class TestGenerate:
+    def test_post_returns_full_variant_payload(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            response = _post(client, LEARNER_A)
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["learner_id"] == LEARNER_A
+        assert body["template_id"] == TEMPLATE
+        assert body["competency_id"] == COMPETENCY  # enrichment from get_template
+        assert body["task_id"].startswith("task-") and len(body["task_id"]) == len("task-") + 16
+        assert body["seed"]  # non-empty D-029 seed
+        assert body["statement"]  # rendered, non-empty
+        assert body["starter_files"] == get_template(TEMPLATE).starter_files
+        assert body["params"]  # seeded slot values
+        assert body["created_at"]  # ISO timestamp travels
+        assert provider.calls == 1  # first POST renders exactly once
+
+    def test_post_then_get_roundtrip(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            posted = _post(client, LEARNER_A)
+            assert posted.status_code == 200
+
+            fetched = client.get(f"/v1/variants/{posted.json()['task_id']}")
+
+        assert fetched.status_code == 200
+        assert fetched.json() == posted.json()  # identical stored variant
+
+    def test_post_regenerate_is_cache_hit_no_llm_call(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            first = _post(client, LEARNER_A)
+            assert first.status_code == 200
+            assert provider.calls == 1
+
+            second = _post(client, LEARNER_A)  # same (learner, template) → cache
+            assert second.status_code == 200
+            assert second.json() == first.json()  # D-029: the SAME variant
+            assert provider.calls == 1, "cached regenerate must not render again"
+
+
+# -- POST: distinctness (REQ-3-005, API level) ------------------------------------
+
+
+class TestDistinctLearners:
+    def test_two_learners_distinct_statements_and_task_ids(
+        self, tmp_path, store, provider
+    ):
+        with _make_client(tmp_path, store, provider) as client:
+            a = _post(client, LEARNER_A)
+            b = _post(client, LEARNER_B)
+            assert a.status_code == 200 and b.status_code == 200
+
+            a_body, b_body = a.json(), b.json()
+
+            # ...and each learner GETs back exactly their own variant
+            a_fetched = client.get(f"/v1/variants/{a_body['task_id']}")
+            b_fetched = client.get(f"/v1/variants/{b_body['task_id']}")
+
+        assert a_body["statement"] != b_body["statement"]
+        assert a_body["seed"] != b_body["seed"]
+        assert a_body["task_id"] != b_body["task_id"]
+        assert a_body["competency_id"] == b_body["competency_id"]  # same bar (a-5)
+        assert a_fetched.status_code == 200 and a_fetched.json() == a_body
+        assert b_fetched.status_code == 200 and b_fetched.json() == b_body
+
+
+# -- POST: template / competency resolution ---------------------------------------
+
+
+class TestTemplateResolution:
+    def test_unknown_template_404(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            response = _post(client, LEARNER_A, template_id="tpl-does-not-exist")
+
+        assert response.status_code == 404
+        assert "no task template" in response.json()["detail"]
+
+    def test_competency_lookup_uses_first_bound_template(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            response = _post(
+                client, LEARNER_A, template_id=None, competency_id=COMPETENCY
+            )
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["template_id"] == TEMPLATE  # the competency's first template
+        assert body["competency_id"] == COMPETENCY
+
+    def test_unknown_competency_404(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            response = _post(
+                client, LEARNER_A, template_id=None, competency_id="stack-none-c999"
+            )
+
+        assert response.status_code == 404
+        assert "no task template for competency" in response.json()["detail"]
+
+    def test_neither_id_given_422(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            response = _post(client, LEARNER_A, template_id=None, competency_id=None)
+
+        assert response.status_code == 422
+
+
+# -- GET: stored reads ------------------------------------------------------------
+
+
+class TestStoredReads:
+    def test_get_unknown_task_404(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            response = client.get("/v1/variants/task-0000000000000000")
+
+        assert response.status_code == 404
+        assert "no stored variant" in response.json()["detail"]
+
+    def test_list_scoped_to_one_learner(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            assert _post(client, LEARNER_A).status_code == 200
+            assert _post(
+                client, LEARNER_A, template_id="tpl-guardrail-schema"
+            ).status_code == 200
+            assert _post(client, LEARNER_B).status_code == 200
+
+            listed = client.get("/v1/variants", params={"learner_id": LEARNER_A})
+            empty = client.get("/v1/variants", params={"learner_id": "nobody"})
+
+        assert listed.status_code == 200
+        variants = listed.json()["variants"]
+        assert len(variants) == 2  # LEARNER_A's two, LEARNER_B's excluded
+        assert {v["learner_id"] for v in variants} == {LEARNER_A}
+        assert {v["template_id"] for v in variants} == {TEMPLATE, "tpl-guardrail-schema"}
+        # chronological ordering; both carry the competency enrichment
+        assert all(v["competency_id"] for v in variants)
+
+        assert empty.status_code == 200
+        assert empty.json()["variants"] == []
+
+    def test_list_missing_learner_param_422(self, tmp_path, store, provider):
+        with _make_client(tmp_path, store, provider) as client:
+            response = client.get("/v1/variants")
+
+        assert response.status_code == 422

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

apps/ai-service/tests/variants/__init__.py

@@ -0,0 +1 @@
+"""Variant task generation tests (REQ-3-005)."""

apps/ai-service/tests/variants/test_generator.py

@@ -0,0 +1,179 @@
+"""Variant generator tests (Task 4-2-01, REQ-3-005) — D-029 + a-5 binding."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+
+from ai_service.grading.features import compute_digest
+from ai_service.llm.mock import MockProvider
+from ai_service.telemetry.models import TelemetryEvent
+from ai_service.variants.generator import (
+    MILESTONE,
+    VariantGenerator,
+    derive_seed,
+    derive_task_id,
+)
+from ai_service.variants.store import SQLiteVariantStore, VariantRecord
+from ai_service.variants.templates import TEMPLATES, get_template
+
+
+class ScriptedRenderProvider(MockProvider):
+    """Deterministic render: the statement embeds the params (distinct per draw)."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.calls = 0
+
+    async def chat(self, messages, model, response_format=None):  # noqa: ANN001
+        self.calls += 1
+        import json
+
+        user = next(m.content for m in reversed(messages) if m.role == "user")
+        # Distinct per distinct params: hash the seeded slot lines.
+        fingerprint = abs(hash(user)) % 10_000
+        return json.dumps({"statement": f"Scripted variant #{fingerprint} — build it."})
+
+
+class FailingRenderProvider(MockProvider):
+    """Always fails D-020 validation -> deterministic fallback path."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.calls = 0
+
+    async def chat(self, messages, model, response_format=None):  # noqa: ANN001
+        self.calls += 1
+        return "this is not json at all"
+
+
+@pytest.fixture()
+def store(tmp_path):  # noqa: ANN001
+    s = SQLiteVariantStore(db_path=tmp_path / "variants.db")
+    yield s
+    s.close()
+
+
+async def test_two_learners_distinct_statements(store) -> None:  # noqa: ANN001
+    provider = ScriptedRenderProvider()
+    gen = VariantGenerator(store, provider, model="mock")
+    a = await gen.generate("learner-a", "tpl-llm-judge")
+    b = await gen.generate("learner-b", "tpl-llm-judge")
+    assert a.statement != b.statement
+    assert a.seed != b.seed
+    assert a.task_id != b.task_id
+
+
+async def test_same_learner_is_cached_no_second_llm_call(store) -> None:  # noqa: ANN001
+    provider = ScriptedRenderProvider()
+    gen = VariantGenerator(store, provider, model="mock")
+    first = await gen.generate("learner-a", "tpl-llm-judge")
+    calls_after_first = provider.calls
+    second = await gen.generate("learner-a", "tpl-llm-judge")
+    assert first == second  # identical stored variant (D-029 reproducible)
+    assert provider.calls == calls_after_first  # cache hit: NO LLM call
+
+
+async def test_seed_derivation_reproducible() -> None:
+    s1 = derive_seed("tpl-llm-judge", "learner-a", MILESTONE)
+    s2 = derive_seed("tpl-llm-judge", "learner-a", MILESTONE)
+    assert s1 == s2
+    assert derive_seed("tpl-llm-judge", "learner-b", MILESTONE) != s1
+    assert derive_task_id(s1).startswith("task-")
+    assert len(derive_task_id(s1)) == len("task-") + 16
+
+
+async def test_params_are_schema_valid(store) -> None:  # noqa: ANN001
+    gen = VariantGenerator(store, ScriptedRenderProvider(), model="mock")
+    record = await gen.generate("learner-a", "tpl-guardrail-schema")
+    template = get_template("tpl-guardrail-schema")
+    for slot in template.slots:  # type: ignore[union-attr]
+        value = record.params[slot.name]
+        assert slot.validate_value(value), f"slot {slot.name} drew invalid {value!r}"
+
+
+async def test_llm_failure_falls_back_deterministically(store) -> None:  # noqa: ANN001
+    provider = FailingRenderProvider()
+    gen = VariantGenerator(store, provider, model="mock")
+    record = await gen.generate("learner-a", "tpl-rag-chunker")
+    template = get_template("tpl-rag-chunker")
+    expected = template.render({k: v for k, v in record.params.items()})  # type: ignore
+    assert record.statement == expected  # skeleton render, seed-auditable
+    assert provider.calls == 2  # D-020 bounded retry, then fallback
+
+
+async def test_unknown_template_raises(store) -> None:  # noqa: ANN001
+    gen = VariantGenerator(store, ScriptedRenderProvider(), model="mock")
+    with pytest.raises(ValueError, match="no task template"):
+        await gen.generate("learner-a", "tpl-does-not-exist")
+
+
+def test_fairness_envelope_same_bar_per_template() -> None:
+    """a-5 (BINDING): every legal variant of one template fits the anchors.
+
+    For 10 different learners: draw the seeded params, then synthesize a
+    trace whose edit count is sampled INSIDE the template's anchor band and
+    whose test runs meet the anchor minimum — the resulting digests must
+    all sit within the template's expected feature envelope. That is the
+    testable form of "same bar": no slot draw can push a variant outside
+    the effort band the grader context assumes.
+    """
+    import random
+
+    for template in TEMPLATES.values():
+        anchors = template.rubric_anchors
+        lo_edits, hi_edits = anchors.expected_edit_count_band
+        t0 = datetime(2026, 9, 12, tzinfo=UTC)
+        for i in range(10):
+            params = template.sample_params(seed=10_000 + i)
+            rng = random.Random(i)
+            n_edits = rng.randint(lo_edits, hi_edits)
+            events = [
+                TelemetryEvent(
+                    learner_id=f"fair-learner-{i}",
+                    task_id=f"fair-task-{i}",
+                    seq=n,
+                    kind="file_diff",
+                    payload={"path": f"f{n}.py"},
+                    ts=t0 + timedelta(seconds=n * 10),
+                    sandbox_id="sbx-fair",
+                )
+                for n in range(n_edits)
+            ]
+            # Meet the anchor's minimum test-run expectation.
+            for t in range(anchors.expected_min_test_runs):
+                events.append(
+                    TelemetryEvent(
+                        learner_id=f"fair-learner-{i}",
+                        task_id=f"fair-task-{i}",
+                        seq=len(events),
+                        kind="test_result",
+                        payload={"passed": t == anchors.expected_min_test_runs - 1},
+                        ts=t0 + timedelta(seconds=(n_edits + t) * 10),
+                        sandbox_id="sbx-fair",
+                    )
+                )
+            digest = compute_digest(events)
+            assert lo_edits <= digest.edit_count <= hi_edits
+            assert digest.test_pass_count + digest.test_fail_count >= (
+                anchors.expected_min_test_runs
+            )
+            # Slot values never appear in the digest (no scenario leakage into
+            # grading features — difficulty stays scenario-independent).
+            digest_json = digest.model_dump_json()
+            for value in params.values():
+                assert str(value) not in digest_json or isinstance(value, int)
+
+
+async def test_variant_record_roundtrips_through_store(store) -> None:  # noqa: ANN001
+    gen = VariantGenerator(store, ScriptedRenderProvider(), model="mock")
+    record = await gen.generate("learner-a", "tpl-llm-judge")
+    fetched = store.get("learner-a", "tpl-llm-judge")
+    assert fetched is not None
+    assert fetched.statement == record.statement
+    assert fetched.seed == record.seed
+    by_task = store.get_by_task(record.task_id)
+    assert by_task is not None
+    assert by_task.learner_id == "learner-a"
+    assert isinstance(record, VariantRecord)

apps/ai-service/tests/variants/test_store.py

@@ -0,0 +1,358 @@
+"""SQLiteVariantStore tests (REQ-3-005, D-027).
+
+Each test gets its own tmp-path SQLite file — no shared disk state. Covers:
+    - save / get roundtrip by (learner_id, template_id) AND by task_id
+      (all fields survive, including nested JSON params, starter_files
+      filename->content map, and the tz-aware created_at contract)
+    - insert-only: a second save for the same (learner_id, template_id)
+      raises IntegrityError (first-wins, documented choice); the original
+      row is untouched — NOT upsert, NOT swallowed
+    - unique task_id: a second variant claiming an existing trace key is
+      rejected even under a different (learner, template) pair
+    - list_for_learner / list_by_template scoped + chronological + auditable
+      (seed + params readable back — the proctoring cross-check path)
+    - unknown learner / template / task -> None / empty lists
+    - scoping: rows for other learners/templates never leak
+    - rows are detached: usable after the store is closed
+    - WAL + synchronous=NORMAL pragmas actually applied to the DB file
+    - concurrent writer + reader against the same DB file (a-3 smoke test)
+"""
+
+import concurrent.futures
+import threading
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import pytest
+import sqlalchemy as sa
+
+from ai_service.variants.store import SQLiteVariantStore, VariantRecord
+
+_BASE_TS = datetime(2026, 9, 12, 12, 0, 0, tzinfo=UTC)
+
+
+def make_variant(
+    task_id: str = "task-1",
+    learner_id: str = "learner-1",
+    template_id: str = "template-1",
+    seed: str = "seed-a1b2c3",
+    params: dict[str, Any] | None = None,
+    statement: str | None = None,
+    starter_files: dict[str, Any] | None = None,
+    created_at: datetime | None = None,
+) -> VariantRecord:
+    """Canonical kwargs builder — tests override only what they assert on."""
+    return VariantRecord(
+        learner_id=learner_id,
+        template_id=template_id,
+        task_id=task_id,
+        seed=seed,
+        params=params
+        if params is not None
+        else {
+            "scenario": "cache invalidation",
+            "constraints": ["no external deps", "streaming"],
+            "data_shape": {"rows": 10_000, "columns": ["ts", "event", "user"]},
+        },
+        statement=statement
+        if statement is not None
+        else (
+            "Implement a cache layer for the event stream service that "
+            "survives restarts without losing buffered rows, learner "
+            f"{learner_id} edition."
+        ),
+        starter_files=starter_files
+        if starter_files is not None
+        else {
+            "src/stream_cache.py": "class StreamCache:\n    pass\n",
+            "tests/test_stream_cache.py": "def test_roundtrip():\n    pass\n",
+        },
+        created_at=created_at if created_at is not None else _BASE_TS,
+    )
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> SQLiteVariantStore:
+    s = SQLiteVariantStore(db_path=tmp_path / "variants.db")
+    yield s
+    s.close()
+
+
+def test_save_and_get_roundtrip(store: SQLiteVariantStore) -> None:
+    variant = make_variant()
+    store.save(variant)
+
+    # Point lookup by variant identity (learner_id, template_id).
+    fetched = store.get("learner-1", "template-1")
+    assert fetched is not None
+    assert fetched.learner_id == "learner-1"
+    assert fetched.template_id == "template-1"
+    assert fetched.task_id == "task-1"
+    assert fetched.seed == "seed-a1b2c3"
+    assert fetched.params == variant.params
+    assert fetched.statement == variant.statement
+    assert fetched.starter_files == variant.starter_files
+    assert fetched.created_at == _BASE_TS
+    assert fetched.created_at.tzinfo is UTC  # tz-normalized on read
+
+    # Same row via the grading/telemetry trace key — the join path a
+    # grade or proctor check uses with only a task_id in hand.
+    by_task = store.get_by_task("task-1")
+    assert by_task is not None
+    assert by_task.learner_id == "learner-1"
+    assert by_task.template_id == "template-1"
+    assert by_task.seed == "seed-a1b2c3"
+    assert by_task.statement == variant.statement
+    assert by_task.starter_files == variant.starter_files
+    assert by_task.params == variant.params
+    assert by_task.created_at.tzinfo is UTC
+
+
+def test_get_unknown_returns_none(store: SQLiteVariantStore) -> None:
+    store.save(make_variant())
+
+    assert store.get("learner-1", "template-missing") is None
+    assert store.get("learner-missing", "template-1") is None
+    assert store.get("nobody", "nothing") is None
+    assert store.get_by_task("task-missing") is None
+
+
+def test_duplicate_learner_template_pair_raises_integrity_error(
+    store: SQLiteVariantStore,
+) -> None:
+    """THE contract of this store (insert-only, first wins).
+
+    The first generated variant is authoritative (D-029 reproducibility:
+    the seed re-derives the same variant); a duplicate save is a
+    programming error or lost race, not a normal flow — the generator's
+    cache path serves `get` instead. So the IntegrityError surfaces to
+    the caller and the stored row is left untouched.
+    """
+    first = make_variant(statement="first authoritative statement")
+    store.save(first)
+
+    second = make_variant(
+        task_id="task-2",  # distinct task key; the PAIR collides
+        seed="seed-999",
+        statement="an impostor statement",
+        created_at=_BASE_TS + timedelta(hours=1),
+    )
+    with pytest.raises(sa.exc.IntegrityError):
+        store.save(second)
+
+    # First save survived intact — nothing was overwritten.
+    fetched = store.get("learner-1", "template-1")
+    assert fetched is not None
+    assert fetched.task_id == "task-1"
+    assert fetched.statement == "first authoritative statement"
+    assert fetched.seed == "seed-a1b2c3"
+    assert fetched.created_at == _BASE_TS
+
+    # The rejected save left no row behind under its task_id either.
+    assert store.get_by_task("task-2") is None
+
+
+def test_duplicate_task_id_raises_integrity_error(store: SQLiteVariantStore) -> None:
+    # task_id is the grading/telemetry trace key — globally unique: a
+    # second variant may never claim an existing trace key, even under a
+    # different (learner_id, template_id) pair.
+    store.save(make_variant(learner_id="learner-1", template_id="template-1"))
+
+    with pytest.raises(sa.exc.IntegrityError):
+        store.save(
+            make_variant(
+                learner_id="learner-2",
+                template_id="template-2",
+                task_id="task-1",  # collides with learner-1's trace key
+            )
+        )
+
+    # Rejected row not partially stored under either identity.
+    assert store.get("learner-2", "template-2") is None
+    assert len(store.list_by_template("template-2")) == 0
+
+
+def test_list_for_learner_roundtrip_and_audit(store: SQLiteVariantStore) -> None:
+    # Created out of insertion order; list must come back chronological.
+    store.save(make_variant(template_id="template-c", task_id="task-c",
+                            created_at=_BASE_TS + timedelta(hours=2)))
+    store.save(make_variant(template_id="template-a", task_id="task-a",
+                            created_at=_BASE_TS))
+    store.save(make_variant(template_id="template-b", task_id="task-b",
+                            created_at=_BASE_TS + timedelta(hours=1)))
+
+    variants = store.list_for_learner("learner-1")
+    assert [v.template_id for v in variants] == [
+        "template-a",
+        "template-b",
+        "template-c",
+    ]
+    assert all(v.learner_id == "learner-1" for v in variants)
+    hours = (timedelta(hours=0), timedelta(hours=1), timedelta(hours=2))
+    assert all(
+        v.created_at == _BASE_TS + offset
+        for v, offset in zip(variants, hours, strict=True)
+    )
+    assert all(v.created_at.tzinfo is UTC for v in variants)
+
+    # Auditable: every stored variant reads back its seed and typed params
+    # (the proctoring cross-check path reads exactly this).
+    for v in variants:
+        assert v.seed.startswith("seed-")
+        assert v.params["scenario"] == "cache invalidation"
+        assert "streaming" in v.params["constraints"]
+        assert v.params["data_shape"]["rows"] == 10_000
+        assert v.starter_files["tests/test_stream_cache.py"].count("\n") >= 1
+
+
+def test_list_by_template_roundtrip_and_audit(store: SQLiteVariantStore) -> None:
+    # Three learners on the same template: distinct, auditable variants.
+    store.save(make_variant(learner_id="learner-b", task_id="task-b",
+                            seed="seed-222",
+                            created_at=_BASE_TS + timedelta(hours=1)))
+    store.save(make_variant(learner_id="learner-a", task_id="task-a",
+                            seed="seed-111", created_at=_BASE_TS))
+    store.save(make_variant(learner_id="learner-c", task_id="task-c",
+                            seed="seed-333",
+                            created_at=_BASE_TS + timedelta(hours=2)))
+
+    variants = store.list_by_template("template-1")
+    assert [v.learner_id for v in variants] == ["learner-a", "learner-b", "learner-c"]
+    assert all(v.template_id == "template-1" for v in variants)
+
+    # Every learner's variant carries its own seed + params (auditable,
+    # REQ-3-005: variant parameters persisted and auditable).
+    seeds = {v.learner_id: v.seed for v in variants}
+    assert seeds == {
+        "learner-a": "seed-111",
+        "learner-b": "seed-222",
+        "learner-c": "seed-333",
+    }
+    assert all(v.params["scenario"] == "cache invalidation" for v in variants)
+    assert all(v.created_at.tzinfo is UTC for v in variants)
+
+
+def test_list_unknown_returns_empty_lists(store: SQLiteVariantStore) -> None:
+    store.save(make_variant())
+
+    assert store.list_for_learner("nobody") == []
+    assert store.list_by_template("no-template") == []
+
+
+def test_lists_are_scoped(store: SQLiteVariantStore) -> None:
+    store.save(make_variant(learner_id="learner-1", template_id="template-1",
+                            task_id="task-1"))
+    store.save(make_variant(learner_id="learner-2", template_id="template-1",
+                            task_id="task-2"))
+    store.save(make_variant(learner_id="learner-1", template_id="template-2",
+                            task_id="task-3"))
+
+    # learner lists see only that learner's rows.
+    assert [v.template_id for v in store.list_for_learner("learner-1")] == [
+        "template-1",
+        "template-2",
+    ]
+    assert [v.template_id for v in store.list_for_learner("learner-2")] == ["template-1"]
+
+    # template lists see one row per learner, none from other templates.
+    template_rows = store.list_by_template("template-1")
+    assert sorted(v.learner_id for v in template_rows) == ["learner-1", "learner-2"]
+    assert all(v.template_id == "template-1" for v in template_rows)
+
+    # get stays a pair-scoped point lookup: same template, other learner.
+    assert store.get("learner-1", "template-2") is not None
+    assert store.get("learner-2", "template-2") is None
+
+
+def test_empty_params_and_starter_files_roundtrip(store: SQLiteVariantStore) -> None:
+    # Legal shapes: a slotless template carries no params; a variant may
+    # ship without a workspace scaffold.
+    store.save(make_variant(params={}, starter_files={}))
+
+    fetched = store.get("learner-1", "template-1")
+    assert fetched is not None
+    assert fetched.params == {}
+    assert fetched.starter_files == {}
+
+
+def test_rows_are_detached_after_save(store: SQLiteVariantStore, tmp_path: Path) -> None:
+    # The API layer hands VariantRecords across layers; rows must survive
+    # the store that produced them being closed (no open-session ORM magic).
+    store.save(make_variant())
+    fetched = store.get("learner-1", "template-1")
+    by_task = store.get_by_task("task-1")
+    store.close()
+
+    assert fetched is not None
+    assert fetched.statement == fetched.statement  # usable post-close
+    assert fetched.starter_files["src/stream_cache.py"] == "class StreamCache:\n    pass\n"
+    assert by_task is not None
+    assert by_task.seed == "seed-a1b2c3"
+
+    # A fresh store on the same file sees the same row (durability).
+    reopened = SQLiteVariantStore(db_path=tmp_path / "variants.db")
+    try:
+        again = reopened.get("learner-1", "template-1")
+        assert again is not None
+        assert again.starter_files["src/stream_cache.py"].startswith("class StreamCache")
+        assert again.created_at.tzinfo is UTC
+    finally:
+        reopened.close()
+
+
+def test_pragmas_are_applied(store: SQLiteVariantStore) -> None:
+    # Pragmas are per-connection; query through the store's engine so the
+    # connect hook (not a default sqlite3 connection) is what we inspect.
+    with store._engine.connect() as conn:
+        (journal_mode,) = conn.execute(sa.text("PRAGMA journal_mode")).one()
+        (synchronous,) = conn.execute(sa.text("PRAGMA synchronous")).one()
+    assert journal_mode == "wal"
+    # synchronous=NORMAL is 1 in SQLite's pragma numbering.
+    assert synchronous == 1
+
+
+def test_concurrent_writer_and_reader_no_database_is_locked(tmp_path: Path) -> None:
+    """One thread saves while another reads in a tight loop (a-3).
+
+    Without WAL + busy_timeout this pattern reliably produces
+    `OperationalError: database is locked` on SQLite. The assertion is
+    that every reader call completes and every distinct-row write lands.
+    """
+    db_path = tmp_path / "variants.db"
+    n_variants = 60  # distinct (learner, template) pairs, one save each
+    stop_writing = threading.Event()
+
+    writer = SQLiteVariantStore(db_path=db_path)
+    reader = SQLiteVariantStore(db_path=db_path)
+    try:
+
+        def write_variants() -> None:
+            for seq in range(n_variants):
+                writer.save(
+                    make_variant(
+                        learner_id=f"learner-{seq}",
+                        template_id="template-1",
+                        task_id=f"task-{seq}",
+                        seed=f"seed-{seq:03d}",
+                        created_at=_BASE_TS + timedelta(seconds=seq),
+                    )
+                )
+            stop_writing.set()
+
+        def read_variants() -> None:
+            while not stop_writing.is_set():
+                reader.list_by_template("template-1")
+            # Final read after the writer is done.
+            assert len(reader.list_by_template("template-1")) == n_variants
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
+            futures = [pool.submit(write_variants), pool.submit(read_variants)]
+            for future in futures:
+                future.result(timeout=30)
+
+        # Every save was a distinct row: nothing lost, none doubled.
+        assert len(writer.list_by_template("template-1")) == n_variants
+    finally:
+        reader.close()
+        writer.close()

apps/ai-service/tests/variants/test_templates.py

@@ -0,0 +1,102 @@
+"""Task template tests (Task 4-1-01, REQ-3-005)."""
+
+from __future__ import annotations
+
+import random
+
+import pytest
+
+from ai_service.variants.templates import (
+    TEMPLATES,
+    ParameterSlot,
+    TaskTemplate,
+    get_template,
+    slots_pattern_ok,
+    template_for_competency,
+    validate_competency_binding,
+)
+
+
+def test_all_templates_bind_to_real_competency_ids() -> None:
+    validate_competency_binding()  # raises on any unknown binding
+    assert len(TEMPLATES) >= 3
+
+
+def test_slot_validation_rejects_bad_values() -> None:
+    slot = ParameterSlot(name="domain", type="enum", values=["a", "b"])
+    assert slot.validate_value("a")
+    assert not slot.validate_value("c")
+    assert not slot.validate_value(3)
+
+    rng = random.Random(42)
+    assert slot.sample(rng) in {"a", "b"}
+
+
+def test_int_range_slot_bounds() -> None:
+    slot = ParameterSlot(name="n", type="int_range", lo=2, hi=4)
+    rng = random.Random(0)
+    for _ in range(20):
+        assert 2 <= slot.sample(rng) <= 4
+    assert slot.validate_value(3)
+    assert not slot.validate_value(5)
+    assert not slot.validate_value("3")
+
+
+def test_seeded_sampling_is_reproducible() -> None:
+    tpl = TEMPLATES["tpl-llm-judge"]
+    first = tpl.sample_params(seed=1234)
+    second = tpl.sample_params(seed=1234)
+    other = tpl.sample_params(seed=1235)
+    assert first == second  # D-029: same seed -> identical params
+    assert first != other  # different seed -> (near-certainly) different draw
+
+
+def test_skeleton_placeholders_match_slots() -> None:
+    for tpl in TEMPLATES.values():
+        assert slots_pattern_ok(tpl.statement_skeleton, tpl.slots), tpl.id
+
+
+def test_render_validates_params_and_fills() -> None:
+    tpl = TEMPLATES["tpl-llm-judge"]
+    params = tpl.sample_params(seed=7)
+    rendered = tpl.render(params)
+    for value in params.values():
+        assert str(value) in rendered
+    with pytest.raises(ValueError, match="invalid value"):
+        tpl.render({**params, "edge_cases": 99})  # out of band
+
+
+def test_rubric_anchors_present_per_template() -> None:
+    for tpl in TEMPLATES.values():
+        anchors = tpl.rubric_anchors
+        assert anchors.expected_edit_count_band[0] <= anchors.expected_edit_count_band[1]
+        assert anchors.expected_min_test_runs >= 1
+        cycles = anchors.expected_error_fix_cycles_band
+        assert cycles[0] <= cycles[1]
+
+
+def test_starter_files_defined_per_template() -> None:
+    for tpl in TEMPLATES.values():
+        assert tpl.starter_files, f"{tpl.id} missing starter scaffolds"
+        assert "README.md" in tpl.starter_files
+        assert tpl.test_command
+
+
+def test_competency_lookup() -> None:
+    tpls = template_for_competency("stack-orchestration-c005")
+    assert len(tpls) == 1
+    assert get_template("nope-xyz") is None
+
+
+def test_bad_skeleton_rejected() -> None:
+    with pytest.raises(ValueError, match="slot"):
+        TaskTemplate(
+            id="tpl-bad",
+            competency_id="stack-orchestration-c001",
+            title="Bad",
+            statement_skeleton="no placeholders at all",
+            slots=[ParameterSlot(name="x", type="enum", values=["a"])],
+            rubric_anchors=TEMPLATES["tpl-llm-judge"].rubric_anchors,
+            starter_files={},
+            test_command="pytest -q",
+        )

packages/types/grading.ts

@@ -0,0 +1,84 @@
+/**
+ * Nextcraft — Grading Types
+ *
+ * Rubric-scored grades over real process traces: per-criterion 0-4
+ * scores, strengths, gaps, and verdict, persisted latest-state per
+ * (learner_id, task_id) with the trace digest that fed the rubric prompt.
+ *
+ * Source of truth: the Python models in
+ * `apps/ai-service/ai_service/grading/engine.py` (`RubricScore`) and
+ * `apps/ai-service/ai_service/grading/store.py` (`GradeRecord`),
+ * served over HTTP by `apps/ai-service/ai_service/api/assessment.py`
+ * (`GradeResponse`). This file mirrors those models field-for-field;
+ * any schema change must be made in both places.
+ */
+
+/**
+ * The graded rubric criteria set (RUBRIC_CRITERIA, grading/engine.py).
+ * Scores are 0-4 per criterion; the exact four keys are required.
+ */
+export interface RubricCriteria {
+  process_quality: number;
+  correctness: number;
+  debugging_discipline: number;
+  test_usage: number;
+}
+
+/** Rubric verdict carried in `RubricScore.verdict`. */
+export type RubricVerdict = 'mastered' | 'developing' | 'not_yet';
+
+/**
+ * Validated LLM rubric output for a GRADED trace. Mirrors `RubricScore`
+ * (ai_service/grading/engine.py): criteria 0-4, strengths/gaps (1-2
+ * sentences each), rubric verdict.
+ */
+export interface RubricScore {
+  criteria: RubricCriteria;
+  strengths: string[];
+  gaps: string[];
+  verdict: RubricVerdict;
+}
+
+/** Machine-readable grade outcome (`GradeRecord.verdict`). */
+export type GradeOutcome =
+  | 'GRADED'
+  | 'UNGRADABLE_TRACE_INCOMPLETE'
+  | 'UNGRADABLE_EMPTY_TRACE';
+
+/**
+ * Gate detail surfaced in `scores` for UNGRADABLE_* records (G-4):
+ * the integrity flag reason and/or missing seq numbers. Never co-present
+ * with rubric scores — a grade carries one or the other.
+ */
+export interface GradeGateDetail {
+  integrity_flag: string | null;
+  missing_seqs: number[];
+}
+
+/** Union of the two `scores` shapes a grade can carry. */
+export type GradeScores = RubricScore | GradeGateDetail;
+
+/**
+ * A persisted grade for one (learner_id, task_id) trace — latest state
+ * (a regrade replaces the row). Mirrors `GradeRecord`
+ * (ai_service/grading/store.py) field-for-field, except `scores` is
+ * typed as the discriminated `GradeScores` union instead of the Python
+ * side's untyped JSON dict.
+ */
+export interface GradeRecord {
+  learner_id: string;
+  /** The graded task — joins to `TaskVariant.task_id`. */
+  task_id: string;
+  /** Task-variant seed (D-029); null while grading is variant-blind. */
+  variant_seed: string | null;
+  /** Compact trace digest (D-028) that fed the rubric prompt; {} for gate records. */
+  digest: Record<string, unknown>;
+  /** Rubric scores (GRADED) or gate detail (UNGRADABLE_*); never both. */
+  scores: GradeScores;
+  /** Machine-readable outcome; rubric verdict rides in `scores.verdict`. */
+  verdict: GradeOutcome;
+  /** Provider model that produced the scores; "none" for gate records. */
+  model: string;
+  /** ISO 8601 UTC grade timestamp; a regrade replaces it. */
+  created_at: string;
+}

packages/types/index.ts

@@ -1,5 +1,7 @@
 export * from './domain';
+export * from './grading';
 export * from './marketplace';
 export * from './telemetry';
 export * from './user';
 export * from './ui';
+export * from './variants';

packages/types/variants.ts

@@ -0,0 +1,50 @@
+/**
+ * Nextcraft — Task Variant Types
+ *
+ * Seeded per-learner task variants: a task template instantiated for one
+ * learner with a reproducible seed, sampled parameter slots, a rendered
+ * statement, and starter-file scaffolds. One (learner_id, template_id)
+ * pair is exactly ONE variant, forever — regeneration is a cache read.
+ *
+ * Source of truth: the Python model in
+ * `apps/ai-service/ai_service/variants/store.py` (`VariantRecord`),
+ * served over HTTP by `apps/ai-service/ai_service/api/variants.py`
+ * (`VariantResponse`, which adds the `competency_id` enrichment). This
+ * file mirrors those fields field-for-field; any schema change must be
+ * made in both places.
+ */
+
+/**
+ * Typed parameter-slot values the generator sampled for the variant's
+ * template (seed-derivable, D-029). Enum/string-set slots draw strings;
+ * int-range slots draw numbers. An empty object is legal (a slotless
+ * template).
+ */
+export interface VariantParams {
+  [slotName: string]: string | number;
+}
+
+/**
+ * A generated task variant. Mirrors `VariantRecord`
+ * (ai_service/variants/store.py) plus the API-layer `competency_id`
+ * enrichment (ai_service/api/variants.py) — the D-021 competency the
+ * variant's template binds to.
+ */
+export interface TaskVariant {
+  learner_id: string;
+  /** Globally unique task key — the grading and telemetry join key. */
+  task_id: string;
+  template_id: string;
+  /** D-021 competency the template binds to (resolved at read time). */
+  competency_id: string;
+  /** Reproducible per-(template, learner, milestone) seed (D-029). */
+  seed: string;
+  /** Seeded slot values the statement was rendered from. */
+  params: VariantParams;
+  /** Rendered task statement shown to the learner. */
+  statement: string;
+  /** Workspace scaffold: filename → file content. */
+  starter_files: Record<string, string>;
+  /** ISO 8601 UTC generation timestamp. */
+  created_at: string;
+}