feat: full feature buildout — streaming, i18n, mastery map, admin, jobs

Progressive lesson streaming via onSegment callback (fixes SSE for non-English
users — locale was shadowed in lesson-reader useEffect). Adds: BullMQ workers,
Redis stream buffer, token budget enforcement, Langfuse tracing, golden-eval
runner, Playwright e2e scaffolding, lesson depth/locale/preferences schema,
mastery map UI, admin panel (blueprints/users/reports/quality/misconceptions),
image queries, source citations, view transitions, reading animations, i18n
(next-intl), PDF export, surprise endpoint, and 402 passing unit tests.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

jobs/bootstrap.ts

@@ -0,0 +1,30 @@
+/**
+ * Worker entrypoint. Stubs the `server-only` import guard before loading any
+ * src/lib code, then hands off to the real worker registry.
+ *
+ * Why: src/lib/** modules begin with `import 'server-only'` to keep RSC-only
+ * code out of client bundles. That package throws when required outside Next's
+ * React Server bundler. Workers are server code but run under plain Node/tsx,
+ * so we resolve `server-only` to a no-op here — and ONLY here, so the app's
+ * build-time guard is untouched. Mirrors the vitest alias in vitest.config.ts.
+ */
+import Module, { createRequire } from 'module';
+import { fileURLToPath } from 'url';
+import { dirname, join } from 'path';
+
+const here = dirname(fileURLToPath(import.meta.url));
+const NOOP = join(here, 'server-only.noop.cjs');
+
+type ResolveFn = (request: string, ...rest: unknown[]) => string;
+const moduleInternals = Module as unknown as { _resolveFilename: ResolveFn };
+const originalResolve = moduleInternals._resolveFilename;
+
+moduleInternals._resolveFilename = function (request, ...rest) {
+  if (request === 'server-only') return NOOP;
+  return originalResolve.call(this, request, ...rest);
+};
+
+// Load the worker graph *after* the stub is installed (require keeps it
+// synchronous — tsx compiles this file to CJS, where top-level await is unavailable).
+const require = createRequire(import.meta.url);
+require('./index');

jobs/generate-lesson-job.ts

@@ -6,13 +6,14 @@
  * On failure: increments attempt count; sets status to 'failed' after processing.
  */
 import { Worker } from 'bullmq';
-import { eq } from 'drizzle-orm';
+import { eq, and } from 'drizzle-orm';
 import { db } from '../src/lib/db';
 import { jobs, blueprints, concepts, lessons } from '../src/lib/db/schema';
 import { generateLesson } from '../src/lib/generation/generate-lesson';
 import { verifyLesson } from '../src/lib/verification/verify-content';
 import { getLessonResponse } from '../src/lib/db/queries';
 import { setCachedLesson } from '../src/lib/cache/lesson';
+import { appendStreamSegment, markStreamDone } from '../src/lib/cache/stream';
 import { enqueuePromoteBlueprint } from '../src/lib/jobs/queue';
 import type { GenerateLessonJobData } from '../src/lib/jobs/queue';
 
@@ -23,7 +24,7 @@ const connection = {
 const worker = new Worker<GenerateLessonJobData>(
   'generate-lesson',
   async (job) => {
-    const { intentKey, blueprintId, idempotencyKey } = job.data;
+    const { intentKey, blueprintId, idempotencyKey, locale = 'en', depth = 'standard', ageGroup = 'adult', difficultyLevel = 1, ord = 0 } = job.data;
 
     // 1. Idempotency check — find the job row by idempotency key
     const [jobRow] = await db
@@ -34,9 +35,9 @@ const worker = new Worker<GenerateLessonJobData>(
 
     if (jobRow?.status === 'done') {
       // Already completed — warm cache if needed and return
-      const cached = await getLessonResponse(intentKey);
+      const cached = await getLessonResponse(intentKey, undefined, locale, depth as import('../src/lib/generation/depth').LessonDepth, ageGroup as import('../src/schemas/preferences').AgeGroup);
       if (cached?.state === 'ready') {
-        await setCachedLesson(intentKey, cached);
+        await setCachedLesson(intentKey, locale, cached, depth as import('../src/lib/generation/depth').LessonDepth, ageGroup as import('../src/schemas/preferences').AgeGroup);
       }
       return;
     }
@@ -47,15 +48,31 @@ const worker = new Worker<GenerateLessonJobData>(
       .set({ status: 'running', attempts: (jobRow?.attempts ?? 0) + 1 })
       .where(eq(jobs.id, idempotencyKey));
 
-    // 2. Check lesson doesn't already exist
-    const existing = await getLessonResponse(intentKey);
-    if (existing?.state === 'ready') {
-      await setCachedLesson(intentKey, existing);
+    // 2. Check lesson at this ord doesn't already exist
+    const [existingLesson] = await db
+      .select({ id: lessons.id })
+      .from(lessons)
+      .where(
+        and(
+          eq(lessons.blueprintId, blueprintId),
+          eq(lessons.locale, locale),
+          eq(lessons.depth, depth),
+          eq(lessons.ageGroup, ageGroup),
+          eq(lessons.ord, ord),
+        ),
+      )
+      .limit(1);
+
+    if (existingLesson) {
+      const cached = await getLessonResponse(intentKey, undefined, locale, depth as import('../src/lib/generation/depth').LessonDepth, ageGroup as import('../src/schemas/preferences').AgeGroup);
+      if (cached?.state === 'ready') {
+        await setCachedLesson(intentKey, locale, cached, depth as import('../src/lib/generation/depth').LessonDepth, ageGroup as import('../src/schemas/preferences').AgeGroup);
+      }
       await db.update(jobs).set({ status: 'done' }).where(eq(jobs.id, idempotencyKey));
       return;
     }
 
-    // 3. Fetch blueprint + concepts
+    // 3. Fetch blueprint + concepts at this ord
     const [blueprint] = await db
       .select()
       .from(blueprints)
@@ -64,32 +81,57 @@ const worker = new Worker<GenerateLessonJobData>(
 
     if (!blueprint) throw new Error(`Blueprint not found: ${blueprintId}`);
 
+    // For ord=0 use all concepts (backward compat); for ord>0 use concepts at that ord.
     const conceptRows = await db
       .select({ id: concepts.id, name: concepts.name, ord: concepts.ord, retrievalCtxRef: concepts.retrievalCtxRef })
       .from(concepts)
-      .where(eq(concepts.blueprintId, blueprintId));
+      .where(
+        ord === 0
+          ? eq(concepts.blueprintId, blueprintId)
+          : and(eq(concepts.blueprintId, blueprintId), eq(concepts.ord, ord)),
+      );
 
-    if (conceptRows.length === 0) throw new Error(`No concepts for blueprint: ${blueprintId}`);
+    if (conceptRows.length === 0) throw new Error(`No concepts at ord=${ord} for blueprint: ${blueprintId}`);
 
-    // 4. Generate lesson
+    // 4. Generate lesson — onSegment pushes each segment to Redis as it's persisted,
+    //    so SSE clients receive progressive delivery without waiting for the full batch.
     const { lessonId } = await generateLesson({
       blueprintId,
-      lessonOrd: 0,
+      lessonOrd: ord,
       topicTitle: blueprint.title,
       conceptsToGenerate: conceptRows.map((c) => ({
         id: c.id,
         name: c.name,
         retrievalCtxRef: c.retrievalCtxRef,
       })),
+      locale,
+      depth: depth as import('../src/lib/generation/depth').LessonDepth,
+      ageGroup: ageGroup as import('../src/schemas/preferences').AgeGroup,
+      difficultyLevel: difficultyLevel as 1 | 2 | 3,
+      onSegment: (seg) =>
+        appendStreamSegment(
+          intentKey, locale, seg,
+          depth as import('../src/lib/generation/depth').LessonDepth,
+          ageGroup as import('../src/schemas/preferences').AgeGroup,
+          ord,
+        ),
     });
 
+    // Signal SSE clients that the stream is complete.
+    await markStreamDone(
+      intentKey, locale,
+      depth as import('../src/lib/generation/depth').LessonDepth,
+      ageGroup as import('../src/schemas/preferences').AgeGroup,
+      ord,
+    );
+
     // 5. T1 verification (non-blocking for serve — T2 + promotion handled by promote-blueprint job)
     await verifyLesson({ lessonId, runT2: false });
 
-    // 6. Write to Redis cache
-    const lesson = await getLessonResponse(intentKey);
+    // 6. Write to Redis cache (position-keyed)
+    const lesson = await getLessonResponse(intentKey, undefined, locale, depth as import('../src/lib/generation/depth').LessonDepth, ageGroup as import('../src/schemas/preferences').AgeGroup);
     if (lesson?.state === 'ready') {
-      await setCachedLesson(intentKey, lesson);
+      await setCachedLesson(intentKey, locale, lesson, depth as import('../src/lib/generation/depth').LessonDepth, ageGroup as import('../src/schemas/preferences').AgeGroup, ord);
     }
 
     // 7. Update job ledger

jobs/index.ts

@@ -1,13 +1,15 @@
 import generateLessonWorker from './generate-lesson-job';
 import promoteBlueprintWorker from './promote-blueprint-job';
+import regenerateBlueprintWorker from './regenerate-blueprint-job';
 
-console.log('[workers] Running: generate-lesson, promote-blueprint');
+console.log('[workers] Running: generate-lesson, promote-blueprint, regenerate-blueprint');
 
 async function shutdown(signal: string) {
   console.log(`[workers] ${signal} received — draining and closing workers...`);
   await Promise.all([
     generateLessonWorker.close(),
     promoteBlueprintWorker.close(),
+    regenerateBlueprintWorker.close(),
   ]);
   console.log('[workers] Shutdown complete.');
   process.exit(0);

jobs/promote-blueprint-job.ts

@@ -100,7 +100,7 @@ const worker = new Worker<PromoteBlueprintJobData>(
     }
 
     // 6. Invalidate cache so next request fetches fresh lesson with updated verify status
-    await invalidateCachedLesson(intentKey);
+    await invalidateCachedLesson(intentKey, 'en');
 
     await db.update(jobs).set({ status: 'done' }).where(eq(jobs.id, idempotencyKey));
   },

jobs/regenerate-blueprint-job.ts

@@ -0,0 +1,165 @@
+/**
+ * BullMQ worker: regenerate all lessons for a stale blueprint at a new contentVersion.
+ *
+ * Idempotent: idempotency key encodes `blueprintId:targetContentVersion` so
+ * retries never double-bill. Serves old content until regeneration completes.
+ */
+import { Worker } from 'bullmq';
+import { eq, and } from 'drizzle-orm';
+import { db } from '../src/lib/db';
+import { jobs, blueprints, concepts, lessons } from '../src/lib/db/schema';
+import { generateLesson } from '../src/lib/generation/generate-lesson';
+import { verifyLesson } from '../src/lib/verification/verify-content';
+import { clearBlueprintStale } from '../src/lib/generation/staleness';
+import { getLessonResponse } from '../src/lib/db/queries';
+import { setCachedLesson } from '../src/lib/cache/lesson';
+import { enqueuePromoteBlueprint } from '../src/lib/jobs/queue';
+import type { RegenerateBlueprintJobData } from '../src/lib/jobs/queue';
+
+const connection = {
+  url: process.env.REDIS_URL ?? 'redis://localhost:6379',
+};
+
+const worker = new Worker<RegenerateBlueprintJobData>(
+  'regenerate-blueprint',
+  async (job) => {
+    const { blueprintId, intentKey, targetContentVersion, idempotencyKey } = job.data;
+
+    // 1. Idempotency check
+    const [jobRow] = await db
+      .select({ id: jobs.id, status: jobs.status, attempts: jobs.attempts })
+      .from(jobs)
+      .where(eq(jobs.id, idempotencyKey))
+      .limit(1);
+
+    if (jobRow?.status === 'done') return;
+
+    await db
+      .update(jobs)
+      .set({ status: 'running', attempts: (jobRow?.attempts ?? 0) + 1 })
+      .where(eq(jobs.id, idempotencyKey));
+
+    // 2. Verify the blueprint still needs this version (may have been superseded)
+    const [bp] = await db
+      .select({ id: blueprints.id, contentVersion: blueprints.contentVersion, title: blueprints.title })
+      .from(blueprints)
+      .where(eq(blueprints.id, blueprintId))
+      .limit(1);
+
+    if (!bp) throw new Error(`Blueprint not found: ${blueprintId}`);
+
+    // If the blueprint has already been rebuilt at or past targetContentVersion, skip.
+    if (bp.contentVersion > targetContentVersion) {
+      await db.update(jobs).set({ status: 'done' }).where(eq(jobs.id, idempotencyKey));
+      return;
+    }
+
+    // 3. Load concepts
+    const conceptRows = await db
+      .select({ id: concepts.id, name: concepts.name, ord: concepts.ord, retrievalCtxRef: concepts.retrievalCtxRef })
+      .from(concepts)
+      .where(eq(concepts.blueprintId, blueprintId));
+
+    if (conceptRows.length === 0) throw new Error(`No concepts for blueprint: ${blueprintId}`);
+
+    // 4. Find all unique (locale, depth, ageGroup) variants already generated for this blueprint
+    //    so we regenerate the same set.
+    const existingVariants = await db
+      .select({ locale: lessons.locale, depth: lessons.depth, ageGroup: lessons.ageGroup })
+      .from(lessons)
+      .where(eq(lessons.blueprintId, blueprintId));
+
+    // Dedupe variants; always include default.
+    const variantSet = new Map<string, { locale: string; depth: string; ageGroup: string }>();
+    variantSet.set('en:standard:adult', { locale: 'en', depth: 'standard', ageGroup: 'adult' });
+    for (const v of existingVariants) {
+      variantSet.set(`${v.locale}:${v.depth}:${v.ageGroup}`, v);
+    }
+
+    // 5. Regenerate each variant
+    for (const variant of variantSet.values()) {
+      const { locale, depth, ageGroup } = variant;
+      const { lessonId } = await generateLesson({
+        blueprintId,
+        lessonOrd: 0,
+        topicTitle: bp.title,
+        conceptsToGenerate: conceptRows.map((c) => ({
+          id: c.id,
+          name: c.name,
+          retrievalCtxRef: c.retrievalCtxRef,
+        })),
+        locale,
+        depth: depth as import('../src/lib/generation/depth').LessonDepth,
+        ageGroup: ageGroup as import('../src/schemas/preferences').AgeGroup,
+      });
+
+      // T1 verify (T2 handled by promote job)
+      await verifyLesson({ lessonId, runT2: false });
+
+      // Update cache
+      const lesson = await getLessonResponse(
+        intentKey, undefined, locale,
+        depth as import('../src/lib/generation/depth').LessonDepth,
+        ageGroup as import('../src/schemas/preferences').AgeGroup,
+      );
+      if (lesson?.state === 'ready') {
+        await setCachedLesson(
+          intentKey, locale, lesson,
+          depth as import('../src/lib/generation/depth').LessonDepth,
+          ageGroup as import('../src/schemas/preferences').AgeGroup,
+        );
+      }
+    }
+
+    // 6. Clear stale flag + stamp current signature
+    await clearBlueprintStale(blueprintId);
+
+    // 7. Bump contentVersion to targetContentVersion (done after generation succeeds)
+    await db
+      .update(blueprints)
+      .set({ contentVersion: targetContentVersion })
+      .where(and(eq(blueprints.id, blueprintId)));
+
+    // 8. Mark job done
+    await db.update(jobs).set({ status: 'done' }).where(eq(jobs.id, idempotencyKey));
+
+    // 9. Enqueue promotion (T2 + misconceptions)
+    const promoteKey = `promote-blueprint:${blueprintId}:v${targetContentVersion}`;
+    const [existingPromote] = await db
+      .select({ id: jobs.id })
+      .from(jobs)
+      .where(eq(jobs.id, promoteKey))
+      .limit(1);
+
+    if (!existingPromote) {
+      const [promoteJobRow] = await db
+        .insert(jobs)
+        .values({
+          type: 'promote-blueprint',
+          idempotencyKey: promoteKey,
+          status: 'pending',
+          payloadJson: { blueprintId, intentKey },
+        })
+        .returning({ id: jobs.id });
+
+      await enqueuePromoteBlueprint({
+        blueprintId,
+        intentKey,
+        idempotencyKey: promoteJobRow.id,
+      });
+    }
+  },
+  { connection, concurrency: 1 },
+);
+
+worker.on('failed', async (job, err) => {
+  if (job?.data.idempotencyKey) {
+    await db
+      .update(jobs)
+      .set({ status: 'failed' })
+      .where(eq(jobs.id, job.data.idempotencyKey));
+  }
+  console.error('[regenerate-blueprint] job failed:', err);
+});
+
+export default worker;

jobs/server-only.noop.cjs

@@ -0,0 +1,5 @@
+// Empty stand-in for the `server-only` package inside the worker runtime.
+// `server-only` throws when imported outside Next's RSC bundler; BullMQ workers
+// are legitimate server code running under plain Node/tsx, so we neutralize it.
+// Scoped to the worker process via jobs/bootstrap.ts — Next keeps the real guard.
+module.exports = {};