Your own lesson pipeline: the spine

import Fastify from "fastify";
import { producer } from "../stream/producer.js";
import { scopeFromClaims } from "../security/scope.js";
import type { LessonRequest } from "../contracts/request.js";

const app = Fastify({ logger: true });

app.post("/requests", async (req, reply) => {
  const scope = scopeFromClaims(req.user);   // district + env, from the token, never the body
  const body = req.body as Partial<LessonRequest>;
  if (!body.standardCode || !body.gradeBand) return reply.code(400).send({ ok: false });

  const request: LessonRequest = {
    requestId: `req_${randomId()}`,
    districtId: scope.districtId,
    env: scope.env,
    standardCode: body.standardCode,               // "3.OA.A.1"
    frameworkId: scope.frameworkId,                // pinned per district
    frameworkVersion: scope.frameworkVersion,      // "2019.1" — revisions change the claim
    gradeBand: body.gradeBand,                     // "3"
    readingLevelTarget: body.readingLevelTarget ?? null,  // 1.0 on grade; 0.6 two years below
    locale: body.locale ?? "en-US",
    requestedBy: scope.userId,
    requestedAt: new Date().toISOString(),         // server-stamped, never client-trusted
  };

  await producer.send({
    topic: "lesson.requests",
    messages: [{ key: `${scope.districtId}:${request.requestId}`, value: JSON.stringify(request) }],
  });
  return reply.code(202).send({ ok: true, requestId: request.requestId });
});

export interface ContentBlock {
  blockHash: string;          // sha256 of (kind, body, assets) — its identity IS its content
  kind: "explanation" | "worked_example" | "practice_item" | "image" | "interactive";
  body: string;               // markdown or a structured item, per kind
  assetIds: string[];         // object-store keys; images carry their own rights record
  provenance: BlockProvenance;
}

export interface BlockProvenance {
  requestId: string;          // "req_5510"
  objectiveId: string;        // "obj_7d31" — what this block is FOR, not just what it is about
  generator: "model" | "human" | "imported";
  modelVersion: string | null;      // null when a human wrote it
  promptHash: string | null;        // the exact prompt, hashed — re-derivable, not guessable
  retrievalSnapshot: string | null; // ids + versions of every source passage used
  createdAt: string;
}

"""Consumes lesson.requests, emits lesson.blocks. Stateless; safe to run N of these."""
from lessonpipe.contracts import LessonRequest, ContentBlock, BlockProvenance
from lessonpipe.graph import objectives_for
from lessonpipe.retrieval import approved_sources
from lessonpipe.stream import consume, publish
from lessonpipe.hashing import block_hash, prompt_hash


def handle(request: LessonRequest) -> None:
    # Objectives come from the content graph, not from the model. Post 2 owns this edge.
    for objective in objectives_for(request.standard_code, request.framework_version):
        sources = approved_sources(
            district_id=request.district_id,      # retrieval is scoped, always
            framework_version=request.framework_version,
            objective_id=objective.id,
        )
        prompt = render_prompt(objective, sources, request.grade_band, request.reading_level_target)
        for draft in call_model(prompt):          # retries and rate limits live here, not in the API
            block = ContentBlock(
                block_hash=block_hash(draft.kind, draft.body, draft.asset_ids),
                kind=draft.kind,
                body=draft.body,
                asset_ids=draft.asset_ids,
                provenance=BlockProvenance(
                    request_id=request.request_id,
                    objective_id=objective.id,
                    generator="model",
                    model_version=draft.model_version,
                    prompt_hash=prompt_hash(prompt),
                    retrieval_snapshot=[s.version_key for s in sources],
                    created_at=now_iso(),
                ),
            )
            publish("lesson.blocks", key=f"{request.district_id}:{block.block_hash}", value=block)


consume("lesson.requests", handle)

export interface TenantScope {
  districtId: string;         // "riverbend"
  env: "dev" | "prod";
}

/** The one place a scope becomes a string. Schema names, topic keys, object prefixes. */
export function namespace(scope: TenantScope): string {
  return `${scope.districtId}__${scope.env}`;
}

export function requestKey(scope: TenantScope, requestId: string): string {
  return `${scope.districtId}:${requestId}`;
}

export interface LessonVersion {
  lessonId: string;            // "lsn_4c19" — stable across versions
  version: number;             // 1, 2, 3 — monotonic per lesson
  districtId: string;
  env: "dev" | "prod";
  standardCode: string;        // "3.OA.A.1"
  frameworkVersion: string;    // "2019.1"
  blockHashes: string[];       // ordered; the lesson IS this list
  derivedFrom: number | null;  // the version this one was edited from
  createdBy: string;           // "usr_2210" — a teacher, or "system"
  createdAt: string;
}