Implement P0 slide workflow foundation

.gitignore

@@ -0,0 +1,5 @@
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.ruff_cache/
+.venv/

README.md

@@ -12,4 +12,15 @@ license: mit
 short_description: 'Slide creation AI assistant '
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Course Slide Factory
+
+Gradio workflow console for generating, reviewing, approving, and export-checking
+course slide decks with deterministic P0 trust and readiness gates.
+
+## Local checks
+
+```bash
+python -m pytest
+python -m ruff check .
+python app.py
+```

app.py

@@ -0,0 +1,435 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import gradio as gr
+
+from course_slide_factory.constants import STAGE_IDS, STAGE_LABELS
+from course_slide_factory.workflow import (
+    audit_table,
+    build_empty_state,
+    current_artifact_diff,
+    current_artifact_text,
+    deck_health_table,
+    final_render_export,
+    generate_stage,
+    grade_stage_action,
+    improve_with_ai,
+    issue_table,
+    objective_matrix_table,
+    preflight_table,
+    run_preflight_action,
+    save_human_edits,
+    slide_inventory_table,
+    stage_status_table,
+    state_from_dict,
+    state_to_dict,
+    update_setup_from_inputs,
+    approve_and_continue,
+)
+from course_slide_factory.quality import get_stage_lock_reasons
+
+
+CSS = """
+.gradio-container { max-width: 1480px !important; }
+textarea, .cm-editor { font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace; }
+.compact-panel table { font-size: 12px; }
+"""
+
+
+def _load_or_update_state(
+    state_data: dict[str, Any] | None,
+    deck_title: str,
+    source_url: str,
+    template_url: str,
+    output_folder_id: str,
+    dry_run: bool,
+    objectives_text: str,
+    source_text: str,
+    mutation_target_url: str,
+    production_export_requested: bool,
+):
+    state = state_from_dict(state_data)
+    if not state_data:
+        state = build_empty_state(
+            deck_title=deck_title,
+            source_url=source_url,
+            template_url=template_url,
+            output_folder_id=output_folder_id,
+            dry_run=dry_run,
+            mutation_target_url=mutation_target_url,
+            production_export_requested=production_export_requested,
+        )
+    state = update_setup_from_inputs(
+        state,
+        deck_title=deck_title,
+        source_url=source_url,
+        template_url=template_url,
+        output_folder_id=output_folder_id,
+        dry_run=dry_run,
+        objectives_text=objectives_text,
+        source_text=source_text,
+        mutation_target_url=mutation_target_url or None,
+        production_export_requested=production_export_requested,
+    )
+    return state
+
+
+def _render_outputs(state, stage_id: str, message: str):
+    lock_reasons = "\n".join(get_stage_lock_reasons(stage_id, state)) or "Unlocked"
+    state_json = json.dumps(state_to_dict(state), indent=2, sort_keys=True)
+    return (
+        state_to_dict(state),
+        message,
+        current_artifact_text(state, stage_id),
+        current_artifact_diff(state, stage_id),
+        lock_reasons,
+        deck_health_table(state),
+        slide_inventory_table(state),
+        objective_matrix_table(state),
+        preflight_table(state),
+        stage_status_table(state),
+        issue_table(state),
+        audit_table(state),
+        state_json,
+    )
+
+
+def handle_update_setup(
+    state_data,
+    deck_title,
+    source_url,
+    template_url,
+    output_folder_id,
+    dry_run,
+    objectives_text,
+    source_text,
+    mutation_target_url,
+    production_export_requested,
+    stage_id,
+):
+    state = _load_or_update_state(
+        state_data,
+        deck_title,
+        source_url,
+        template_url,
+        output_folder_id,
+        dry_run,
+        objectives_text,
+        source_text,
+        mutation_target_url,
+        production_export_requested,
+    )
+    return _render_outputs(state, stage_id, "Setup saved in working state.")
+
+
+def handle_generate(
+    state_data,
+    deck_title,
+    source_url,
+    template_url,
+    output_folder_id,
+    dry_run,
+    objectives_text,
+    source_text,
+    mutation_target_url,
+    production_export_requested,
+    stage_id,
+):
+    state = _load_or_update_state(
+        state_data,
+        deck_title,
+        source_url,
+        template_url,
+        output_folder_id,
+        dry_run,
+        objectives_text,
+        source_text,
+        mutation_target_url,
+        production_export_requested,
+    )
+    state, message = generate_stage(state, stage_id)
+    return _render_outputs(state, stage_id, message)
+
+
+def handle_grade(state_data, stage_id):
+    state = state_from_dict(state_data)
+    state, message = grade_stage_action(state, stage_id)
+    return _render_outputs(state, stage_id, message)
+
+
+def handle_save_edits(
+    state_data,
+    stage_id,
+    edited_content,
+    reviewer_name,
+    reviewer_summary,
+    requested_changes_json,
+):
+    state = state_from_dict(state_data)
+    try:
+        state, message = save_human_edits(
+            state,
+            stage_id,
+            edited_content,
+            reviewer_name,
+            reviewer_summary,
+            requested_changes_json or "[]",
+        )
+    except (json.JSONDecodeError, ValueError) as exc:
+        message = f"Reviewer notes were not saved: {exc}"
+    return _render_outputs(state, stage_id, message)
+
+
+def handle_improve(state_data, stage_id):
+    state = state_from_dict(state_data)
+    state, message = improve_with_ai(state, stage_id)
+    return _render_outputs(state, stage_id, message)
+
+
+def handle_approve(state_data, stage_id, reviewer_name):
+    state = state_from_dict(state_data)
+    state, message = approve_and_continue(state, stage_id, reviewer_name)
+    return _render_outputs(state, stage_id, message)
+
+
+def handle_preflight(state_data, stage_id):
+    state = state_from_dict(state_data)
+    state, message = run_preflight_action(state)
+    return _render_outputs(state, stage_id, message)
+
+
+def handle_export(state_data, stage_id):
+    state = state_from_dict(state_data)
+    state, message = final_render_export(state)
+    return _render_outputs(state, stage_id, message)
+
+
+def handle_stage_change(state_data, stage_id):
+    state = state_from_dict(state_data)
+    return _render_outputs(state, stage_id, f"Selected {STAGE_LABELS[stage_id]}.")
+
+
+def build_app() -> gr.Blocks:
+    stage_choices = [(f"{index + 1}. {STAGE_LABELS[stage_id]}", stage_id) for index, stage_id in enumerate(STAGE_IDS)]
+    with gr.Blocks(title="Course Slide Factory") as demo:
+        state_store = gr.State()
+
+        gr.Markdown("## Course Slide Factory")
+        with gr.Row():
+            with gr.Column(scale=1, min_width=320):
+                deck_title = gr.Textbox(label="Deck title", value="Course Slide Deck")
+                source_url = gr.Textbox(label="Source URL", value="mock://source/course")
+                template_url = gr.Textbox(label="Template URL", value="mock://template/course")
+                output_folder_id = gr.Textbox(label="Output folder ID", value="")
+                mutation_target_url = gr.Textbox(label="Mutation target URL", value="")
+                with gr.Row():
+                    dry_run = gr.Checkbox(label="Dry run", value=True)
+                    production_export_requested = gr.Checkbox(label="Production export", value=False)
+                objectives_text = gr.Textbox(
+                    label="Learning objectives",
+                    lines=5,
+                    value=(
+                        "obj_1: Explain the core course concept.\n"
+                        "obj_2: Apply the concept to a worked example."
+                    ),
+                )
+                source_text = gr.Textbox(
+                    label="Source text",
+                    lines=5,
+                    value="Mock source text for deterministic dry-run generation.",
+                )
+                update_setup = gr.Button("Save Setup", variant="secondary")
+
+            with gr.Column(scale=2):
+                stage_id = gr.Dropdown(
+                    label="Workflow stage",
+                    choices=stage_choices,
+                    value=STAGE_IDS[0],
+                    interactive=True,
+                )
+                with gr.Row():
+                    generate = gr.Button("Generate", variant="primary")
+                    grade = gr.Button("Grade")
+                    improve = gr.Button("Improve with AI")
+                    approve = gr.Button("Approve & Continue", variant="primary")
+                with gr.Row():
+                    preflight = gr.Button("Run Preflight")
+                    export = gr.Button("Final Render & Export", variant="stop")
+                status = gr.Markdown()
+                lock_reasons = gr.Textbox(label="Lock reasons", lines=4, interactive=False)
+
+                with gr.Tabs():
+                    with gr.Tab("Artifact"):
+                        artifact_editor = gr.Textbox(
+                            label="Current artifact / human edit",
+                            lines=16,
+                            interactive=True,
+                        )
+                        with gr.Row():
+                            reviewer_name = gr.Textbox(label="Reviewer", value="human_reviewer")
+                            reviewer_summary = gr.Textbox(label="Reviewer summary")
+                        requested_changes_json = gr.Textbox(
+                            label="Requested changes JSON",
+                            lines=6,
+                            value="[]",
+                        )
+                        save_edits = gr.Button("Save Human Edits")
+                        artifact_diff = gr.Textbox(label="Artifact diff", lines=12, interactive=False)
+                    with gr.Tab("Deck Health"):
+                        deck_health = gr.Dataframe(
+                            headers=["Metric", "Value"],
+                            interactive=False,
+                            elem_classes=["compact-panel"],
+                        )
+                    with gr.Tab("Slide Inventory"):
+                        slide_inventory = gr.Dataframe(
+                            headers=[
+                                "Slide #",
+                                "Title",
+                                "Role",
+                                "Objectives",
+                                "Claims",
+                                "Visuals",
+                                "Aggregate",
+                                "Technical",
+                                "Pedagogical",
+                                "Aesthetic",
+                                "Status",
+                                "Issues",
+                                "Stale?",
+                            ],
+                            interactive=False,
+                            elem_classes=["compact-panel"],
+                        )
+                    with gr.Tab("Objectives"):
+                        objective_matrix = gr.Dataframe(
+                            headers=[
+                                "Objective ID",
+                                "Objective",
+                                "Mapped Slides",
+                                "Coverage Score",
+                                "Coverage Status",
+                                "Evidence Count",
+                                "Issues",
+                            ],
+                            interactive=False,
+                            elem_classes=["compact-panel"],
+                        )
+                    with gr.Tab("Preflight"):
+                        preflight_results = gr.Dataframe(
+                            headers=["Field", "Value"],
+                            interactive=False,
+                            elem_classes=["compact-panel"],
+                        )
+                    with gr.Tab("Stages"):
+                        stage_status = gr.Dataframe(
+                            headers=[
+                                "#",
+                                "Stage",
+                                "Score",
+                                "Artifact",
+                                "Artifact Status",
+                                "Stale?",
+                                "Can Unlock Next?",
+                                "Lock Reasons",
+                            ],
+                            interactive=False,
+                            elem_classes=["compact-panel"],
+                        )
+                    with gr.Tab("Issues"):
+                        issues = gr.Dataframe(
+                            headers=[
+                                "Issue ID",
+                                "Type",
+                                "Severity",
+                                "Stage",
+                                "Slide",
+                                "Message",
+                                "Resolved?",
+                            ],
+                            interactive=False,
+                            elem_classes=["compact-panel"],
+                        )
+                    with gr.Tab("Audit"):
+                        audit = gr.Dataframe(
+                            headers=[
+                                "Event ID",
+                                "Type",
+                                "Timestamp",
+                                "Stage",
+                                "Slide",
+                                "Issue",
+                                "Reason",
+                            ],
+                            interactive=False,
+                            elem_classes=["compact-panel"],
+                        )
+                    with gr.Tab("State JSON"):
+                        state_json = gr.Textbox(label="State", lines=18, interactive=False)
+
+        outputs = [
+            state_store,
+            status,
+            artifact_editor,
+            artifact_diff,
+            lock_reasons,
+            deck_health,
+            slide_inventory,
+            objective_matrix,
+            preflight_results,
+            stage_status,
+            issues,
+            audit,
+            state_json,
+        ]
+        setup_inputs = [
+            state_store,
+            deck_title,
+            source_url,
+            template_url,
+            output_folder_id,
+            dry_run,
+            objectives_text,
+            source_text,
+            mutation_target_url,
+            production_export_requested,
+            stage_id,
+        ]
+
+        demo.load(
+            handle_update_setup,
+            inputs=setup_inputs,
+            outputs=outputs,
+            show_progress="hidden",
+        )
+        update_setup.click(handle_update_setup, inputs=setup_inputs, outputs=outputs)
+        generate.click(handle_generate, inputs=setup_inputs, outputs=outputs)
+        grade.click(handle_grade, inputs=[state_store, stage_id], outputs=outputs)
+        save_edits.click(
+            handle_save_edits,
+            inputs=[
+                state_store,
+                stage_id,
+                artifact_editor,
+                reviewer_name,
+                reviewer_summary,
+                requested_changes_json,
+            ],
+            outputs=outputs,
+        )
+        improve.click(handle_improve, inputs=[state_store, stage_id], outputs=outputs)
+        approve.click(handle_approve, inputs=[state_store, stage_id, reviewer_name], outputs=outputs)
+        preflight.click(handle_preflight, inputs=[state_store, stage_id], outputs=outputs)
+        export.click(handle_export, inputs=[state_store, stage_id], outputs=outputs)
+        stage_id.change(handle_stage_change, inputs=[state_store, stage_id], outputs=outputs)
+
+    return demo
+
+
+demo = build_app()
+
+
+if __name__ == "__main__":
+    demo.launch(css=CSS)

course_slide_factory/__init__.py

@@ -0,0 +1,2 @@
+"""Course Slide Factory application package."""
+

course_slide_factory/compat.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import json
+from typing import Any, TypeVar
+
+from pydantic import BaseModel
+
+
+ModelT = TypeVar("ModelT", bound=BaseModel)
+
+
+def model_to_dict(model: BaseModel) -> dict[str, Any]:
+    if hasattr(model, "model_dump"):
+        return model.model_dump(mode="json")  # type: ignore[attr-defined]
+    return json.loads(model.json())
+
+
+def model_validate(model_cls: type[ModelT], data: Any) -> ModelT:
+    if hasattr(model_cls, "model_validate"):
+        return model_cls.model_validate(data)  # type: ignore[attr-defined]
+    return model_cls.parse_obj(data)
+

course_slide_factory/constants.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+STAGE_SEQUENCE: list[tuple[str, str]] = [
+    ("setup_inputs", "Setup & Inputs"),
+    ("source_extraction_objective_mapping", "Source Extraction / Objective Mapping"),
+    ("slide_outline_order", "Slide Outline & Order"),
+    ("title_generation", "Title Generation"),
+    ("text_generation", "Text Generation"),
+    ("image_visual_asset_generation", "Image / Visual Asset Generation"),
+    ("aesthetic_ordering_visual_composition", "Aesthetic Ordering / Visual Composition"),
+    ("technical_review", "Technical Review"),
+    ("pedagogical_review", "Pedagogical Review"),
+    ("aesthetic_review", "Aesthetic Review"),
+    ("final_render_export", "Final Render & Export"),
+    ("audit_log_version_history", "Audit Log / Version History"),
+]
+
+STAGE_IDS = [stage_id for stage_id, _label in STAGE_SEQUENCE]
+STAGE_LABELS = dict(STAGE_SEQUENCE)
+
+APPROVAL_REQUIRED_STAGE_IDS = STAGE_IDS[:10]
+
+STAGE_RUBRICS: dict[str, list[tuple[str, str, float]]] = {
+    "setup_inputs": [
+        ("required_inputs_present", "Required inputs present", 1.4),
+        ("url_validity", "URL validity", 1.0),
+        ("render_mode_safety", "Render mode safety", 1.0),
+    ],
+    "source_extraction_objective_mapping": [
+        ("source_extraction_completeness", "Source extraction completeness", 1.0),
+        ("objective_clarity", "Objective clarity", 1.1),
+        ("objective_uniqueness", "Objective uniqueness", 0.8),
+    ],
+    "slide_outline_order": [
+        ("objective_coverage", "Objective coverage", 1.4),
+        ("logical_flow", "Logical flow", 1.0),
+        ("appropriate_slide_count", "Appropriate slide count", 0.8),
+        ("no_major_redundancy", "No major redundancy", 0.8),
+    ],
+    "title_generation": [
+        ("title_clarity", "Title clarity", 1.1),
+        ("title_specificity", "Title specificity", 1.0),
+        ("objective_alignment", "Objective alignment", 1.2),
+    ],
+    "text_generation": [
+        ("clarity", "Clarity", 1.0),
+        ("text_density", "Text density", 1.3),
+        ("source_grounding", "Source grounding", 1.3),
+        ("speaker_notes_quality", "Speaker notes quality", 1.0),
+    ],
+    "image_visual_asset_generation": [
+        ("visual_relevance", "Visual relevance", 1.0),
+        ("asset_completeness", "Asset completeness", 1.3),
+        ("alt_text_presence", "Alt text presence", 1.1),
+        ("license_or_generation_metadata", "License or generation metadata", 1.0),
+    ],
+    "aesthetic_ordering_visual_composition": [
+        ("layout_schema_validity", "Layout schema validity", 1.4),
+        ("slot_compliance", "Slot compliance", 1.1),
+        ("visual_hierarchy", "Visual hierarchy", 0.8),
+        ("readability", "Readability", 1.0),
+    ],
+    "technical_review": [
+        ("technical_accuracy", "Technical accuracy", 1.3),
+        ("unsupported_claim_detection", "Unsupported claim detection", 1.5),
+        ("terminology_consistency", "Terminology consistency", 0.8),
+    ],
+    "pedagogical_review": [
+        ("objective_coverage", "Objective coverage", 1.4),
+        ("learning_progression", "Learning progression", 1.0),
+        ("examples_or_applications", "Examples or applications", 0.8),
+        ("speaker_notes_teachability", "Speaker notes teachability", 1.1),
+    ],
+    "aesthetic_review": [
+        ("layout_consistency", "Layout consistency", 1.1),
+        ("readability", "Readability", 1.0),
+        ("accessibility", "Accessibility", 1.3),
+        ("brand_template_compliance", "Brand/template compliance", 1.0),
+    ],
+    "final_render_export": [
+        ("preflight_passed", "Preflight passed", 1.6),
+        ("all_required_approvals_valid", "All required approvals valid", 1.2),
+        ("no_stale_stages", "No stale stages", 1.1),
+        ("export_target_safety", "Export target safety", 1.0),
+    ],
+    "audit_log_version_history": [
+        ("audit_events_present", "Audit events present", 1.0),
+        ("artifact_versions_traceable", "Artifact versions traceable", 1.2),
+        ("review_notes_traceable", "Reviewer notes traceable", 0.8),
+    ],
+}
+
+TEXT_DENSITY_LIMITS: dict[str, dict[str, int]] = {
+    "intro": {"max_visible_words": 35, "max_bullets": 0},
+    "motivation": {"max_visible_words": 55, "max_bullets": 3},
+    "concept": {"max_visible_words": 65, "max_bullets": 4},
+    "worked_example": {"max_visible_words": 90, "max_bullets": 5},
+    "procedure": {"max_visible_words": 90, "max_bullets": 6},
+    "comparison": {"max_visible_words": 85, "max_bullets": 5},
+    "common_misconception": {"max_visible_words": 75, "max_bullets": 4},
+    "knowledge_check": {"max_visible_words": 100, "max_bullets": 6},
+    "demo_setup": {"max_visible_words": 80, "max_bullets": 5},
+    "demo_walkthrough": {"max_visible_words": 90, "max_bullets": 6},
+    "summary": {"max_visible_words": 70, "max_bullets": 5},
+    "transition": {"max_visible_words": 25, "max_bullets": 0},
+    "unknown": {"max_visible_words": 65, "max_bullets": 4},
+}
+
+INSTRUCTIONAL_ROLES = {
+    "concept",
+    "worked_example",
+    "procedure",
+    "comparison",
+    "common_misconception",
+    "knowledge_check",
+    "demo_setup",
+    "demo_walkthrough",
+    "summary",
+}
+
+APPROVED_LAYOUTS: dict[str, dict[str, object]] = {
+    "title_only": {
+        "required_slots": {"title"},
+        "optional_slots": set(),
+        "slot_word_limits": {"title": 14},
+    },
+    "title_body": {
+        "required_slots": {"title", "body"},
+        "optional_slots": {"subtitle"},
+        "slot_word_limits": {"title": 14, "body": 80, "subtitle": 24},
+    },
+    "title_bullets_visual": {
+        "required_slots": {"title", "bullets", "visual"},
+        "optional_slots": {"caption"},
+        "slot_word_limits": {"title": 14, "bullets": 90, "visual": 8, "caption": 18},
+    },
+    "worked_example": {
+        "required_slots": {"title", "problem", "steps"},
+        "optional_slots": {"visual", "answer"},
+        "slot_word_limits": {"title": 14, "problem": 45, "steps": 100, "visual": 8, "answer": 30},
+    },
+    "knowledge_check": {
+        "required_slots": {"title", "question", "options"},
+        "optional_slots": {"answer"},
+        "slot_word_limits": {"title": 14, "question": 55, "options": 70, "answer": 28},
+    },
+}
+
+RAW_COORDINATE_KEYS = {"x", "y", "width", "height", "left", "top", "w", "h"}
+

course_slide_factory/fixtures.py

@@ -0,0 +1,210 @@
+from __future__ import annotations
+
+from copy import deepcopy
+
+from .constants import APPROVAL_REQUIRED_STAGE_IDS
+from .compat import model_to_dict, model_validate
+from .models import (
+    ArtifactStatus,
+    LayoutSpec,
+    PedagogicalRole,
+    Slide,
+    SlideClaim,
+    SpeakerNotes,
+    VisualAsset,
+)
+from .quality import approve_current_artifact, create_artifact_version, grade_stage
+from .workflow import build_empty_state, record_prompt_run
+
+
+def valid_minimal_job():
+    state = build_empty_state(
+        deck_title="Intro to Deterministic Slide QA",
+        source_url="mock://source/course-notes",
+        template_url="mock://template/course",
+        dry_run=True,
+    )
+    state.source_chunks = {
+        "chunk_1": "Deterministic slide QA requires objective mapping and supported claims.",
+        "chunk_2": "A worked example helps learners apply the concept.",
+    }
+    state.objectives = {
+        "obj_1": "Explain deterministic slide quality checks.",
+        "obj_2": "Apply quality checks to a worked example.",
+    }
+    state.slides = {
+        "slide_1": Slide(
+            slide_id="slide_1",
+            slide_number=1,
+            title="Deterministic Slide QA",
+            visible_text="Quality checks make slide readiness explicit.",
+            bullet_points=["Trace objectives", "Check evidence", "Validate layout"],
+            objective_ids=["obj_1"],
+            pedagogical_role=PedagogicalRole.CONCEPT,
+            requires_visual=True,
+            speaker_notes=SpeakerNotes(
+                slide_id="slide_1",
+                notes_text="Explain why readiness must be explicit before export.",
+                instructor_intent="Connect quality gates to production trust.",
+                estimated_teaching_time_seconds=180,
+            ),
+        ),
+        "slide_2": Slide(
+            slide_id="slide_2",
+            slide_number=2,
+            title="Worked QA Example",
+            visible_text="Use the checklist to evaluate one generated slide.",
+            bullet_points=["Map objective", "Inspect claim support", "Confirm layout"],
+            objective_ids=["obj_2"],
+            pedagogical_role=PedagogicalRole.WORKED_EXAMPLE,
+            requires_visual=False,
+            speaker_notes=SpeakerNotes(
+                slide_id="slide_2",
+                notes_text="Walk through each checklist item and ask learners what blocks export.",
+                instructor_intent="Make the gating rule concrete.",
+                estimated_teaching_time_seconds=240,
+            ),
+        ),
+    }
+    state.claims = {
+        "claim_1": SlideClaim(
+            claim_id="claim_1",
+            slide_id="slide_1",
+            claim_text="Quality checks make slide readiness explicit.",
+            source_ids=["source_1"],
+            source_chunk_ids=["chunk_1"],
+            review_status="supported",
+        ),
+        "claim_2": SlideClaim(
+            claim_id="claim_2",
+            slide_id="slide_2",
+            claim_text="A worked example helps learners apply the concept.",
+            source_ids=["source_1"],
+            source_chunk_ids=["chunk_2"],
+            review_status="supported",
+        ),
+    }
+    state.visual_assets = {
+        "asset_1": VisualAsset(
+            asset_id="asset_1",
+            slide_id="slide_1",
+            asset_type="diagram",
+            path_or_url="mock://asset/qa-flow",
+            prompt="Simple quality gate diagram",
+            purpose="instructional",
+            alt_text="Flow from generation through grading, approval, and export.",
+            source="mock",
+            license_status="generated",
+            approved_for_export=True,
+        )
+    }
+    state.layout_specs = {
+        "slide_1": LayoutSpec(
+            slide_id="slide_1",
+            layout_id="title_bullets_visual",
+            approved_template_id="default_course_template",
+            slot_assignments={
+                "title": "Deterministic Slide QA",
+                "bullets": ["Trace objectives", "Check evidence", "Validate layout"],
+                "visual": "asset_1",
+            },
+        ),
+        "slide_2": LayoutSpec(
+            slide_id="slide_2",
+            layout_id="worked_example",
+            approved_template_id="default_course_template",
+            slot_assignments={
+                "title": "Worked QA Example",
+                "problem": "Evaluate one generated slide.",
+                "steps": ["Map objective", "Inspect claim support", "Confirm layout"],
+            },
+        ),
+    }
+    # Pydantic validates nested dicts assigned above during this explicit round-trip;
+    # this explicit round-trip keeps fixture construction terse and typed.
+    state = model_validate(type(state), model_to_dict(state))
+
+    for stage_id in APPROVAL_REQUIRED_STAGE_IDS:
+        prompt_run = record_prompt_run(
+            state,
+            stage_id,
+            rendered_prompt=f"Fixture generation for {stage_id}",
+        )
+        artifact = create_artifact_version(
+            state,
+            stage_id,
+            {"fixture": stage_id},
+            created_by="mock",
+            status=ArtifactStatus.CANDIDATE,
+            prompt_run_id=prompt_run.prompt_run_id,
+            mark_downstream_stale=False,
+        )
+        prompt_run.output_artifact_version_id = artifact.artifact_version_id
+        grade_stage(stage_id, state)
+        approve_current_artifact(state, stage_id, reviewer_name="fixture_reviewer")
+    return state
+
+
+def missing_objective_mapping_job():
+    state = deepcopy(valid_minimal_job())
+    state.slides["slide_2"].objective_ids = []
+    return state
+
+
+def stale_downstream_job():
+    state = deepcopy(valid_minimal_job())
+    create_artifact_version(
+        state,
+        "slide_outline_order",
+        {"fixture": "changed outline"},
+        created_by="human",
+        status=ArtifactStatus.CANDIDATE,
+        mark_downstream_stale=True,
+    )
+    return state
+
+
+def invalidated_approval_job():
+    state = deepcopy(valid_minimal_job())
+    create_artifact_version(
+        state,
+        "text_generation",
+        {"fixture": "human edit after approval"},
+        created_by="human",
+        status=ArtifactStatus.CANDIDATE,
+        mark_downstream_stale=False,
+    )
+    return state
+
+
+def unsupported_claim_job():
+    state = deepcopy(valid_minimal_job())
+    claim = state.claims["claim_1"]
+    claim.review_status = "unsupported"
+    claim.source_ids = []
+    claim.source_chunk_ids = []
+    return state
+
+
+def missing_visual_asset_job():
+    state = deepcopy(valid_minimal_job())
+    state.visual_assets = {}
+    return state
+
+
+def invalid_layout_job():
+    state = deepcopy(valid_minimal_job())
+    state.layout_specs["slide_1"] = LayoutSpec(
+        slide_id="slide_1",
+        layout_id="raw_coordinates",
+        approved_template_id=None,
+        slot_assignments={"x": 10, "y": 20, "width": 400, "height": 300},
+    )
+    return state
+
+
+def text_density_failure_job():
+    state = deepcopy(valid_minimal_job())
+    state.slides["slide_1"].visible_text = " ".join(["dense"] * 80)
+    state.slides["slide_1"].bullet_points = ["one", "two", "three", "four", "five"]
+    return state

course_slide_factory/models.py

@@ -0,0 +1,351 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+
+class RubricDimensionScore(BaseModel):
+    dimension_id: str
+    label: str
+    score: int
+    weight: float = 1.0
+    rationale: str | None = None
+
+
+class StageGradeResult(BaseModel):
+    stage_id: str
+    artifact_version_id: str | None = None
+    score: int
+    passed_threshold: bool
+    rubric_scores: list[RubricDimensionScore] = Field(default_factory=list)
+    issue_ids: list[str] = Field(default_factory=list)
+    recommended_actions: list[str] = Field(default_factory=list)
+    graded_at: str
+    grader: Literal["deterministic", "mock_ai", "human", "system"] = "deterministic"
+
+
+class IssueType(str, Enum):
+    MISSING_REQUIRED_INPUT = "missing_required_input"
+    SOURCE_EXTRACTION_FAILED = "source_extraction_failed"
+    OBJECTIVE_UNCOVERED = "objective_uncovered"
+    OBJECTIVE_WEAKLY_COVERED = "objective_weakly_covered"
+    UNSUPPORTED_CLAIM = "unsupported_claim"
+    TECHNICAL_INACCURACY = "technical_inaccuracy"
+    TEXT_DENSITY_EXCEEDED = "text_density_exceeded"
+    COGNITIVE_LOAD_HIGH = "cognitive_load_high"
+    SPEAKER_NOTES_MISSING = "speaker_notes_missing"
+    LAYOUT_SCHEMA_INVALID = "layout_schema_invalid"
+    LAYOUT_SLOT_VIOLATION = "layout_slot_violation"
+    VISUAL_ASSET_MISSING = "visual_asset_missing"
+    VISUAL_ASSET_PURPOSE_MISSING = "visual_asset_purpose_missing"
+    ALT_TEXT_MISSING = "alt_text_missing"
+    ACCESSIBILITY_CONTRAST_RISK = "accessibility_contrast_risk"
+    COPYRIGHT_OR_LICENSE_RISK = "copyright_or_license_risk"
+    RENDER_SAFETY_VIOLATION = "render_safety_violation"
+    GOOGLE_EXPORT_FAILED = "google_export_failed"
+    HUMAN_APPROVAL_MISSING = "human_approval_missing"
+    SCORE_BELOW_THRESHOLD = "score_below_threshold"
+    STALE_DOWNSTREAM_STAGE = "stale_downstream_stage"
+    APPROVAL_INVALIDATED = "approval_invalidated"
+    EXPORT_PREFLIGHT_FAILED = "export_preflight_failed"
+
+
+class IssueSeverity(str, Enum):
+    INFO = "info"
+    MINOR = "minor"
+    MAJOR = "major"
+    BLOCKER = "blocker"
+
+
+class QualityIssue(BaseModel):
+    issue_id: str
+    issue_type: IssueType
+    severity: IssueSeverity
+    message: str
+    stage_id: str | None = None
+    slide_id: str | None = None
+    objective_id: str | None = None
+    artifact_version_id: str | None = None
+    claim_id: str | None = None
+    suggested_fix: str | None = None
+    resolved: bool = False
+    created_at: str
+    resolved_at: str | None = None
+
+
+class ObjectiveTrace(BaseModel):
+    objective_id: str
+    objective_text: str
+    mapped_slide_ids: list[str] = Field(default_factory=list)
+    coverage_score: int = 0
+    coverage_status: Literal["uncovered", "weak", "partial", "strong"] = "uncovered"
+    evidence: list[str] = Field(default_factory=list)
+    issue_ids: list[str] = Field(default_factory=list)
+
+
+class PedagogicalRole(str, Enum):
+    INTRO = "intro"
+    MOTIVATION = "motivation"
+    CONCEPT = "concept"
+    WORKED_EXAMPLE = "worked_example"
+    PROCEDURE = "procedure"
+    COMPARISON = "comparison"
+    COMMON_MISCONCEPTION = "common_misconception"
+    KNOWLEDGE_CHECK = "knowledge_check"
+    DEMO_SETUP = "demo_setup"
+    DEMO_WALKTHROUGH = "demo_walkthrough"
+    SUMMARY = "summary"
+    TRANSITION = "transition"
+    UNKNOWN = "unknown"
+
+
+class SpeakerNotes(BaseModel):
+    slide_id: str
+    notes_text: str | None = None
+    instructor_intent: str | None = None
+    estimated_teaching_time_seconds: int | None = None
+    possible_student_confusions: list[str] = Field(default_factory=list)
+    teaching_tips: list[str] = Field(default_factory=list)
+
+
+class SlideClaim(BaseModel):
+    claim_id: str
+    slide_id: str
+    claim_text: str
+    source_ids: list[str] = Field(default_factory=list)
+    source_chunk_ids: list[str] = Field(default_factory=list)
+    confidence: float = Field(default=1.0, ge=0.0, le=1.0)
+    review_status: Literal[
+        "unreviewed",
+        "supported",
+        "unsupported",
+        "needs_human_review",
+    ] = "unreviewed"
+    issue_ids: list[str] = Field(default_factory=list)
+
+
+class VisualAsset(BaseModel):
+    asset_id: str
+    slide_id: str | None = None
+    asset_type: Literal[
+        "generated_image",
+        "diagram",
+        "chart",
+        "icon",
+        "screenshot",
+        "placeholder",
+    ]
+    path_or_url: str | None = None
+    prompt: str | None = None
+    purpose: Literal["instructional", "decorative", "background", "unknown"] = "unknown"
+    alt_text: str | None = None
+    source: Literal["mock", "generated", "uploaded", "google_drive", "unknown"] = "unknown"
+    license_status: Literal[
+        "generated",
+        "user_provided",
+        "unknown",
+        "needs_review",
+    ] = "unknown"
+    approved_for_export: bool = False
+    issue_ids: list[str] = Field(default_factory=list)
+
+
+class LayoutSpec(BaseModel):
+    slide_id: str
+    layout_id: str
+    approved_template_id: str | None = None
+    slot_assignments: dict[str, Any] = Field(default_factory=dict)
+    schema_version: str = "1.0"
+
+
+class ArtifactStatus(str, Enum):
+    DRAFT = "draft"
+    CANDIDATE = "candidate"
+    APPROVED = "approved"
+    EXPORTED = "exported"
+    STALE = "stale"
+    INVALIDATED = "invalidated"
+
+
+class ArtifactVersion(BaseModel):
+    artifact_version_id: str
+    artifact_id: str
+    stage_id: str
+    version_number: int
+    created_at: str
+    created_by: Literal["system", "human", "ai", "mock"]
+    content_hash: str
+    status: ArtifactStatus = ArtifactStatus.DRAFT
+    parent_artifact_version_ids: list[str] = Field(default_factory=list)
+    upstream_stage_versions: dict[str, str] = Field(default_factory=dict)
+    prompt_run_id: str | None = None
+    is_current: bool = True
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class RequestedChange(BaseModel):
+    slide_id: str | None = None
+    stage_id: str | None = None
+    change_type: Literal[
+        "add_example",
+        "reduce_text",
+        "fix_technical_accuracy",
+        "improve_layout",
+        "add_visual",
+        "remove_visual",
+        "add_speaker_notes",
+        "improve_objective_alignment",
+        "other",
+    ] = "other"
+    priority: Literal["low", "medium", "high", "blocker"] = "medium"
+    note: str
+
+
+class ReviewerNotes(BaseModel):
+    stage_id: str
+    artifact_version_id: str | None = None
+    reviewer_name: str = "human_reviewer"
+    summary: str | None = None
+    requested_changes: list[RequestedChange] = Field(default_factory=list)
+    created_at: str
+
+
+class PromptRun(BaseModel):
+    prompt_run_id: str
+    stage_id: str
+    artifact_version_id: str | None = None
+    provider: Literal["mock", "openai", "anthropic", "gemini", "system"] = "mock"
+    model_name: str | None = None
+    prompt_template_id: str | None = None
+    prompt_template_version: str | None = None
+    rendered_prompt: str | None = None
+    input_artifact_version_ids: list[str] = Field(default_factory=list)
+    output_artifact_version_id: str | None = None
+    settings: dict[str, Any] = Field(default_factory=dict)
+    created_at: str
+
+
+class DeckHealthSummary(BaseModel):
+    job_id: str
+    deck_title: str | None = None
+    approved_stage_count: int
+    total_stage_count: int
+    stale_stage_count: int
+    invalidated_approval_count: int
+    unresolved_blocker_count: int
+    unresolved_major_issue_count: int
+    slide_count: int
+    average_slide_score: int | None = None
+    objectives_total: int
+    objectives_strong: int
+    objectives_partial_or_weak: int
+    objectives_uncovered: int
+    unsupported_claim_count: int
+    can_export: bool
+    top_blockers: list[str] = Field(default_factory=list)
+
+
+class SlideQAStatus(BaseModel):
+    slide_id: str
+    slide_number: int
+    title: str | None = None
+    pedagogical_role: PedagogicalRole = PedagogicalRole.UNKNOWN
+    objective_ids: list[str] = Field(default_factory=list)
+    claim_ids: list[str] = Field(default_factory=list)
+    visual_asset_ids: list[str] = Field(default_factory=list)
+    layout_id: str | None = None
+    aggregate_score: int | None = None
+    technical_score: int | None = None
+    pedagogical_score: int | None = None
+    aesthetic_score: int | None = None
+    issue_ids: list[str] = Field(default_factory=list)
+    stale: bool = False
+    status: Literal["draft", "needs_review", "needs_revision", "approved", "stale", "blocked"] = (
+        "draft"
+    )
+
+
+class ExportPreflightReport(BaseModel):
+    can_export: bool
+    checked_at: str
+    blocking_issue_ids: list[str] = Field(default_factory=list)
+    warning_issue_ids: list[str] = Field(default_factory=list)
+    summary: str
+
+
+class StageApproval(BaseModel):
+    stage_id: str
+    artifact_version_id: str
+    reviewer_name: str = "human_reviewer"
+    approved_at: str
+    approval_status: Literal["approved", "invalidated", "revoked"] = "approved"
+    invalidated_at: str | None = None
+    invalidation_reason: str | None = None
+
+
+class AuditEvent(BaseModel):
+    event_id: str
+    event_type: str
+    timestamp: str
+    stage_id: str | None = None
+    slide_id: str | None = None
+    objective_id: str | None = None
+    artifact_version_id: str | None = None
+    issue_id: str | None = None
+    claim_id: str | None = None
+    reason: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class StageState(BaseModel):
+    stage_id: str
+    label: str
+    score: int | None = None
+    is_stale: bool = False
+    current_artifact_version_id: str | None = None
+    grade_result: StageGradeResult | None = None
+
+
+class Slide(BaseModel):
+    slide_id: str
+    slide_number: int
+    title: str | None = None
+    visible_text: str = ""
+    bullet_points: list[str] = Field(default_factory=list)
+    objective_ids: list[str] = Field(default_factory=list)
+    pedagogical_role: PedagogicalRole = PedagogicalRole.UNKNOWN
+    requires_visual: bool = False
+    speaker_notes: SpeakerNotes | None = None
+    objective_coverage_scores: dict[str, int] = Field(default_factory=dict)
+    distinct_concept_count: int | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class PipelineState(BaseModel):
+    job_id: str
+    deck_title: str | None = None
+    dry_run: bool = True
+    production_export_requested: bool = False
+    source_url: str | None = None
+    template_url: str | None = None
+    output_folder_id: str | None = None
+    mutation_target_url: str | None = None
+    source_chunks: dict[str, str] = Field(default_factory=dict)
+    objectives: dict[str, str] = Field(default_factory=dict)
+    stages: dict[str, StageState] = Field(default_factory=dict)
+    artifacts: dict[str, ArtifactVersion] = Field(default_factory=dict)
+    stage_artifact_versions: dict[str, list[str]] = Field(default_factory=dict)
+    issues: dict[str, QualityIssue] = Field(default_factory=dict)
+    objective_traces: dict[str, ObjectiveTrace] = Field(default_factory=dict)
+    slides: dict[str, Slide] = Field(default_factory=dict)
+    slide_statuses: dict[str, SlideQAStatus] = Field(default_factory=dict)
+    claims: dict[str, SlideClaim] = Field(default_factory=dict)
+    visual_assets: dict[str, VisualAsset] = Field(default_factory=dict)
+    layout_specs: dict[str, LayoutSpec] = Field(default_factory=dict)
+    approvals: list[StageApproval] = Field(default_factory=list)
+    reviewer_notes: list[ReviewerNotes] = Field(default_factory=list)
+    prompt_runs: dict[str, PromptRun] = Field(default_factory=dict)
+    audit_events: list[AuditEvent] = Field(default_factory=list)
+    export_preflight_report: ExportPreflightReport | None = None

course_slide_factory/quality.py

@@ -0,0 +1,1451 @@
+from __future__ import annotations
+
+import difflib
+import hashlib
+import json
+import re
+from datetime import UTC, datetime
+from typing import Any
+
+from .constants import (
+    APPROVAL_REQUIRED_STAGE_IDS,
+    APPROVED_LAYOUTS,
+    INSTRUCTIONAL_ROLES,
+    RAW_COORDINATE_KEYS,
+    STAGE_IDS,
+    STAGE_RUBRICS,
+    TEXT_DENSITY_LIMITS,
+)
+from .models import (
+    ArtifactStatus,
+    ArtifactVersion,
+    AuditEvent,
+    DeckHealthSummary,
+    ExportPreflightReport,
+    IssueSeverity,
+    IssueType,
+    LayoutSpec,
+    ObjectiveTrace,
+    PipelineState,
+    QualityIssue,
+    RubricDimensionScore,
+    SlideQAStatus,
+    StageApproval,
+    StageGradeResult,
+)
+
+
+PASSING_SCORE = 80
+SLIDE_REVIEW_STAGE_IDS = {
+    "text_generation",
+    "image_visual_asset_generation",
+    "aesthetic_ordering_visual_composition",
+    "technical_review",
+    "pedagogical_review",
+    "aesthetic_review",
+}
+
+
+def now_iso() -> str:
+    return datetime.now(tz=UTC).replace(microsecond=0).isoformat()
+
+
+def clamp_score(score: float) -> int:
+    return max(0, min(100, int(round(score))))
+
+
+def stable_hash(content: Any) -> str:
+    rendered = json.dumps(content, sort_keys=True, default=str)
+    return hashlib.sha256(rendered.encode("utf-8")).hexdigest()
+
+
+def record_audit(
+    state: PipelineState,
+    event_type: str,
+    *,
+    stage_id: str | None = None,
+    slide_id: str | None = None,
+    objective_id: str | None = None,
+    artifact_version_id: str | None = None,
+    issue_id: str | None = None,
+    claim_id: str | None = None,
+    reason: str | None = None,
+    metadata: dict[str, Any] | None = None,
+) -> AuditEvent:
+    event = AuditEvent(
+        event_id=f"evt_{len(state.audit_events) + 1:05d}",
+        event_type=event_type,
+        timestamp=now_iso(),
+        stage_id=stage_id,
+        slide_id=slide_id,
+        objective_id=objective_id,
+        artifact_version_id=artifact_version_id,
+        issue_id=issue_id,
+        claim_id=claim_id,
+        reason=reason,
+        metadata=metadata or {},
+    )
+    state.audit_events.append(event)
+    return event
+
+
+def aggregate_rubric_score(rubric_scores: list[RubricDimensionScore]) -> int:
+    if not rubric_scores:
+        return 0
+    total_weight = sum(max(0.0, item.weight) for item in rubric_scores)
+    if total_weight <= 0:
+        return 0
+    weighted = sum(clamp_score(item.score) * max(0.0, item.weight) for item in rubric_scores)
+    return clamp_score(weighted / total_weight)
+
+
+def issue_id_for(
+    issue_type: IssueType,
+    *,
+    stage_id: str | None = None,
+    slide_id: str | None = None,
+    objective_id: str | None = None,
+    claim_id: str | None = None,
+    artifact_version_id: str | None = None,
+) -> str:
+    parts = [
+        issue_type.value,
+        stage_id or "deck",
+        slide_id or "all_slides",
+        objective_id or "all_objectives",
+        claim_id or "all_claims",
+        artifact_version_id or "current",
+    ]
+    safe = [re.sub(r"[^a-zA-Z0-9_.-]+", "_", part) for part in parts]
+    return "issue:" + ":".join(safe)
+
+
+def upsert_issue(
+    state: PipelineState,
+    issue_type: IssueType,
+    severity: IssueSeverity,
+    message: str,
+    *,
+    stage_id: str | None = None,
+    slide_id: str | None = None,
+    objective_id: str | None = None,
+    artifact_version_id: str | None = None,
+    claim_id: str | None = None,
+    suggested_fix: str | None = None,
+) -> QualityIssue:
+    issue_id = issue_id_for(
+        issue_type,
+        stage_id=stage_id,
+        slide_id=slide_id,
+        objective_id=objective_id,
+        claim_id=claim_id,
+        artifact_version_id=artifact_version_id,
+    )
+    existing = state.issues.get(issue_id)
+    if existing is None:
+        issue = QualityIssue(
+            issue_id=issue_id,
+            issue_type=issue_type,
+            severity=severity,
+            message=message,
+            stage_id=stage_id,
+            slide_id=slide_id,
+            objective_id=objective_id,
+            artifact_version_id=artifact_version_id,
+            claim_id=claim_id,
+            suggested_fix=suggested_fix,
+            created_at=now_iso(),
+        )
+        state.issues[issue_id] = issue
+        record_audit(
+            state,
+            "issue_created",
+            stage_id=stage_id,
+            slide_id=slide_id,
+            objective_id=objective_id,
+            artifact_version_id=artifact_version_id,
+            issue_id=issue_id,
+            claim_id=claim_id,
+            reason=message,
+        )
+        return issue
+
+    existing.severity = severity
+    existing.message = message
+    existing.suggested_fix = suggested_fix
+    existing.resolved = False
+    existing.resolved_at = None
+    return existing
+
+
+def resolve_issue(state: PipelineState, issue_id: str) -> None:
+    issue = state.issues.get(issue_id)
+    if issue and not issue.resolved:
+        issue.resolved = True
+        issue.resolved_at = now_iso()
+        record_audit(state, "issue_resolved", issue_id=issue_id, reason="Condition cleared")
+
+
+def unresolved_issues(state: PipelineState) -> list[QualityIssue]:
+    return [issue for issue in state.issues.values() if not issue.resolved]
+
+
+def unresolved_stage_issues(stage_id: str, state: PipelineState) -> list[QualityIssue]:
+    return [issue for issue in unresolved_issues(state) if issue.stage_id == stage_id]
+
+
+def has_unresolved_blockers(stage_id: str, state: PipelineState) -> bool:
+    return any(
+        issue.severity == IssueSeverity.BLOCKER for issue in unresolved_stage_issues(stage_id, state)
+    )
+
+
+def has_unresolved_blockers_anywhere(state: PipelineState) -> bool:
+    return any(issue.severity == IssueSeverity.BLOCKER for issue in unresolved_issues(state))
+
+
+def get_current_stage_artifact(stage_id: str, state: PipelineState) -> ArtifactVersion | None:
+    stage = state.stages.get(stage_id)
+    if stage and stage.current_artifact_version_id:
+        artifact = state.artifacts.get(stage.current_artifact_version_id)
+        if artifact and artifact.is_current:
+            return artifact
+    for artifact_id in reversed(state.stage_artifact_versions.get(stage_id, [])):
+        artifact = state.artifacts.get(artifact_id)
+        if artifact and artifact.is_current:
+            return artifact
+    return None
+
+
+def create_artifact_version(
+    state: PipelineState,
+    stage_id: str,
+    content: Any,
+    *,
+    created_by: str,
+    status: ArtifactStatus = ArtifactStatus.CANDIDATE,
+    prompt_run_id: str | None = None,
+    mark_downstream_stale: bool = True,
+) -> ArtifactVersion:
+    previous = get_current_stage_artifact(stage_id, state)
+    if previous:
+        previous.is_current = False
+        if previous.status in {ArtifactStatus.DRAFT, ArtifactStatus.CANDIDATE}:
+            previous.status = ArtifactStatus.INVALIDATED
+
+    version_number = len(state.stage_artifact_versions.get(stage_id, [])) + 1
+    artifact_version_id = f"{stage_id}_v{version_number}"
+    artifact = ArtifactVersion(
+        artifact_version_id=artifact_version_id,
+        artifact_id=stage_id,
+        stage_id=stage_id,
+        version_number=version_number,
+        created_at=now_iso(),
+        created_by=created_by,  # type: ignore[arg-type]
+        content_hash=stable_hash(content),
+        status=status,
+        parent_artifact_version_ids=[previous.artifact_version_id] if previous else [],
+        upstream_stage_versions={
+            upstream_id: upstream.current_artifact_version_id
+            for upstream_id, upstream in state.stages.items()
+            if STAGE_IDS.index(upstream_id) < STAGE_IDS.index(stage_id)
+            and upstream.current_artifact_version_id
+        },
+        prompt_run_id=prompt_run_id,
+        metadata={"content": content},
+    )
+    state.artifacts[artifact_version_id] = artifact
+    state.stage_artifact_versions.setdefault(stage_id, []).append(artifact_version_id)
+    state.stages[stage_id].current_artifact_version_id = artifact_version_id
+    state.stages[stage_id].is_stale = False
+    for issue in list(state.issues.values()):
+        if issue.stage_id == stage_id and issue.issue_type == IssueType.STALE_DOWNSTREAM_STAGE:
+            resolve_issue(state, issue.issue_id)
+    record_audit(
+        state,
+        "artifact_promoted" if status == ArtifactStatus.APPROVED else "artifact_created",
+        stage_id=stage_id,
+        artifact_version_id=artifact_version_id,
+        metadata={"status": status.value, "created_by": created_by},
+    )
+    invalidate_approval_for_stage(state, stage_id, "Artifact changed after approval")
+    if mark_downstream_stale:
+        mark_downstream_stages_stale(stage_id, state)
+    return artifact
+
+
+def invalidate_approval_for_stage(state: PipelineState, stage_id: str, reason: str) -> None:
+    for approval in state.approvals:
+        if approval.stage_id != stage_id or approval.approval_status != "approved":
+            continue
+        approval.approval_status = "invalidated"
+        approval.invalidated_at = now_iso()
+        approval.invalidation_reason = reason
+        upsert_issue(
+            state,
+            IssueType.APPROVAL_INVALIDATED,
+            IssueSeverity.BLOCKER,
+            f"Approval for {stage_id} was invalidated: {reason}",
+            stage_id=stage_id,
+            artifact_version_id=approval.artifact_version_id,
+            suggested_fix="Re-grade and approve the current artifact version.",
+        )
+        record_audit(
+            state,
+            "approval_invalidated",
+            stage_id=stage_id,
+            artifact_version_id=approval.artifact_version_id,
+            reason=reason,
+        )
+
+
+def has_valid_human_approval(stage_id: str, state: PipelineState) -> bool:
+    artifact = get_current_stage_artifact(stage_id, state)
+    if artifact is None:
+        return False
+    if artifact.status not in {ArtifactStatus.APPROVED, ArtifactStatus.EXPORTED}:
+        return False
+    return any(
+        approval.stage_id == stage_id
+        and approval.artifact_version_id == artifact.artifact_version_id
+        and approval.approval_status == "approved"
+        for approval in state.approvals
+    )
+
+
+def approve_current_artifact(
+    state: PipelineState,
+    stage_id: str,
+    reviewer_name: str = "human_reviewer",
+) -> StageApproval:
+    artifact = get_current_stage_artifact(stage_id, state)
+    if artifact is None:
+        raise ValueError(f"Cannot approve {stage_id}: current artifact is missing.")
+    artifact.status = ArtifactStatus.APPROVED
+    approval = StageApproval(
+        stage_id=stage_id,
+        artifact_version_id=artifact.artifact_version_id,
+        reviewer_name=reviewer_name or "human_reviewer",
+        approved_at=now_iso(),
+    )
+    state.approvals.append(approval)
+    record_audit(
+        state,
+        "approval_created",
+        stage_id=stage_id,
+        artifact_version_id=artifact.artifact_version_id,
+        metadata={"reviewer_name": approval.reviewer_name},
+    )
+    record_audit(
+        state,
+        "artifact_promoted",
+        stage_id=stage_id,
+        artifact_version_id=artifact.artifact_version_id,
+        metadata={"status": artifact.status.value},
+    )
+    return approval
+
+
+def mark_downstream_stages_stale(changed_stage_id: str, state: PipelineState) -> PipelineState:
+    if changed_stage_id not in STAGE_IDS:
+        return state
+    changed_index = STAGE_IDS.index(changed_stage_id)
+    for stage_id in STAGE_IDS[changed_index + 1 :]:
+        stage = state.stages[stage_id]
+        stage.is_stale = True
+        artifact = get_current_stage_artifact(stage_id, state)
+        artifact_version_id = None
+        if artifact is not None:
+            artifact.status = ArtifactStatus.STALE
+            artifact_version_id = artifact.artifact_version_id
+            record_audit(
+                state,
+                "artifact_marked_stale",
+                stage_id=stage_id,
+                artifact_version_id=artifact.artifact_version_id,
+                reason=f"Upstream stage {changed_stage_id} changed",
+            )
+        invalidate_approval_for_stage(
+            state,
+            stage_id,
+            f"Upstream stage {changed_stage_id} changed",
+        )
+        upsert_issue(
+            state,
+            IssueType.STALE_DOWNSTREAM_STAGE,
+            IssueSeverity.BLOCKER,
+            f"{stage_id} is stale because {changed_stage_id} changed.",
+            stage_id=stage_id,
+            artifact_version_id=artifact_version_id,
+            suggested_fix="Regenerate or re-grade this stage against the current upstream artifacts.",
+        )
+        record_audit(
+            state,
+            "stage_marked_stale",
+            stage_id=stage_id,
+            reason=f"Upstream stage {changed_stage_id} changed",
+        )
+    return state
+
+
+def can_unlock_next_stage(stage_id: str, state: PipelineState) -> bool:
+    stage = state.stages[stage_id]
+    if stage.score is None or stage.score < PASSING_SCORE:
+        return False
+    if has_unresolved_blockers(stage_id, state):
+        return False
+    if stage.is_stale:
+        return False
+    if not has_valid_human_approval(stage_id, state):
+        return False
+    current_artifact = get_current_stage_artifact(stage_id, state)
+    if current_artifact is None:
+        return False
+    return current_artifact.status in {ArtifactStatus.APPROVED, ArtifactStatus.EXPORTED}
+
+
+def get_stage_lock_reasons(stage_id: str, state: PipelineState) -> list[str]:
+    reasons: list[str] = []
+    stage = state.stages[stage_id]
+    if stage.is_stale:
+        reasons.append("Stage is stale.")
+
+    if stage_id == STAGE_IDS[0]:
+        return reasons
+
+    previous_stage_id = STAGE_IDS[STAGE_IDS.index(stage_id) - 1]
+    previous = state.stages[previous_stage_id]
+    if previous.score is None:
+        reasons.append("Previous stage has not been scored.")
+    elif previous.score < PASSING_SCORE:
+        reasons.append("Previous stage score is below 80.")
+
+    if has_unresolved_blockers(previous_stage_id, state):
+        reasons.append("Previous stage has unresolved blockers.")
+
+    previous_artifact = get_current_stage_artifact(previous_stage_id, state)
+    if previous_artifact is None:
+        reasons.append("Current artifact is missing.")
+    elif previous_artifact.status not in {ArtifactStatus.APPROVED, ArtifactStatus.EXPORTED}:
+        reasons.append("Current artifact is not approved.")
+
+    if not has_valid_human_approval(previous_stage_id, state):
+        invalidated = any(
+            approval.stage_id == previous_stage_id
+            and approval.approval_status == "invalidated"
+            for approval in state.approvals
+        )
+        reasons.append("Human approval was invalidated." if invalidated else "Human approval is missing.")
+
+    if previous.is_stale:
+        reasons.append("Previous stage is stale.")
+    return list(dict.fromkeys(reasons))
+
+
+def _visible_words(text: str) -> int:
+    return len(re.findall(r"\b[\w'-]+\b", text))
+
+
+def _slide_visible_text(slide_id: str, state: PipelineState) -> str:
+    slide = state.slides[slide_id]
+    return " ".join([slide.title or "", slide.visible_text, " ".join(slide.bullet_points)]).strip()
+
+
+def compute_objective_traces(state: PipelineState) -> list[ObjectiveTrace]:
+    traces: list[ObjectiveTrace] = []
+    for objective_id, objective_text in state.objectives.items():
+        mapped = [
+            slide.slide_id
+            for slide in state.slides.values()
+            if objective_id in slide.objective_ids
+        ]
+        evidence = [
+            state.slides[slide_id].title or state.slides[slide_id].visible_text[:80]
+            for slide_id in mapped
+        ]
+        issue_ids: list[str] = []
+        if not mapped:
+            coverage_score = 0
+            coverage_status = "uncovered"
+            issue = upsert_issue(
+                state,
+                IssueType.OBJECTIVE_UNCOVERED,
+                IssueSeverity.BLOCKER,
+                f"Learning objective {objective_id} is not mapped to any slide.",
+                stage_id="slide_outline_order",
+                objective_id=objective_id,
+                suggested_fix="Map this objective to at least one slide.",
+            )
+            issue_ids.append(issue.issue_id)
+        else:
+            explicit_scores = [
+                state.slides[slide_id].objective_coverage_scores.get(objective_id)
+                for slide_id in mapped
+                if objective_id in state.slides[slide_id].objective_coverage_scores
+            ]
+            if explicit_scores:
+                coverage_score = clamp_score(sum(explicit_scores) / len(explicit_scores))
+            elif len(mapped) >= 2:
+                coverage_score = 90
+            else:
+                coverage_score = 75
+
+            if coverage_score < 50:
+                coverage_status = "weak"
+                issue = upsert_issue(
+                    state,
+                    IssueType.OBJECTIVE_WEAKLY_COVERED,
+                    IssueSeverity.MAJOR,
+                    f"Learning objective {objective_id} is only weakly covered.",
+                    stage_id="slide_outline_order",
+                    objective_id=objective_id,
+                    suggested_fix="Add stronger evidence or another mapped slide.",
+                )
+                issue_ids.append(issue.issue_id)
+            elif coverage_score < 80:
+                coverage_status = "partial"
+            else:
+                coverage_status = "strong"
+
+        trace = ObjectiveTrace(
+            objective_id=objective_id,
+            objective_text=objective_text,
+            mapped_slide_ids=mapped,
+            coverage_score=coverage_score,
+            coverage_status=coverage_status,  # type: ignore[arg-type]
+            evidence=evidence,
+            issue_ids=issue_ids,
+        )
+        state.objective_traces[objective_id] = trace
+        traces.append(trace)
+        record_audit(
+            state,
+            "objective_trace_updated",
+            objective_id=objective_id,
+            metadata={"coverage_status": trace.coverage_status, "coverage_score": trace.coverage_score},
+        )
+    return traces
+
+
+def validate_claim_support(state: PipelineState) -> list[QualityIssue]:
+    issues: list[QualityIssue] = []
+    for claim in state.claims.values():
+        unsupported = claim.review_status == "unsupported"
+        missing_support = not claim.source_ids and not claim.source_chunk_ids
+        if not unsupported and not missing_support:
+            continue
+        issue = upsert_issue(
+            state,
+            IssueType.UNSUPPORTED_CLAIM,
+            IssueSeverity.BLOCKER,
+            f"Claim {claim.claim_id} is not source-grounded.",
+            stage_id="technical_review",
+            slide_id=claim.slide_id,
+            claim_id=claim.claim_id,
+            suggested_fix="Attach source chunks or mark the claim supported after review.",
+        )
+        if issue.issue_id not in claim.issue_ids:
+            claim.issue_ids.append(issue.issue_id)
+        issues.append(issue)
+    return issues
+
+
+def check_text_density(slide_id: str, state: PipelineState) -> list[QualityIssue]:
+    slide = state.slides[slide_id]
+    role = slide.pedagogical_role.value
+    limits = TEXT_DENSITY_LIMITS.get(role, TEXT_DENSITY_LIMITS["unknown"])
+    issues: list[QualityIssue] = []
+    visible_words = _visible_words(_slide_visible_text(slide_id, state))
+    if visible_words > limits["max_visible_words"]:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.TEXT_DENSITY_EXCEEDED,
+                IssueSeverity.MAJOR,
+                f"Slide {slide_id} has {visible_words} visible words; limit is "
+                f"{limits['max_visible_words']} for role {role}.",
+                stage_id="text_generation",
+                slide_id=slide_id,
+                suggested_fix="Reduce visible text or move explanation into speaker notes.",
+            )
+        )
+    if len(slide.bullet_points) > limits["max_bullets"]:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.TEXT_DENSITY_EXCEEDED,
+                IssueSeverity.MAJOR,
+                f"Slide {slide_id} has {len(slide.bullet_points)} bullets; limit is "
+                f"{limits['max_bullets']} for role {role}.",
+                stage_id="text_generation",
+                slide_id=slide_id,
+                suggested_fix="Combine or remove bullets.",
+            )
+        )
+    concept_count = slide.distinct_concept_count if slide.distinct_concept_count is not None else 1
+    if len(slide.objective_ids) > 2 or concept_count > 3:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.COGNITIVE_LOAD_HIGH,
+                IssueSeverity.MAJOR,
+                f"Slide {slide_id} carries too many objectives or concepts.",
+                stage_id="pedagogical_review",
+                slide_id=slide_id,
+                suggested_fix="Split the slide or narrow its instructional focus.",
+            )
+        )
+    return issues
+
+
+def check_speaker_notes(
+    slide_id: str,
+    state: PipelineState,
+    *,
+    strict: bool = False,
+) -> list[QualityIssue]:
+    slide = state.slides[slide_id]
+    if slide.pedagogical_role.value not in INSTRUCTIONAL_ROLES:
+        return []
+    notes_text = slide.speaker_notes.notes_text if slide.speaker_notes else None
+    if notes_text and notes_text.strip():
+        return []
+    severity = IssueSeverity.BLOCKER if strict else IssueSeverity.MAJOR
+    stage_id = "pedagogical_review" if strict else "text_generation"
+    return [
+        upsert_issue(
+            state,
+            IssueType.SPEAKER_NOTES_MISSING,
+            severity,
+            f"Instructional slide {slide_id} is missing speaker notes.",
+            stage_id=stage_id,
+            slide_id=slide_id,
+            suggested_fix="Add instructor intent and teaching notes for this slide.",
+        )
+    ]
+
+
+def _slot_text(value: Any) -> str:
+    if isinstance(value, list):
+        return " ".join(str(item) for item in value)
+    if isinstance(value, dict):
+        return json.dumps(value, sort_keys=True)
+    return str(value)
+
+
+def _has_raw_coordinates(layout: LayoutSpec) -> bool:
+    if RAW_COORDINATE_KEYS & set(layout.slot_assignments):
+        return True
+    for value in layout.slot_assignments.values():
+        if isinstance(value, dict) and RAW_COORDINATE_KEYS & set(value):
+            return True
+    return False
+
+
+def validate_layout_spec(layout: LayoutSpec, state: PipelineState) -> list[QualityIssue]:
+    issues: list[QualityIssue] = []
+    if _has_raw_coordinates(layout):
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.LAYOUT_SCHEMA_INVALID,
+                IssueSeverity.BLOCKER,
+                f"Slide {layout.slide_id} layout uses raw coordinates instead of template slots.",
+                stage_id="aesthetic_ordering_visual_composition",
+                slide_id=layout.slide_id,
+                suggested_fix="Use an approved layout_id and named template slots.",
+            )
+        )
+        return issues
+
+    registry_entry = APPROVED_LAYOUTS.get(layout.layout_id)
+    if registry_entry is None:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.LAYOUT_SCHEMA_INVALID,
+                IssueSeverity.BLOCKER,
+                f"Slide {layout.slide_id} uses unknown layout_id {layout.layout_id}.",
+                stage_id="aesthetic_ordering_visual_composition",
+                slide_id=layout.slide_id,
+                suggested_fix="Choose an approved layout ID.",
+            )
+        )
+        return issues
+
+    required_slots = registry_entry["required_slots"]
+    optional_slots = registry_entry["optional_slots"]
+    slot_word_limits = registry_entry["slot_word_limits"]
+    used_slots = set(layout.slot_assignments)
+    missing = required_slots - used_slots  # type: ignore[operator]
+    unknown = used_slots - required_slots - optional_slots  # type: ignore[operator]
+    if missing:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.LAYOUT_SLOT_VIOLATION,
+                IssueSeverity.BLOCKER,
+                f"Slide {layout.slide_id} is missing required layout slots: {sorted(missing)}.",
+                stage_id="aesthetic_ordering_visual_composition",
+                slide_id=layout.slide_id,
+                suggested_fix="Populate all required slots for the selected layout.",
+            )
+        )
+    if unknown:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.LAYOUT_SLOT_VIOLATION,
+                IssueSeverity.MAJOR,
+                f"Slide {layout.slide_id} uses unknown slots: {sorted(unknown)}.",
+                stage_id="aesthetic_ordering_visual_composition",
+                slide_id=layout.slide_id,
+                suggested_fix="Remove arbitrary slots and use the template schema.",
+            )
+        )
+    for slot, limit in slot_word_limits.items():  # type: ignore[union-attr]
+        if slot not in layout.slot_assignments:
+            continue
+        word_count = _visible_words(_slot_text(layout.slot_assignments[slot]))
+        if word_count > limit:
+            issues.append(
+                upsert_issue(
+                    state,
+                    IssueType.LAYOUT_SLOT_VIOLATION,
+                    IssueSeverity.MAJOR,
+                    f"Slide {layout.slide_id} slot {slot} has {word_count} words; limit is {limit}.",
+                    stage_id="aesthetic_ordering_visual_composition",
+                    slide_id=layout.slide_id,
+                    suggested_fix="Shorten text in this layout slot.",
+                )
+            )
+    return issues
+
+
+def check_accessibility(slide_id: str, state: PipelineState) -> list[QualityIssue]:
+    issues: list[QualityIssue] = []
+    slide = state.slides[slide_id]
+    visual_assets = [asset for asset in state.visual_assets.values() if asset.slide_id == slide_id]
+    for asset in visual_assets:
+        meaningful = asset.purpose in {"instructional", "unknown"}
+        if meaningful and not (asset.alt_text and asset.alt_text.strip()):
+            issues.append(
+                upsert_issue(
+                    state,
+                    IssueType.ALT_TEXT_MISSING,
+                    IssueSeverity.MAJOR,
+                    f"Visual asset {asset.asset_id} on slide {slide_id} is missing alt text.",
+                    stage_id="aesthetic_review",
+                    slide_id=slide_id,
+                    suggested_fix="Add concise alt text for meaningful visuals.",
+                )
+            )
+    has_visible_text = bool(_slide_visible_text(slide_id, state).strip())
+    has_notes = bool(slide.speaker_notes and slide.speaker_notes.notes_text)
+    if visual_assets and not has_visible_text and not has_notes:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.ALT_TEXT_MISSING,
+                IssueSeverity.MAJOR,
+                f"Image-only slide {slide_id} lacks accessible text or speaker notes.",
+                stage_id="aesthetic_review",
+                slide_id=slide_id,
+                suggested_fix="Add accessible slide text or speaker notes.",
+            )
+        )
+    contrast = slide.metadata.get("contrast_ratio")
+    if isinstance(contrast, int | float) and contrast < 4.5:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.ACCESSIBILITY_CONTRAST_RISK,
+                IssueSeverity.MAJOR,
+                f"Slide {slide_id} has a contrast risk below 4.5:1.",
+                stage_id="aesthetic_review",
+                slide_id=slide_id,
+                suggested_fix="Adjust foreground/background colors in the template metadata.",
+            )
+        )
+    return issues
+
+
+def check_visual_assets(slide_id: str, state: PipelineState) -> list[QualityIssue]:
+    slide = state.slides[slide_id]
+    assets = [asset for asset in state.visual_assets.values() if asset.slide_id == slide_id]
+    issues: list[QualityIssue] = []
+    if slide.requires_visual and not assets:
+        issues.append(
+            upsert_issue(
+                state,
+                IssueType.VISUAL_ASSET_MISSING,
+                IssueSeverity.BLOCKER,
+                f"Slide {slide_id} requires a visual asset but none exists.",
+                stage_id="image_visual_asset_generation",
+                slide_id=slide_id,
+                suggested_fix="Generate, upload, or assign a visual asset.",
+            )
+        )
+    for asset in assets:
+        if asset.purpose == "unknown":
+            issues.append(
+                upsert_issue(
+                    state,
+                    IssueType.VISUAL_ASSET_PURPOSE_MISSING,
+                    IssueSeverity.MAJOR,
+                    f"Visual asset {asset.asset_id} has no instructional/decorative purpose.",
+                    stage_id="image_visual_asset_generation",
+                    slide_id=slide_id,
+                    suggested_fix="Mark the asset purpose before export.",
+                )
+            )
+        if asset.purpose in {"instructional", "unknown"} and not asset.alt_text:
+            issues.append(
+                upsert_issue(
+                    state,
+                    IssueType.ALT_TEXT_MISSING,
+                    IssueSeverity.MAJOR,
+                    f"Meaningful visual asset {asset.asset_id} is missing alt text.",
+                    stage_id="image_visual_asset_generation",
+                    slide_id=slide_id,
+                    suggested_fix="Add alt text.",
+                )
+            )
+        if asset.license_status in {"unknown", "needs_review"}:
+            issues.append(
+                upsert_issue(
+                    state,
+                    IssueType.COPYRIGHT_OR_LICENSE_RISK,
+                    IssueSeverity.MAJOR,
+                    f"Visual asset {asset.asset_id} has unresolved license status.",
+                    stage_id="image_visual_asset_generation",
+                    slide_id=slide_id,
+                    suggested_fix="Use generated/user-provided assets or review the license.",
+                )
+            )
+    return issues
+
+
+def compute_slide_status(slide_id: str, state: PipelineState) -> SlideQAStatus:
+    slide = state.slides[slide_id]
+    check_text_density(slide_id, state)
+    check_speaker_notes(slide_id, state)
+    check_visual_assets(slide_id, state)
+    check_accessibility(slide_id, state)
+    layout = state.layout_specs.get(slide_id)
+    if layout:
+        validate_layout_spec(layout, state)
+
+    issue_ids = [
+        issue.issue_id
+        for issue in unresolved_issues(state)
+        if issue.slide_id == slide_id
+    ]
+    slide_issues = [state.issues[issue_id] for issue_id in issue_ids]
+    blocker_count = sum(issue.severity == IssueSeverity.BLOCKER for issue in slide_issues)
+    major_count = sum(issue.severity == IssueSeverity.MAJOR for issue in slide_issues)
+    aggregate_score = clamp_score(100 - blocker_count * 45 - major_count * 15)
+    technical_score = 0 if any(issue.issue_type == IssueType.UNSUPPORTED_CLAIM for issue in slide_issues) else 100
+    pedagogical_score = clamp_score(
+        100
+        - 20
+        * sum(
+            issue.issue_type
+            in {
+                IssueType.SPEAKER_NOTES_MISSING,
+                IssueType.COGNITIVE_LOAD_HIGH,
+                IssueType.TEXT_DENSITY_EXCEEDED,
+            }
+            for issue in slide_issues
+        )
+    )
+    aesthetic_score = clamp_score(
+        100
+        - 20
+        * sum(
+            issue.issue_type
+            in {
+                IssueType.LAYOUT_SCHEMA_INVALID,
+                IssueType.LAYOUT_SLOT_VIOLATION,
+                IssueType.ALT_TEXT_MISSING,
+                IssueType.ACCESSIBILITY_CONTRAST_RISK,
+            }
+            for issue in slide_issues
+        )
+    )
+    stale = any(state.stages[stage_id].is_stale for stage_id in SLIDE_REVIEW_STAGE_IDS)
+    if blocker_count:
+        status = "blocked"
+    elif stale:
+        status = "stale"
+    elif major_count or aggregate_score < PASSING_SCORE:
+        status = "needs_revision"
+    elif all(has_valid_human_approval(stage_id, state) for stage_id in ("technical_review", "pedagogical_review", "aesthetic_review")):
+        status = "approved"
+    else:
+        status = "needs_review"
+
+    qa_status = SlideQAStatus(
+        slide_id=slide_id,
+        slide_number=slide.slide_number,
+        title=slide.title,
+        pedagogical_role=slide.pedagogical_role,
+        objective_ids=slide.objective_ids,
+        claim_ids=[claim.claim_id for claim in state.claims.values() if claim.slide_id == slide_id],
+        visual_asset_ids=[asset.asset_id for asset in state.visual_assets.values() if asset.slide_id == slide_id],
+        layout_id=layout.layout_id if layout else None,
+        aggregate_score=aggregate_score,
+        technical_score=technical_score,
+        pedagogical_score=pedagogical_score,
+        aesthetic_score=aesthetic_score,
+        issue_ids=issue_ids,
+        stale=stale,
+        status=status,  # type: ignore[arg-type]
+    )
+    state.slide_statuses[slide_id] = qa_status
+    record_audit(
+        state,
+        "slide_status_updated",
+        slide_id=slide_id,
+        metadata={"status": qa_status.status, "aggregate_score": qa_status.aggregate_score},
+    )
+    return qa_status
+
+
+def _coverage_average(state: PipelineState) -> int:
+    if not state.objective_traces:
+        compute_objective_traces(state)
+    if not state.objective_traces:
+        return 0
+    return clamp_score(
+        sum(trace.coverage_score for trace in state.objective_traces.values())
+        / len(state.objective_traces)
+    )
+
+
+def _deduct_for_issue_types(
+    state: PipelineState,
+    issue_types: set[IssueType],
+    *,
+    stage_id: str | None = None,
+) -> int:
+    issues = [
+        issue
+        for issue in unresolved_issues(state)
+        if issue.issue_type in issue_types and (stage_id is None or issue.stage_id == stage_id)
+    ]
+    if any(issue.severity == IssueSeverity.BLOCKER for issue in issues):
+        return 0
+    penalty = sum(45 if issue.severity == IssueSeverity.BLOCKER else 15 for issue in issues)
+    return clamp_score(100 - penalty)
+
+
+def _score_dimension(stage_id: str, dimension_id: str, state: PipelineState) -> tuple[int, str]:
+    if stage_id == "setup_inputs":
+        required = [state.deck_title, state.source_url, state.template_url]
+        if dimension_id == "required_inputs_present":
+            return (100 if all(required) else 30, "Deck title, source URL, and template URL are checked.")
+        if dimension_id == "url_validity":
+            urls = [state.source_url or "", state.template_url or ""]
+            valid = all(url.startswith(("http://", "https://", "mock://")) for url in urls)
+            return (100 if valid else 40, "Source and template URLs must be explicit URLs.")
+        safe = state.dry_run or bool(state.output_folder_id)
+        if state.mutation_target_url and state.mutation_target_url in {state.source_url, state.template_url}:
+            safe = False
+        return (100 if safe else 0, "Render/export targets must be safe.")
+
+    if stage_id == "source_extraction_objective_mapping":
+        if dimension_id == "source_extraction_completeness":
+            return (100 if state.source_chunks else 40, "At least one source chunk is required.")
+        if dimension_id == "objective_clarity":
+            clear = state.objectives and all(_visible_words(text) >= 3 for text in state.objectives.values())
+            return (100 if clear else 45, "Objectives should be readable, specific statements.")
+        objective_texts = [text.strip().lower() for text in state.objectives.values()]
+        unique = len(objective_texts) == len(set(objective_texts))
+        return (100 if unique else 55, "Objectives should not duplicate one another.")
+
+    if stage_id == "slide_outline_order":
+        compute_objective_traces(state)
+        if dimension_id == "objective_coverage":
+            return (_coverage_average(state), "Coverage is aggregated from objective traces.")
+        if dimension_id == "logical_flow":
+            ordered = [slide.slide_number for slide in state.slides.values()]
+            roles_known = all(slide.pedagogical_role.value != "unknown" for slide in state.slides.values())
+            return (100 if ordered == sorted(ordered) and roles_known else 65, "Slide order and roles are checked.")
+        if dimension_id == "appropriate_slide_count":
+            upper = max(1, len(state.objectives) * 3 + 2)
+            ok = 1 <= len(state.slides) <= upper
+            return (100 if ok else 60, "Slide count should fit the number of objectives.")
+        titles = [slide.title for slide in state.slides.values() if slide.title]
+        unique = len(titles) == len(set(titles))
+        return (100 if unique else 65, "Slide titles are used as a redundancy signal.")
+
+    if stage_id == "title_generation":
+        titled = [slide for slide in state.slides.values() if slide.title and len(slide.title.split()) >= 2]
+        if dimension_id == "title_clarity":
+            return (100 if len(titled) == len(state.slides) and state.slides else 45, "Every slide needs a clear title.")
+        if dimension_id == "title_specificity":
+            specific = all(slide.title and _visible_words(slide.title) >= 2 for slide in state.slides.values())
+            return (100 if specific else 60, "Titles should be more specific than a section label.")
+        aligned = all(slide.objective_ids for slide in state.slides.values())
+        return (100 if aligned else 60, "Titles are expected on objective-mapped slides.")
+
+    if stage_id == "text_generation":
+        for slide_id in state.slides:
+            check_text_density(slide_id, state)
+            check_speaker_notes(slide_id, state)
+        validate_claim_support(state)
+        if dimension_id == "clarity":
+            populated = all(_slide_visible_text(slide_id, state) for slide_id in state.slides)
+            return (100 if populated else 45, "Slides should contain visible instructional text.")
+        if dimension_id == "text_density":
+            return (
+                _deduct_for_issue_types(
+                    state,
+                    {IssueType.TEXT_DENSITY_EXCEEDED, IssueType.COGNITIVE_LOAD_HIGH},
+                ),
+                "Density and cognitive-load checks are deterministic.",
+            )
+        if dimension_id == "source_grounding":
+            return (_deduct_for_issue_types(state, {IssueType.UNSUPPORTED_CLAIM}), "Claims must be source grounded.")
+        return (
+            _deduct_for_issue_types(state, {IssueType.SPEAKER_NOTES_MISSING}),
+            "Instructional roles should have speaker notes.",
+        )
+
+    if stage_id == "image_visual_asset_generation":
+        for slide_id in state.slides:
+            check_visual_assets(slide_id, state)
+        if dimension_id == "asset_completeness":
+            return (_deduct_for_issue_types(state, {IssueType.VISUAL_ASSET_MISSING}), "Required visuals must exist.")
+        if dimension_id == "alt_text_presence":
+            return (_deduct_for_issue_types(state, {IssueType.ALT_TEXT_MISSING}), "Meaningful visuals need alt text.")
+        if dimension_id == "license_or_generation_metadata":
+            return (
+                _deduct_for_issue_types(state, {IssueType.COPYRIGHT_OR_LICENSE_RISK}),
+                "Visual sources and license metadata are checked.",
+            )
+        return (
+            _deduct_for_issue_types(state, {IssueType.VISUAL_ASSET_PURPOSE_MISSING}),
+            "Visual purpose metadata is checked.",
+        )
+
+    if stage_id == "aesthetic_ordering_visual_composition":
+        for layout in state.layout_specs.values():
+            validate_layout_spec(layout, state)
+        if dimension_id == "layout_schema_validity":
+            return (_deduct_for_issue_types(state, {IssueType.LAYOUT_SCHEMA_INVALID}), "Layout IDs must be approved.")
+        if dimension_id == "slot_compliance":
+            return (_deduct_for_issue_types(state, {IssueType.LAYOUT_SLOT_VIOLATION}), "Slots must fit layout schema.")
+        if dimension_id == "visual_hierarchy":
+            return (100 if state.layout_specs else 50, "Slides should have explicit layouts.")
+        return (_deduct_for_issue_types(state, {IssueType.TEXT_DENSITY_EXCEEDED}), "Readability uses density signals.")
+
+    if stage_id == "technical_review":
+        validate_claim_support(state)
+        if dimension_id == "unsupported_claim_detection":
+            return (_deduct_for_issue_types(state, {IssueType.UNSUPPORTED_CLAIM}), "Unsupported claims are blockers.")
+        if dimension_id == "technical_accuracy":
+            return (_deduct_for_issue_types(state, {IssueType.TECHNICAL_INACCURACY}), "Technical issues are tracked.")
+        return (100 if state.claims else 85, "Claims provide terminology review surface.")
+
+    if stage_id == "pedagogical_review":
+        compute_objective_traces(state)
+        for slide_id in state.slides:
+            check_text_density(slide_id, state)
+            check_speaker_notes(slide_id, state, strict=True)
+        if dimension_id == "objective_coverage":
+            return (_coverage_average(state), "Objective coverage is aggregated from traces.")
+        if dimension_id == "speaker_notes_teachability":
+            return (_deduct_for_issue_types(state, {IssueType.SPEAKER_NOTES_MISSING}), "Speaker notes are required.")
+        if dimension_id == "examples_or_applications":
+            has_example = any(
+                slide.pedagogical_role.value in {"worked_example", "knowledge_check"}
+                for slide in state.slides.values()
+            )
+            return (100 if has_example else 75, "At least one practice/application role is preferred.")
+        return (100 if state.slides else 0, "Slides should progress through an ordered sequence.")
+
+    if stage_id == "aesthetic_review":
+        for slide_id in state.slides:
+            check_accessibility(slide_id, state)
+        for layout in state.layout_specs.values():
+            validate_layout_spec(layout, state)
+        if dimension_id == "accessibility":
+            return (
+                _deduct_for_issue_types(
+                    state,
+                    {IssueType.ALT_TEXT_MISSING, IssueType.ACCESSIBILITY_CONTRAST_RISK},
+                ),
+                "P0 accessibility checks alt text and contrast metadata only.",
+            )
+        if dimension_id == "brand_template_compliance":
+            compliant = all(layout.approved_template_id for layout in state.layout_specs.values())
+            return (100 if compliant and state.layout_specs else 65, "Layouts should bind to an approved template.")
+        if dimension_id == "layout_consistency":
+            known = all(layout.layout_id in APPROVED_LAYOUTS for layout in state.layout_specs.values())
+            return (100 if known and state.layout_specs else 45, "Layouts must be from the approved registry.")
+        return (_deduct_for_issue_types(state, {IssueType.TEXT_DENSITY_EXCEEDED}), "Readability uses density signals.")
+
+    if stage_id == "final_render_export":
+        report = run_export_preflight(state)
+        if dimension_id == "preflight_passed":
+            return (100 if report.can_export else 0, report.summary)
+        if dimension_id == "all_required_approvals_valid":
+            valid_count = sum(has_valid_human_approval(stage_id, state) for stage_id in APPROVAL_REQUIRED_STAGE_IDS)
+            return (
+                clamp_score(100 * valid_count / len(APPROVAL_REQUIRED_STAGE_IDS)),
+                "All stages 1-10 require valid approvals.",
+            )
+        if dimension_id == "no_stale_stages":
+            stale_count = sum(stage.is_stale for stage in state.stages.values())
+            return (100 if stale_count == 0 else 0, "Stale stages block export.")
+        unsafe = state.mutation_target_url and state.mutation_target_url in {state.source_url, state.template_url}
+        return (0 if unsafe else 100, "Export target safety is checked.")
+
+    if stage_id == "audit_log_version_history":
+        if dimension_id == "audit_events_present":
+            return (100 if state.audit_events else 50, "Audit events should exist.")
+        if dimension_id == "artifact_versions_traceable":
+            traceable = all(artifact.content_hash for artifact in state.artifacts.values())
+            return (100 if traceable and state.artifacts else 50, "Artifact hashes provide traceability.")
+        return (100 if state.reviewer_notes else 80, "Reviewer notes are tracked when present.")
+
+    return (80, "Default deterministic rubric score.")
+
+
+def grade_stage(stage_id: str, state: PipelineState) -> StageGradeResult:
+    artifact = get_current_stage_artifact(stage_id, state)
+    rubric_scores: list[RubricDimensionScore] = []
+    for dimension_id, label, weight in STAGE_RUBRICS[stage_id]:
+        dimension_score, rationale = _score_dimension(stage_id, dimension_id, state)
+        rubric_scores.append(
+            RubricDimensionScore(
+                dimension_id=dimension_id,
+                label=label,
+                score=dimension_score,
+                weight=weight,
+                rationale=rationale,
+            )
+        )
+        record_audit(
+            state,
+            "rubric_score_created",
+            stage_id=stage_id,
+            artifact_version_id=artifact.artifact_version_id if artifact else None,
+            metadata={"dimension_id": dimension_id, "score": dimension_score},
+        )
+    score = aggregate_rubric_score(rubric_scores)
+    issue_ids = [issue.issue_id for issue in unresolved_stage_issues(stage_id, state)]
+    recommended_actions = [
+        issue.suggested_fix or issue.message
+        for issue in unresolved_stage_issues(stage_id, state)
+        if issue.severity in {IssueSeverity.MAJOR, IssueSeverity.BLOCKER}
+    ]
+    result = StageGradeResult(
+        stage_id=stage_id,
+        artifact_version_id=artifact.artifact_version_id if artifact else None,
+        score=score,
+        passed_threshold=score >= PASSING_SCORE,
+        rubric_scores=rubric_scores,
+        issue_ids=issue_ids,
+        recommended_actions=list(dict.fromkeys(recommended_actions)),
+        graded_at=now_iso(),
+    )
+    state.stages[stage_id].score = score
+    state.stages[stage_id].grade_result = result
+    record_audit(
+        state,
+        "stage_grade_created",
+        stage_id=stage_id,
+        artifact_version_id=result.artifact_version_id,
+        metadata={"score": score, "passed_threshold": result.passed_threshold},
+    )
+    return result
+
+
+def compute_deck_health_summary(state: PipelineState) -> DeckHealthSummary:
+    if state.objectives:
+        compute_objective_traces(state)
+    for slide_id in state.slides:
+        compute_slide_status(slide_id, state)
+    unresolved = unresolved_issues(state)
+    blockers = [issue for issue in unresolved if issue.severity == IssueSeverity.BLOCKER]
+    majors = [issue for issue in unresolved if issue.severity == IssueSeverity.MAJOR]
+    slide_scores = [
+        status.aggregate_score
+        for status in state.slide_statuses.values()
+        if status.aggregate_score is not None
+    ]
+    invalidated_approvals = [
+        approval for approval in state.approvals if approval.approval_status == "invalidated"
+    ]
+    unsupported_claims = [
+        issue for issue in unresolved if issue.issue_type == IssueType.UNSUPPORTED_CLAIM
+    ]
+    objectives_strong = sum(
+        trace.coverage_status == "strong" for trace in state.objective_traces.values()
+    )
+    objectives_partial_or_weak = sum(
+        trace.coverage_status in {"partial", "weak"}
+        for trace in state.objective_traces.values()
+    )
+    objectives_uncovered = sum(
+        trace.coverage_status == "uncovered" for trace in state.objective_traces.values()
+    )
+    can_export_now = (
+        not blockers
+        and not invalidated_approvals
+        and not any(stage.is_stale for stage in state.stages.values())
+        and all(has_valid_human_approval(stage_id, state) for stage_id in APPROVAL_REQUIRED_STAGE_IDS)
+    )
+    return DeckHealthSummary(
+        job_id=state.job_id,
+        deck_title=state.deck_title,
+        approved_stage_count=sum(
+            has_valid_human_approval(stage_id, state) for stage_id in state.stages
+        ),
+        total_stage_count=len(state.stages),
+        stale_stage_count=sum(stage.is_stale for stage in state.stages.values()),
+        invalidated_approval_count=len(invalidated_approvals),
+        unresolved_blocker_count=len(blockers),
+        unresolved_major_issue_count=len(majors),
+        slide_count=len(state.slides),
+        average_slide_score=(
+            clamp_score(sum(slide_scores) / len(slide_scores)) if slide_scores else None
+        ),
+        objectives_total=len(state.objectives),
+        objectives_strong=objectives_strong,
+        objectives_partial_or_weak=objectives_partial_or_weak,
+        objectives_uncovered=objectives_uncovered,
+        unsupported_claim_count=len(unsupported_claims),
+        can_export=can_export_now,
+        top_blockers=[issue.message for issue in blockers[:5]],
+    )
+
+
+def compute_artifact_diff(previous_content: str | dict, current_content: str | dict) -> str:
+    if isinstance(previous_content, dict):
+        previous = json.dumps(previous_content, sort_keys=True, indent=2).splitlines(keepends=True)
+    else:
+        previous = str(previous_content).splitlines(keepends=True)
+    if isinstance(current_content, dict):
+        current = json.dumps(current_content, sort_keys=True, indent=2).splitlines(keepends=True)
+    else:
+        current = str(current_content).splitlines(keepends=True)
+    diff = difflib.unified_diff(previous, current, fromfile="previous", tofile="current")
+    return "".join(diff)
+
+
+def _preflight_blocker(
+    state: PipelineState,
+    message: str,
+    *,
+    stage_id: str = "final_render_export",
+    slide_id: str | None = None,
+    artifact_version_id: str | None = None,
+    suggested_fix: str | None = None,
+) -> QualityIssue:
+    return upsert_issue(
+        state,
+        IssueType.EXPORT_PREFLIGHT_FAILED,
+        IssueSeverity.BLOCKER,
+        message,
+        stage_id=stage_id,
+        slide_id=slide_id,
+        artifact_version_id=artifact_version_id,
+        suggested_fix=suggested_fix,
+    )
+
+
+def _check_preflight_prerequisites(state: PipelineState) -> None:
+    if not state.deck_title or not state.source_url or not state.template_url:
+        upsert_issue(
+            state,
+            IssueType.MISSING_REQUIRED_INPUT,
+            IssueSeverity.BLOCKER,
+            "Deck title, source URL, and template URL are required.",
+            stage_id="setup_inputs",
+            suggested_fix="Complete setup inputs before export.",
+        )
+    for stage_id in APPROVAL_REQUIRED_STAGE_IDS:
+        stage = state.stages[stage_id]
+        artifact = get_current_stage_artifact(stage_id, state)
+        if stage.score is None or stage.score < PASSING_SCORE:
+            upsert_issue(
+                state,
+                IssueType.SCORE_BELOW_THRESHOLD,
+                IssueSeverity.BLOCKER,
+                f"{stage_id} must be scored at 80 or higher before export.",
+                stage_id=stage_id,
+                suggested_fix="Grade or improve this stage.",
+            )
+        if not has_valid_human_approval(stage_id, state):
+            upsert_issue(
+                state,
+                IssueType.HUMAN_APPROVAL_MISSING,
+                IssueSeverity.BLOCKER,
+                f"{stage_id} does not have valid approval for its current artifact.",
+                stage_id=stage_id,
+                artifact_version_id=artifact.artifact_version_id if artifact else None,
+                suggested_fix="Approve the current passing artifact.",
+            )
+        if artifact is None:
+            _preflight_blocker(
+                state,
+                f"{stage_id} has no current artifact.",
+                stage_id=stage_id,
+                suggested_fix="Generate this stage.",
+            )
+        elif artifact.status not in {ArtifactStatus.APPROVED, ArtifactStatus.EXPORTED}:
+            _preflight_blocker(
+                state,
+                f"{stage_id} current artifact is {artifact.status.value}; export requires approved artifacts.",
+                stage_id=stage_id,
+                artifact_version_id=artifact.artifact_version_id,
+                suggested_fix="Approve the current artifact.",
+            )
+    for stage_id, stage in state.stages.items():
+        if stage.is_stale:
+            upsert_issue(
+                state,
+                IssueType.STALE_DOWNSTREAM_STAGE,
+                IssueSeverity.BLOCKER,
+                f"{stage_id} is stale and cannot be exported.",
+                stage_id=stage_id,
+                suggested_fix="Regenerate or re-grade stale stages.",
+            )
+
+
+def _check_preflight_slide_requirements(state: PipelineState) -> None:
+    if not state.slides:
+        _preflight_blocker(
+            state,
+            "Slide inventory is missing.",
+            suggested_fix="Generate a slide outline before export.",
+        )
+        return
+    for slide_id, slide in state.slides.items():
+        if not slide.title:
+            _preflight_blocker(
+                state,
+                f"Slide {slide_id} is missing a title.",
+                slide_id=slide_id,
+                suggested_fix="Generate or edit slide titles.",
+            )
+        if slide.pedagogical_role.value == "unknown":
+            _preflight_blocker(
+                state,
+                f"Slide {slide_id} is missing a pedagogical role.",
+                slide_id=slide_id,
+                suggested_fix="Assign a pedagogical role.",
+            )
+        check_speaker_notes(slide_id, state, strict=True)
+        check_text_density(slide_id, state)
+        check_visual_assets(slide_id, state)
+        check_accessibility(slide_id, state)
+        if slide_id not in state.layout_specs:
+            upsert_issue(
+                state,
+                IssueType.LAYOUT_SCHEMA_INVALID,
+                IssueSeverity.BLOCKER,
+                f"Slide {slide_id} is missing layout JSON.",
+                stage_id="aesthetic_ordering_visual_composition",
+                slide_id=slide_id,
+                suggested_fix="Assign an approved template-first layout.",
+            )
+        else:
+            validate_layout_spec(state.layout_specs[slide_id], state)
+        status = compute_slide_status(slide_id, state)
+        if status.aggregate_score is None or status.aggregate_score < PASSING_SCORE:
+            _preflight_blocker(
+                state,
+                f"Slide {slide_id} aggregate score is below 80.",
+                slide_id=slide_id,
+                suggested_fix="Resolve slide-level issues.",
+            )
+
+
+def _check_preflight_render_safety(state: PipelineState) -> None:
+    if not state.dry_run and not state.output_folder_id:
+        upsert_issue(
+            state,
+            IssueType.MISSING_REQUIRED_INPUT,
+            IssueSeverity.BLOCKER,
+            "Output folder ID is required when dry-run mode is off.",
+            stage_id="setup_inputs",
+            suggested_fix="Provide an output folder ID or keep dry-run mode enabled.",
+        )
+    if state.production_export_requested and state.dry_run:
+        upsert_issue(
+            state,
+            IssueType.RENDER_SAFETY_VIOLATION,
+            IssueSeverity.BLOCKER,
+            "Dry-run mode cannot perform a production export.",
+            stage_id="final_render_export",
+            suggested_fix="Disable dry-run mode only after all export targets are safe.",
+        )
+    if state.mutation_target_url and state.mutation_target_url in {state.source_url, state.template_url}:
+        upsert_issue(
+            state,
+            IssueType.RENDER_SAFETY_VIOLATION,
+            IssueSeverity.BLOCKER,
+            "Source/template URLs cannot be used as mutation targets.",
+            stage_id="final_render_export",
+            suggested_fix="Use a separate output folder or generated deck target.",
+        )
+
+
+def run_export_preflight(state: PipelineState) -> ExportPreflightReport:
+    record_audit(state, "preflight_run", stage_id="final_render_export")
+    _check_preflight_prerequisites(state)
+    if state.objectives:
+        compute_objective_traces(state)
+    else:
+        upsert_issue(
+            state,
+            IssueType.OBJECTIVE_UNCOVERED,
+            IssueSeverity.BLOCKER,
+            "No learning objectives exist.",
+            stage_id="source_extraction_objective_mapping",
+            suggested_fix="Extract or provide learning objectives.",
+        )
+    validate_claim_support(state)
+    _check_preflight_slide_requirements(state)
+    _check_preflight_render_safety(state)
+
+    blocking_issue_ids = [
+        issue.issue_id
+        for issue in unresolved_issues(state)
+        if issue.severity == IssueSeverity.BLOCKER
+    ]
+    warning_issue_ids = [
+        issue.issue_id
+        for issue in unresolved_issues(state)
+        if issue.severity in {IssueSeverity.MAJOR, IssueSeverity.MINOR}
+    ]
+    can_export = not blocking_issue_ids
+    summary = (
+        "Export preflight passed."
+        if can_export
+        else f"Export preflight failed with {len(blocking_issue_ids)} blocker(s)."
+    )
+    report = ExportPreflightReport(
+        can_export=can_export,
+        checked_at=now_iso(),
+        blocking_issue_ids=blocking_issue_ids,
+        warning_issue_ids=warning_issue_ids,
+        summary=summary,
+    )
+    state.export_preflight_report = report
+    record_audit(
+        state,
+        "preflight_passed" if can_export else "preflight_failed",
+        stage_id="final_render_export",
+        metadata={"blocking_issue_count": len(blocking_issue_ids)},
+    )
+    return report
+
+
+def can_export(state: PipelineState) -> bool:
+    return run_export_preflight(state).can_export

course_slide_factory/workflow.py

@@ -0,0 +1,617 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from .constants import STAGE_IDS, STAGE_LABELS, STAGE_SEQUENCE
+from .compat import model_to_dict, model_validate
+from .models import (
+    ArtifactStatus,
+    LayoutSpec,
+    PedagogicalRole,
+    PipelineState,
+    PromptRun,
+    RequestedChange,
+    ReviewerNotes,
+    Slide,
+    SlideClaim,
+    SpeakerNotes,
+    StageState,
+    VisualAsset,
+)
+from .quality import (
+    approve_current_artifact,
+    can_unlock_next_stage,
+    compute_artifact_diff,
+    compute_deck_health_summary,
+    compute_objective_traces,
+    compute_slide_status,
+    create_artifact_version,
+    get_current_stage_artifact,
+    get_stage_lock_reasons,
+    grade_stage,
+    has_unresolved_blockers,
+    now_iso,
+    record_audit,
+    run_export_preflight,
+    stable_hash,
+)
+
+
+def build_empty_state(
+    *,
+    deck_title: str | None = None,
+    source_url: str | None = None,
+    template_url: str | None = None,
+    output_folder_id: str | None = None,
+    dry_run: bool = True,
+    mutation_target_url: str | None = None,
+    production_export_requested: bool = False,
+) -> PipelineState:
+    job_seed = "|".join([deck_title or "course-deck", source_url or "mock://source"])
+    state = PipelineState(
+        job_id=f"job_{stable_hash(job_seed)[:10]}",
+        deck_title=deck_title,
+        source_url=source_url,
+        template_url=template_url,
+        output_folder_id=output_folder_id,
+        dry_run=dry_run,
+        mutation_target_url=mutation_target_url,
+        production_export_requested=production_export_requested,
+        stages={
+            stage_id: StageState(stage_id=stage_id, label=label)
+            for stage_id, label in STAGE_SEQUENCE
+        },
+    )
+    record_audit(state, "state_initialized", metadata={"dry_run": dry_run})
+    return state
+
+
+def state_to_dict(state: PipelineState) -> dict[str, Any]:
+    return model_to_dict(state)
+
+
+def state_from_dict(data: dict[str, Any] | PipelineState | None) -> PipelineState:
+    if isinstance(data, PipelineState):
+        return data
+    if not data:
+        return build_empty_state()
+    return model_validate(PipelineState, data)
+
+
+def parse_objective_lines(objectives_text: str | None) -> dict[str, str]:
+    objectives: dict[str, str] = {}
+    for index, raw_line in enumerate((objectives_text or "").splitlines(), start=1):
+        line = raw_line.strip(" -\t")
+        if not line:
+            continue
+        if ":" in line and line.split(":", 1)[0].strip().lower().startswith("obj"):
+            objective_id, objective = line.split(":", 1)
+            objectives[objective_id.strip()] = objective.strip()
+        else:
+            objectives[f"obj_{index}"] = line
+    return objectives
+
+
+def update_setup_from_inputs(
+    state: PipelineState,
+    *,
+    deck_title: str | None,
+    source_url: str | None,
+    template_url: str | None,
+    output_folder_id: str | None,
+    dry_run: bool,
+    objectives_text: str | None,
+    source_text: str | None,
+    mutation_target_url: str | None = None,
+    production_export_requested: bool = False,
+) -> PipelineState:
+    state.deck_title = deck_title
+    state.source_url = source_url
+    state.template_url = template_url
+    state.output_folder_id = output_folder_id
+    state.dry_run = dry_run
+    state.mutation_target_url = mutation_target_url
+    state.production_export_requested = production_export_requested
+    parsed_objectives = parse_objective_lines(objectives_text)
+    if parsed_objectives:
+        state.objectives = parsed_objectives
+    if source_text and source_text.strip():
+        state.source_chunks = {"chunk_1": source_text.strip()}
+    return state
+
+
+def record_prompt_run(
+    state: PipelineState,
+    stage_id: str,
+    *,
+    rendered_prompt: str,
+    provider: str = "mock",
+    input_artifact_version_ids: list[str] | None = None,
+) -> PromptRun:
+    prompt_run_id = f"prompt_{len(state.prompt_runs) + 1:05d}"
+    run = PromptRun(
+        prompt_run_id=prompt_run_id,
+        stage_id=stage_id,
+        provider=provider,  # type: ignore[arg-type]
+        model_name="mock-deterministic-v1" if provider == "mock" else None,
+        prompt_template_id=f"{stage_id}_template",
+        prompt_template_version="p0.1",
+        rendered_prompt=rendered_prompt,
+        input_artifact_version_ids=input_artifact_version_ids or [],
+        settings={"temperature": 0, "dry_run": state.dry_run},
+        created_at=now_iso(),
+    )
+    state.prompt_runs[prompt_run_id] = run
+    record_audit(
+        state,
+        "prompt_run_recorded",
+        stage_id=stage_id,
+        metadata={"prompt_run_id": prompt_run_id, "provider": provider},
+    )
+    return run
+
+
+def _ensure_default_objectives_and_source(state: PipelineState) -> None:
+    if not state.source_chunks:
+        state.source_chunks = {
+            "chunk_1": "Mock source chunk describing the key course ideas and examples."
+        }
+    if not state.objectives:
+        state.objectives = {
+            "obj_1": "Explain the core concept in plain language.",
+            "obj_2": "Apply the concept to a worked example.",
+        }
+
+
+def _ensure_default_slides(state: PipelineState) -> None:
+    _ensure_default_objectives_and_source(state)
+    if state.slides:
+        return
+    slides: dict[str, Slide] = {}
+    for index, objective_id in enumerate(state.objectives, start=1):
+        role = PedagogicalRole.CONCEPT if index == 1 else PedagogicalRole.WORKED_EXAMPLE
+        if index == len(state.objectives) and len(state.objectives) > 2:
+            role = PedagogicalRole.SUMMARY
+        slide_id = f"slide_{index}"
+        slides[slide_id] = Slide(
+            slide_id=slide_id,
+            slide_number=index,
+            title=f"{state.objectives[objective_id].rstrip('.')}",
+            visible_text=f"This slide teaches {state.objectives[objective_id].lower()}",
+            bullet_points=["Key idea", "Example", "Takeaway"] if role != PedagogicalRole.CONCEPT else ["Key idea"],
+            objective_ids=[objective_id],
+            pedagogical_role=role,
+            requires_visual=index == 1,
+            speaker_notes=SpeakerNotes(
+                slide_id=slide_id,
+                notes_text=f"Guide learners through objective {objective_id} with one concrete example.",
+                instructor_intent="Keep the explanation concrete and source-grounded.",
+                estimated_teaching_time_seconds=180,
+                possible_student_confusions=["Students may overgeneralize the example."],
+                teaching_tips=["Ask learners to restate the idea before moving on."],
+            ),
+        )
+    state.slides = slides
+
+
+def _ensure_claims(state: PipelineState) -> None:
+    if state.claims:
+        return
+    for slide in state.slides.values():
+        claim_id = f"claim_{slide.slide_number}"
+        state.claims[claim_id] = SlideClaim(
+            claim_id=claim_id,
+            slide_id=slide.slide_id,
+            claim_text=slide.visible_text or slide.title or "Instructional claim",
+            source_ids=["source_1"],
+            source_chunk_ids=list(state.source_chunks)[:1],
+            review_status="supported",
+        )
+        record_audit(
+            state,
+            "claim_created",
+            slide_id=slide.slide_id,
+            claim_id=claim_id,
+            metadata={"review_status": "supported"},
+        )
+
+
+def _ensure_visual_assets(state: PipelineState) -> None:
+    for slide in state.slides.values():
+        if not slide.requires_visual:
+            continue
+        if any(asset.slide_id == slide.slide_id for asset in state.visual_assets.values()):
+            continue
+        asset_id = f"asset_{slide.slide_number}"
+        state.visual_assets[asset_id] = VisualAsset(
+            asset_id=asset_id,
+            slide_id=slide.slide_id,
+            asset_type="diagram",
+            path_or_url=f"mock://assets/{asset_id}",
+            prompt=f"Instructional diagram for {slide.title}",
+            purpose="instructional",
+            alt_text=f"Diagram illustrating {slide.title}",
+            source="mock",
+            license_status="generated",
+            approved_for_export=True,
+        )
+
+
+def _ensure_layouts(state: PipelineState) -> None:
+    for slide in state.slides.values():
+        if slide.slide_id in state.layout_specs:
+            continue
+        if slide.requires_visual:
+            layout_id = "title_bullets_visual"
+            slots = {
+                "title": slide.title or "",
+                "bullets": slide.bullet_points,
+                "visual": "primary_visual",
+            }
+        elif slide.pedagogical_role == PedagogicalRole.WORKED_EXAMPLE:
+            layout_id = "worked_example"
+            slots = {
+                "title": slide.title or "",
+                "problem": slide.visible_text,
+                "steps": slide.bullet_points,
+            }
+        else:
+            layout_id = "title_body"
+            slots = {"title": slide.title or "", "body": slide.visible_text}
+        state.layout_specs[slide.slide_id] = LayoutSpec(
+            slide_id=slide.slide_id,
+            layout_id=layout_id,
+            approved_template_id="default_course_template",
+            slot_assignments=slots,
+        )
+
+
+def generate_stage(state: PipelineState, stage_id: str) -> tuple[PipelineState, str]:
+    if stage_id not in STAGE_IDS:
+        raise ValueError(f"Unknown stage: {stage_id}")
+    input_ids = [
+        artifact.artifact_version_id
+        for artifact in state.artifacts.values()
+        if artifact.stage_id in STAGE_IDS[: STAGE_IDS.index(stage_id)]
+        and artifact.is_current
+    ]
+    prompt_run = record_prompt_run(
+        state,
+        stage_id,
+        rendered_prompt=f"Mock deterministic generation for {STAGE_LABELS[stage_id]}",
+        input_artifact_version_ids=input_ids,
+    )
+
+    if stage_id == "setup_inputs":
+        content = {
+            "deck_title": state.deck_title,
+            "source_url": state.source_url,
+            "template_url": state.template_url,
+            "dry_run": state.dry_run,
+            "output_folder_id": state.output_folder_id,
+        }
+    elif stage_id == "source_extraction_objective_mapping":
+        _ensure_default_objectives_and_source(state)
+        content = {"source_chunks": state.source_chunks, "objectives": state.objectives}
+    elif stage_id == "slide_outline_order":
+        _ensure_default_slides(state)
+        compute_objective_traces(state)
+        content = {"slides": [model_to_dict(slide) for slide in state.slides.values()]}
+    elif stage_id == "title_generation":
+        _ensure_default_slides(state)
+        for slide in state.slides.values():
+            slide.title = slide.title or f"Slide {slide.slide_number}"
+        content = {"titles": {slide.slide_id: slide.title for slide in state.slides.values()}}
+    elif stage_id == "text_generation":
+        _ensure_default_slides(state)
+        _ensure_claims(state)
+        content = {
+            "slides": [model_to_dict(slide) for slide in state.slides.values()],
+            "claims": [model_to_dict(claim) for claim in state.claims.values()],
+        }
+    elif stage_id == "image_visual_asset_generation":
+        _ensure_default_slides(state)
+        _ensure_visual_assets(state)
+        content = {"visual_assets": [model_to_dict(asset) for asset in state.visual_assets.values()]}
+    elif stage_id == "aesthetic_ordering_visual_composition":
+        _ensure_default_slides(state)
+        _ensure_layouts(state)
+        content = {"layout_specs": [model_to_dict(layout) for layout in state.layout_specs.values()]}
+    elif stage_id in {"technical_review", "pedagogical_review", "aesthetic_review"}:
+        content = {
+            "review_stage": stage_id,
+            "checked_at": now_iso(),
+            "slide_count": len(state.slides),
+        }
+    elif stage_id == "final_render_export":
+        report = run_export_preflight(state)
+        content = {"preflight": model_to_dict(report)}
+    else:
+        content = {
+            "audit_event_count": len(state.audit_events),
+            "artifact_count": len(state.artifacts),
+        }
+
+    artifact = create_artifact_version(
+        state,
+        stage_id,
+        content,
+        created_by="mock",
+        status=ArtifactStatus.CANDIDATE,
+        prompt_run_id=prompt_run.prompt_run_id,
+    )
+    prompt_run.output_artifact_version_id = artifact.artifact_version_id
+    prompt_run.artifact_version_id = artifact.artifact_version_id
+    return state, f"Generated {STAGE_LABELS[stage_id]} as {artifact.artifact_version_id}."
+
+
+def grade_stage_action(state: PipelineState, stage_id: str) -> tuple[PipelineState, str]:
+    result = grade_stage(stage_id, state)
+    for slide_id in state.slides:
+        compute_slide_status(slide_id, state)
+    compute_deck_health_summary(state)
+    return state, f"Grade for {STAGE_LABELS[stage_id]}: {result.score}."
+
+
+def save_human_edits(
+    state: PipelineState,
+    stage_id: str,
+    edited_content: str,
+    reviewer_name: str,
+    reviewer_summary: str,
+    requested_changes_json: str,
+) -> tuple[PipelineState, str]:
+    requested_changes: list[RequestedChange] = []
+    if requested_changes_json.strip():
+        parsed = json.loads(requested_changes_json)
+        if not isinstance(parsed, list):
+            raise ValueError("Requested changes JSON must be a list.")
+        requested_changes = [model_validate(RequestedChange, item) for item in parsed]
+    current = get_current_stage_artifact(stage_id, state)
+    notes = ReviewerNotes(
+        stage_id=stage_id,
+        artifact_version_id=current.artifact_version_id if current else None,
+        reviewer_name=reviewer_name or "human_reviewer",
+        summary=reviewer_summary or None,
+        requested_changes=requested_changes,
+        created_at=now_iso(),
+    )
+    state.reviewer_notes.append(notes)
+    record_audit(
+        state,
+        "reviewer_notes_saved",
+        stage_id=stage_id,
+        artifact_version_id=notes.artifact_version_id,
+        metadata={"requested_change_count": len(requested_changes)},
+    )
+    artifact = create_artifact_version(
+        state,
+        stage_id,
+        {"human_edit": edited_content, "reviewer_notes": model_to_dict(notes)},
+        created_by="human",
+        status=ArtifactStatus.CANDIDATE,
+    )
+    return state, f"Saved human edits as candidate {artifact.artifact_version_id}."
+
+
+def improve_with_ai(state: PipelineState, stage_id: str) -> tuple[PipelineState, str]:
+    current = get_current_stage_artifact(stage_id, state)
+    input_ids = [current.artifact_version_id] if current else []
+    prompt_run = record_prompt_run(
+        state,
+        stage_id,
+        rendered_prompt=f"Mock AI improvement for {STAGE_LABELS[stage_id]}",
+        input_artifact_version_ids=input_ids,
+    )
+    current_content = current.metadata.get("content") if current else {}
+    content = {
+        "ai_improved_from": current.artifact_version_id if current else None,
+        "content": current_content,
+        "improvement_note": "Mock AI pass reduced ambiguity and preserved traceability.",
+    }
+    artifact = create_artifact_version(
+        state,
+        stage_id,
+        content,
+        created_by="ai",
+        status=ArtifactStatus.CANDIDATE,
+        prompt_run_id=prompt_run.prompt_run_id,
+    )
+    prompt_run.output_artifact_version_id = artifact.artifact_version_id
+    prompt_run.artifact_version_id = artifact.artifact_version_id
+    return state, f"Created AI-improved candidate {artifact.artifact_version_id}."
+
+
+def approve_and_continue(
+    state: PipelineState,
+    stage_id: str,
+    reviewer_name: str = "human_reviewer",
+) -> tuple[PipelineState, str]:
+    stage = state.stages[stage_id]
+    if stage.score is None or stage.score < 80:
+        return state, "Approval blocked: stage score is below 80 or missing."
+    if has_unresolved_blockers(stage_id, state):
+        return state, "Approval blocked: stage has unresolved blockers."
+    if stage.is_stale:
+        return state, "Approval blocked: stage is stale."
+    if get_current_stage_artifact(stage_id, state) is None:
+        return state, "Approval blocked: current artifact is missing."
+    approval = approve_current_artifact(state, stage_id, reviewer_name)
+    unlocks = can_unlock_next_stage(stage_id, state)
+    message = (
+        f"Approved {approval.artifact_version_id}; next stage can unlock."
+        if unlocks
+        else f"Approved {approval.artifact_version_id}; next stage remains locked."
+    )
+    return state, message
+
+
+def run_preflight_action(state: PipelineState) -> tuple[PipelineState, str]:
+    report = run_export_preflight(state)
+    return state, report.summary
+
+
+def final_render_export(state: PipelineState) -> tuple[PipelineState, str]:
+    report = run_export_preflight(state)
+    if not report.can_export:
+        record_audit(
+            state,
+            "export_blocked_by_preflight",
+            stage_id="final_render_export",
+            metadata={"blocking_issue_ids": report.blocking_issue_ids},
+        )
+        return state, report.summary
+    for stage_id in STAGE_IDS[:11]:
+        artifact = get_current_stage_artifact(stage_id, state)
+        if artifact and artifact.status == ArtifactStatus.APPROVED:
+            artifact.status = ArtifactStatus.EXPORTED
+    record_audit(
+        state,
+        "export_completed",
+        stage_id="final_render_export",
+        metadata={"dry_run": state.dry_run},
+    )
+    return state, "Dry-run export completed." if state.dry_run else "Production export completed."
+
+
+def current_artifact_text(state: PipelineState, stage_id: str) -> str:
+    artifact = get_current_stage_artifact(stage_id, state)
+    if artifact is None:
+        return ""
+    return json.dumps(artifact.metadata.get("content", {}), indent=2, sort_keys=True)
+
+
+def current_artifact_diff(state: PipelineState, stage_id: str) -> str:
+    versions = state.stage_artifact_versions.get(stage_id, [])
+    if len(versions) < 2:
+        return ""
+    previous = state.artifacts[versions[-2]].metadata.get("content", {})
+    current = state.artifacts[versions[-1]].metadata.get("content", {})
+    record_audit(state, "artifact_diff_viewed", stage_id=stage_id)
+    return compute_artifact_diff(previous, current)
+
+
+def deck_health_table(state: PipelineState) -> list[list[Any]]:
+    summary = compute_deck_health_summary(state)
+    return [
+        ["Job ID", summary.job_id],
+        ["Deck title", summary.deck_title or ""],
+        ["Approved stages", f"{summary.approved_stage_count}/{summary.total_stage_count}"],
+        ["Stale stages", summary.stale_stage_count],
+        ["Invalidated approvals", summary.invalidated_approval_count],
+        ["Unresolved blockers", summary.unresolved_blocker_count],
+        ["Unresolved major issues", summary.unresolved_major_issue_count],
+        ["Slide count", summary.slide_count],
+        ["Average slide score", summary.average_slide_score],
+        ["Objectives strong", summary.objectives_strong],
+        ["Objectives partial/weak", summary.objectives_partial_or_weak],
+        ["Objectives uncovered", summary.objectives_uncovered],
+        ["Unsupported claims", summary.unsupported_claim_count],
+        ["Can export", summary.can_export],
+        ["Top blockers", "; ".join(summary.top_blockers)],
+    ]
+
+
+def slide_inventory_table(state: PipelineState) -> list[list[Any]]:
+    for slide_id in state.slides:
+        compute_slide_status(slide_id, state)
+    rows: list[list[Any]] = []
+    for status in sorted(state.slide_statuses.values(), key=lambda item: item.slide_number):
+        rows.append(
+            [
+                status.slide_number,
+                status.title or "",
+                status.pedagogical_role.value,
+                ", ".join(status.objective_ids),
+                ", ".join(status.claim_ids),
+                ", ".join(status.visual_asset_ids),
+                status.aggregate_score,
+                status.technical_score,
+                status.pedagogical_score,
+                status.aesthetic_score,
+                status.status,
+                len(status.issue_ids),
+                status.stale,
+            ]
+        )
+    return rows
+
+
+def objective_matrix_table(state: PipelineState) -> list[list[Any]]:
+    compute_objective_traces(state)
+    return [
+        [
+            trace.objective_id,
+            trace.objective_text,
+            ", ".join(trace.mapped_slide_ids),
+            trace.coverage_score,
+            trace.coverage_status,
+            len(trace.evidence),
+            ", ".join(trace.issue_ids),
+        ]
+        for trace in state.objective_traces.values()
+    ]
+
+
+def stage_status_table(state: PipelineState) -> list[list[Any]]:
+    rows: list[list[Any]] = []
+    for stage_id in STAGE_IDS:
+        stage = state.stages[stage_id]
+        artifact = get_current_stage_artifact(stage_id, state)
+        rows.append(
+            [
+                STAGE_IDS.index(stage_id) + 1,
+                stage.label,
+                stage.score,
+                artifact.artifact_version_id if artifact else "",
+                artifact.status.value if artifact else "",
+                stage.is_stale,
+                can_unlock_next_stage(stage_id, state),
+                "; ".join(get_stage_lock_reasons(stage_id, state)),
+            ]
+        )
+    return rows
+
+
+def issue_table(state: PipelineState) -> list[list[Any]]:
+    return [
+        [
+            issue.issue_id,
+            issue.issue_type.value,
+            issue.severity.value,
+            issue.stage_id or "",
+            issue.slide_id or "",
+            issue.message,
+            issue.resolved,
+        ]
+        for issue in state.issues.values()
+    ]
+
+
+def preflight_table(state: PipelineState) -> list[list[Any]]:
+    report = state.export_preflight_report
+    if report is None:
+        return []
+    return [
+        ["can_export", report.can_export],
+        ["checked_at", report.checked_at],
+        ["blocking_issue_ids", "\n".join(report.blocking_issue_ids)],
+        ["warning_issue_ids", "\n".join(report.warning_issue_ids)],
+        ["summary", report.summary],
+    ]
+
+
+def audit_table(state: PipelineState) -> list[list[Any]]:
+    return [
+        [
+            event.event_id,
+            event.event_type,
+            event.timestamp,
+            event.stage_id or "",
+            event.slide_id or "",
+            event.issue_id or "",
+            event.reason or "",
+        ]
+        for event in state.audit_events[-50:]
+    ]

pyproject.toml

@@ -0,0 +1,12 @@
+[tool.pytest.ini_options]
+pythonpath = ["."]
+testpaths = ["tests"]
+addopts = ["-p", "no:cacheprovider"]
+
+[tool.ruff]
+line-length = 140
+target-version = "py311"
+cache-dir = "/tmp/slide_creation_engine_ruff_cache"
+
+[tool.ruff.lint]
+select = ["E", "F", "B"]

requirements.txt

@@ -0,0 +1,4 @@
+gradio==6.19.0
+pydantic>=2.7
+pytest>=8.0
+ruff>=0.8

tests/test_p0_quality.py

@@ -0,0 +1,253 @@
+from __future__ import annotations
+
+from course_slide_factory.constants import STAGE_IDS
+from course_slide_factory.fixtures import (
+    invalid_layout_job,
+    invalidated_approval_job,
+    missing_objective_mapping_job,
+    missing_visual_asset_job,
+    stale_downstream_job,
+    unsupported_claim_job,
+    valid_minimal_job,
+)
+from course_slide_factory.models import ArtifactStatus, IssueSeverity, IssueType, LayoutSpec
+from course_slide_factory.quality import (
+    aggregate_rubric_score,
+    approve_current_artifact,
+    can_unlock_next_stage,
+    check_text_density,
+    compute_artifact_diff,
+    compute_objective_traces,
+    get_current_stage_artifact,
+    get_stage_lock_reasons,
+    grade_stage,
+    has_valid_human_approval,
+    run_export_preflight,
+    upsert_issue,
+    validate_claim_support,
+    validate_layout_spec,
+)
+from course_slide_factory.workflow import (
+    build_empty_state,
+    final_render_export,
+    generate_stage,
+    improve_with_ai,
+    save_human_edits,
+)
+
+
+def test_weighted_rubric_score_aggregates_and_clamps():
+    from course_slide_factory.models import RubricDimensionScore
+
+    score = aggregate_rubric_score(
+        [
+            RubricDimensionScore(dimension_id="a", label="A", score=100, weight=3),
+            RubricDimensionScore(dimension_id="b", label="B", score=50, weight=1),
+        ]
+    )
+
+    assert score == 88
+
+
+def test_gating_requires_score_no_blockers_approval_and_current_artifact():
+    state = valid_minimal_job()
+    stage_id = "setup_inputs"
+
+    assert can_unlock_next_stage(stage_id, state)
+
+    state.stages[stage_id].score = 79
+    assert not can_unlock_next_stage(stage_id, state)
+
+    state = valid_minimal_job()
+    upsert_issue(
+        state,
+        IssueType.TECHNICAL_INACCURACY,
+        IssueSeverity.BLOCKER,
+        "Blocking issue",
+        stage_id=stage_id,
+    )
+    assert not can_unlock_next_stage(stage_id, state)
+
+    state = valid_minimal_job()
+    state.approvals = [approval for approval in state.approvals if approval.stage_id != stage_id]
+    assert not can_unlock_next_stage(stage_id, state)
+
+
+def test_approval_invalidates_after_human_edit_and_ai_improvement():
+    state = valid_minimal_job()
+    stage_id = "text_generation"
+    assert has_valid_human_approval(stage_id, state)
+
+    save_human_edits(state, stage_id, "edited text", "Reviewer", "Needs precision", "[]")
+    assert not has_valid_human_approval(stage_id, state)
+    assert any(approval.approval_status == "invalidated" for approval in state.approvals)
+
+    state = valid_minimal_job()
+    improve_with_ai(state, stage_id)
+    assert not has_valid_human_approval(stage_id, state)
+
+
+def test_upstream_change_marks_downstream_stale_and_locks():
+    state = stale_downstream_job()
+
+    for stage_id in STAGE_IDS[STAGE_IDS.index("slide_outline_order") + 1 :]:
+        assert state.stages[stage_id].is_stale
+        assert "Stage is stale." in get_stage_lock_reasons(stage_id, state)
+
+    assert not can_unlock_next_stage("text_generation", state)
+
+
+def test_objective_traceability_issues():
+    state = valid_minimal_job()
+    traces = compute_objective_traces(state)
+    assert all(trace.mapped_slide_ids for trace in traces)
+    assert not any(
+        issue.issue_type == IssueType.OBJECTIVE_UNCOVERED
+        for issue in state.issues.values()
+        if not issue.resolved
+    )
+
+    state = missing_objective_mapping_job()
+    traces = compute_objective_traces(state)
+    uncovered = [trace for trace in traces if trace.coverage_status == "uncovered"]
+    assert uncovered
+    assert any(
+        issue.issue_type == IssueType.OBJECTIVE_UNCOVERED
+        and issue.severity == IssueSeverity.BLOCKER
+        for issue in state.issues.values()
+    )
+
+    state = valid_minimal_job()
+    state.slides["slide_1"].objective_coverage_scores = {"obj_1": 40}
+    traces = compute_objective_traces(state)
+    assert any(trace.coverage_status == "weak" for trace in traces)
+    assert any(issue.issue_type == IssueType.OBJECTIVE_WEAKLY_COVERED for issue in state.issues.values())
+
+
+def test_claim_support_blocks_technical_review_and_preflight():
+    state = valid_minimal_job()
+    assert validate_claim_support(state) == []
+
+    state = unsupported_claim_job()
+    issues = validate_claim_support(state)
+    assert issues
+    assert issues[0].severity == IssueSeverity.BLOCKER
+
+    result = grade_stage("technical_review", state)
+    assert not result.passed_threshold
+
+    report = run_export_preflight(state)
+    assert not report.can_export
+    assert any("unsupported_claim" in issue_id for issue_id in report.blocking_issue_ids)
+
+
+def test_artifact_lifecycle_and_export_require_approved_current_artifacts():
+    state = build_empty_state(
+        deck_title="Lifecycle",
+        source_url="mock://source",
+        template_url="mock://template",
+    )
+    state, _message = generate_stage(state, "setup_inputs")
+    artifact = get_current_stage_artifact("setup_inputs", state)
+
+    assert artifact is not None
+    assert artifact.status == ArtifactStatus.CANDIDATE
+
+    state.stages["setup_inputs"].score = 90
+    approve_current_artifact(state, "setup_inputs")
+    assert get_current_stage_artifact("setup_inputs", state).status == ArtifactStatus.APPROVED
+
+    state = invalidated_approval_job()
+    report = run_export_preflight(state)
+    assert not report.can_export
+
+    state = stale_downstream_job()
+    report = run_export_preflight(state)
+    assert not report.can_export
+
+
+def test_preflight_fixture_matrix():
+    assert run_export_preflight(valid_minimal_job()).can_export
+
+    for fixture in [
+        missing_objective_mapping_job,
+        unsupported_claim_job,
+        missing_visual_asset_job,
+        invalid_layout_job,
+        stale_downstream_job,
+        invalidated_approval_job,
+    ]:
+        report = run_export_preflight(fixture())
+        assert not report.can_export
+
+    state = valid_minimal_job()
+    state.production_export_requested = True
+    assert not run_export_preflight(state).can_export
+
+    state = valid_minimal_job()
+    state.mutation_target_url = state.source_url
+    assert not run_export_preflight(state).can_export
+
+
+def test_text_density_limits_and_cognitive_load():
+    state = valid_minimal_job()
+    assert check_text_density("slide_1", state) == []
+
+    state = valid_minimal_job()
+    state.slides["slide_1"].visible_text = " ".join(["word"] * 80)
+    issues = check_text_density("slide_1", state)
+    assert any(issue.issue_type == IssueType.TEXT_DENSITY_EXCEEDED for issue in issues)
+
+    state = valid_minimal_job()
+    state.slides["slide_1"].bullet_points = ["a", "b", "c", "d", "e"]
+    issues = check_text_density("slide_1", state)
+    assert any(issue.issue_type == IssueType.TEXT_DENSITY_EXCEEDED for issue in issues)
+
+    state = valid_minimal_job()
+    state.slides["slide_1"].objective_ids = ["obj_1", "obj_2", "obj_3"]
+    issues = check_text_density("slide_1", state)
+    assert any(issue.issue_type == IssueType.COGNITIVE_LOAD_HIGH for issue in issues)
+
+
+def test_layout_validation_schema_and_slots():
+    state = valid_minimal_job()
+    assert validate_layout_spec(state.layout_specs["slide_1"], state) == []
+
+    state = valid_minimal_job()
+    unknown = LayoutSpec(slide_id="slide_1", layout_id="unknown", slot_assignments={})
+    issues = validate_layout_spec(unknown, state)
+    assert any(issue.issue_type == IssueType.LAYOUT_SCHEMA_INVALID for issue in issues)
+
+    state = valid_minimal_job()
+    missing = LayoutSpec(
+        slide_id="slide_1",
+        layout_id="title_bullets_visual",
+        slot_assignments={"title": "Only title"},
+    )
+    issues = validate_layout_spec(missing, state)
+    assert any(issue.issue_type == IssueType.LAYOUT_SLOT_VIOLATION for issue in issues)
+
+    state = valid_minimal_job()
+    raw = LayoutSpec(slide_id="slide_1", layout_id="title_body", slot_assignments={"x": 1})
+    issues = validate_layout_spec(raw, state)
+    assert any(issue.issue_type == IssueType.LAYOUT_SCHEMA_INVALID for issue in issues)
+
+
+def test_final_export_marks_approved_artifacts_exported_and_blocks_failures():
+    state = valid_minimal_job()
+    state, message = final_render_export(state)
+    assert "completed" in message
+    assert get_current_stage_artifact("setup_inputs", state).status == ArtifactStatus.EXPORTED
+
+    state = unsupported_claim_job()
+    state, message = final_render_export(state)
+    assert "failed" in message
+    assert any(event.event_type == "export_blocked_by_preflight" for event in state.audit_events)
+
+
+def test_artifact_diff_supports_text_and_json():
+    text_diff = compute_artifact_diff("alpha\n", "beta\n")
+    json_diff = compute_artifact_diff({"b": 2, "a": 1}, {"a": 1, "b": 3})
+
+    assert "-alpha" in text_diff
+    assert '+  "b": 3' in json_diff