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, IssueStatus, IssueType, LayoutSpec, ObjectiveTrace, PipelineState, QualityIssue, ReviewRole, 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) priority = { IssueSeverity.BLOCKER: "critical", IssueSeverity.MAJOR: "high", IssueSeverity.MINOR: "medium", IssueSeverity.INFO: "low", }[severity] 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, priority=priority, last_seen_artifact_version_id=artifact_version_id, 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.status = IssueStatus.OPEN existing.priority = priority existing.resolved_at = None existing.resolution_note = None existing.last_seen_artifact_version_id = artifact_version_id 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.status = IssueStatus.RESOLVED issue.resolved_at = now_iso() record_audit(state, "issue_resolved", issue_id=issue_id, reason="Condition cleared") def _issue_blocks_quality(issue: QualityIssue) -> bool: if issue.resolved or issue.status == IssueStatus.RESOLVED: return False if issue.status == IssueStatus.WAIVED: return issue.severity == IssueSeverity.BLOCKER if issue.status == IssueStatus.WONT_FIX: return issue.severity == IssueSeverity.BLOCKER return True def unresolved_issues(state: PipelineState) -> list[QualityIssue]: return [issue for issue in state.issues.values() if _issue_blocks_quality(issue)] 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", reviewer_role: ReviewRole | None = None, ) -> 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", reviewer_role=reviewer_role, 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, "reviewer_role": approval.reviewer_role.value if approval.reviewer_role else None, }, ) 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": source_material_present = bool(state.source_url or state.source_chunks) required = [state.deck_title, state.template_url, source_material_present] if dimension_id == "required_inputs_present": return ( 100 if all(required) else 30, "Deck title, template URL, and either uploaded material or a material URL are checked.", ) if dimension_id == "url_validity": urls = [url for url in [state.source_url, state.template_url] if url] valid = all(url.startswith(("http://", "https://", "mock://")) for url in urls) return (100 if valid else 40, "Provided material/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.template_url or not (state.source_url or state.source_chunks): upsert_issue( state, IssueType.MISSING_REQUIRED_INPUT, IssueSeverity.BLOCKER, "Deck title, template URL, and course material upload or 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