from __future__ import annotations import json import re import zipfile from pathlib import Path from typing import Any from xml.etree import ElementTree from .constants import APPROVAL_REQUIRED_STAGE_IDS, STAGE_IDS, STAGE_LABELS, STAGE_SEQUENCE from .compat import model_to_dict, model_validate from .models import ( ArtifactStatus, IssueStatus, LayoutSpec, PedagogicalRole, PipelineState, PromptRun, ReviewRole, RevisionConstraints, RequestedChange, ReviewerNotes, Slide, SlideClaim, SpeakerNotes, StageState, VisualAsset, ) from .quality import ( PASSING_SCORE, 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, ) from .review import ( REVIEW_MODE_DEFAULTS, apply_proposed_change_set, assign_issue, build_review_queue, compare_artifact_versions_semantically, compute_reviewer_productivity_metrics, create_proposed_change_set, create_review_packet, critique_artifact_for_improvement, generate_suggested_fix_for_issue, generate_suggested_fixes, improve_selected_slides, list_prompt_runs_for_stage, reject_proposed_change_set, restore_artifact_version_as_candidate, role_review_mode_summary, update_issue_status, ) TEXT_MATERIAL_EXTENSIONS = {".txt", ".md", ".csv"} STRUCTURED_MATERIAL_EXTENSIONS = {".json"} DOCUMENT_MATERIAL_EXTENSIONS = {".pdf", ".docx", ".pptx"} IMAGE_DRAFT_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp"} START_MODES = {"instructions_file", "input_outline", "import_existing_draft", "start_from_zero"} 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 _normalize_file_paths(file_paths: Any) -> list[Path]: if not file_paths: return [] if isinstance(file_paths, str | Path): return [Path(file_paths)] normalized: list[Path] = [] for item in file_paths: if isinstance(item, str | Path): normalized.append(Path(item)) continue path = getattr(item, "path", None) or getattr(item, "name", None) if path: normalized.append(Path(path)) return normalized def _file_content_hash(path: Path) -> str: return stable_hash({"path": path.name, "bytes": path.read_bytes().hex()}) def _read_text_file(path: Path) -> str: return path.read_text(encoding="utf-8", errors="replace").strip() def _read_pdf_text(path: Path) -> str: try: from pypdf import PdfReader reader = PdfReader(str(path)) page_text: list[str] = [] for page in reader.pages: try: text = page.extract_text(extraction_mode="layout") or "" except TypeError: text = page.extract_text() or "" page_text.append(text) return "\n\n".join(page_text).strip() except ImportError: try: import pdfplumber except ImportError as exc: raise RuntimeError("pypdf is required to parse PDF files.") from exc with pdfplumber.open(str(path)) as pdf: return "\n\n".join(page.extract_text() or "" for page in pdf.pages).strip() def _read_docx_text(path: Path) -> str: with zipfile.ZipFile(path) as archive: document_xml = archive.read("word/document.xml") root = ElementTree.fromstring(document_xml) paragraph_tag = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}p" text_tag = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}t" paragraphs: list[str] = [] for paragraph in root.iter(paragraph_tag): text = "".join(node.text or "" for node in paragraph.iter(text_tag)).strip() if text: paragraphs.append(text) return "\n".join(paragraphs).strip() def _read_pptx_text(path: Path) -> str: try: from pptx import Presentation except ImportError as exc: raise RuntimeError("python-pptx is required to parse PPTX files.") from exc presentation = Presentation(str(path)) slide_text: list[str] = [] for index, pptx_slide in enumerate(presentation.slides, start=1): lines: list[str] = [] if pptx_slide.shapes.title and pptx_slide.shapes.title.has_text_frame: title = pptx_slide.shapes.title.text.strip() if title: lines.append(title) for shape in pptx_slide.shapes: if getattr(shape, "has_text_frame", False): text = shape.text.strip() if text and text not in lines: lines.append(text) notes_text = _slide_notes_text(pptx_slide) if notes_text: lines.append(notes_text) if lines: slide_text.append(f"Slide {index}\n" + "\n".join(lines)) return "\n\n".join(slide_text).strip() def _extract_document_text(path: Path) -> str: suffix = path.suffix.lower() if suffix == ".pdf": return _read_pdf_text(path) if suffix == ".docx": return _read_docx_text(path) if suffix == ".pptx": return _read_pptx_text(path) return "" def extract_uploaded_source_chunks(file_paths: Any) -> tuple[dict[str, str], list[dict[str, Any]], list[dict[str, Any]]]: chunks: dict[str, str] = {} parsed_materials: list[dict[str, Any]] = [] unsupported_materials: list[dict[str, Any]] = [] for index, path in enumerate(_normalize_file_paths(file_paths), start=1): suffix = path.suffix.lower() metadata = { "filename": path.name, "path": str(path), "extension": suffix, } if suffix in TEXT_MATERIAL_EXTENSIONS: text = _read_text_file(path) if text: chunk_id = f"upload_{index}" chunks[chunk_id] = text parsed_materials.append({**metadata, "chunk_id": chunk_id, "parsed": True}) continue if suffix in STRUCTURED_MATERIAL_EXTENSIONS: raw_text = _read_text_file(path) if not raw_text: continue parsed = json.loads(raw_text) chunk_id = f"upload_{index}" chunks[chunk_id] = json.dumps(parsed, indent=2, sort_keys=True) parsed_materials.append({**metadata, "chunk_id": chunk_id, "parsed": True}) continue if suffix in DOCUMENT_MATERIAL_EXTENSIONS: try: text = _extract_document_text(path) except Exception as exc: # noqa: BLE001 - user uploads should report parse failures, not crash setup. unsupported_materials.append( { **metadata, "parsed": False, "reason": f"Could not parse {suffix} material: {exc}", } ) continue if text: chunk_id = f"upload_{index}" chunks[chunk_id] = text parsed_materials.append({**metadata, "chunk_id": chunk_id, "parsed": True}) continue unsupported_materials.append( { **metadata, "parsed": False, "reason": f"No extractable text was found in {suffix} material.", } ) continue unsupported_materials.append( { **metadata, "parsed": False, "reason": ( "Only .txt, .md, .csv, .json, .pdf, .docx, and .pptx material uploads " "are parsed in this version." ), } ) return chunks, parsed_materials, unsupported_materials def _normalize_start_mode(start_mode: str | None) -> str: if start_mode in START_MODES: return start_mode return "instructions_file" def _clean_outline_line(raw_line: str) -> str: line = raw_line.strip() line = re.sub(r"^#{1,6}\s*", "", line) line = re.sub(r"^(?:[-*+]|\d+[.)]|[A-Za-z][.)]|[\u2022\u25cf\u25cb\u25a0])\s*", "", line) return re.sub(r"\s+", " ", line).strip() def _outline_line_level(raw_line: str) -> int: stripped = raw_line.lstrip() indent = len(raw_line) - len(stripped) if stripped.startswith("\u25cf"): return 0 if stripped.startswith(("\u25cb", "\u25a0")) or indent >= 2: return 1 return 0 def _title_from_text(text: str, fallback: str) -> str: title = re.sub(r"\s+", " ", text).strip(" .:-") if not title: return fallback if len(title) <= 72: return title truncated = title[:69].rsplit(" ", 1)[0].rstrip(" .:-") return f"{truncated or title[:69]}..." def _bullet_from_text(text: str) -> str: bullet = re.sub(r"\s+", " ", text).strip() if len(bullet) <= 120: return bullet truncated = bullet[:117].rsplit(" ", 1)[0].rstrip(" .:-") return f"{truncated or bullet[:117]}..." def _sections_from_outline_text(outline_text: str, *, max_sections: int = 12) -> list[dict[str, Any]]: sections: list[dict[str, Any]] = [] current: dict[str, Any] | None = None for raw_line in outline_text.splitlines(): clean = _clean_outline_line(raw_line) if not clean or clean.lower().startswith("slide "): continue level = _outline_line_level(raw_line) if current is None or level == 0: current = {"title": clean, "bullets": []} sections.append(current) if len(sections) >= max_sections: break continue current["bullets"].append(clean) return sections def _role_for_section(title: str, index: int, total: int) -> PedagogicalRole: lowered = title.lower() if index == 1 or any(keyword in lowered for keyword in ["purpose", "intro", "motivat"]): return PedagogicalRole.MOTIVATION if any(keyword in lowered for keyword in ["process", "steps", "procedure", "compute", "calculate"]): return PedagogicalRole.PROCEDURE if any(keyword in lowered for keyword in ["example", "worked"]): return PedagogicalRole.WORKED_EXAMPLE if any(keyword in lowered for keyword in ["takeaway", "wrap", "summary"]): return PedagogicalRole.SUMMARY if index == total: return PedagogicalRole.SUMMARY return PedagogicalRole.CONCEPT def _section_requires_visual(title: str, bullets: list[str]) -> bool: text = " ".join([title, *bullets]).lower() return any( keyword in text for keyword in [ "accuracy", "chart", "diagram", "dimensionality", "equation", "math", "matrix", "model", "process", "sigmoid", "visual", ] ) def _build_slides_from_sections(sections: list[dict[str, Any]], objective_ids: list[str] | None = None) -> list[Slide]: slides: list[Slide] = [] objective_ids = objective_ids or [] total = len(sections) for index, section in enumerate(sections, start=1): title = _title_from_text(str(section.get("title") or ""), f"Slide {index}") bullets = [_bullet_from_text(str(item)) for item in section.get("bullets", []) if str(item).strip()] bullets = [item for item in bullets if item][:5] slide_id = f"slide_{index}" mapped_objective_ids = [objective_ids[min(index - 1, len(objective_ids) - 1)]] if objective_ids else [] visible_text = "\n".join(bullets) if bullets else title slides.append( Slide( slide_id=slide_id, slide_number=index, title=title, visible_text=visible_text, bullet_points=bullets, objective_ids=mapped_objective_ids, pedagogical_role=_role_for_section(title, index, total), requires_visual=_section_requires_visual(title, bullets), speaker_notes=SpeakerNotes( slide_id=slide_id, notes_text=f"Use the provided source material to teach: {title}.", instructor_intent="Keep this slide source-grounded and concise.", estimated_teaching_time_seconds=180, ), distinct_concept_count=max(1, len(bullets) or 1), ) ) return slides def _derive_objectives_from_sections(sections: list[dict[str, Any]], *, max_objectives: int = 4) -> dict[str, str]: objectives: dict[str, str] = {} for index, section in enumerate(sections[:max_objectives], start=1): title = _title_from_text(str(section.get("title") or ""), f"topic {index}") objectives[f"obj_{index}"] = f"Explain {title[0].lower() + title[1:] if title else f'topic {index}'}." return objectives def _outline_text_from_state(state: PipelineState) -> str: if state.start_mode == "input_outline": return state.source_chunks.get("pasted_outline", "").strip() if state.start_mode in {"instructions_file", "start_from_zero"}: return "\n".join(state.source_chunks.values()).strip() return "" def _sequence_items(value: Any) -> list[Any]: if value is None: return [] if isinstance(value, dict): return list(value.values()) if isinstance(value, list): return value return [] def _import_json_draft(path: Path) -> dict[str, Any]: data = json.loads(_read_text_file(path)) if not isinstance(data, dict): raise ValueError("Draft JSON must be an object.") objectives = data.get("objectives") or {} source_chunks = data.get("source_chunks") or {} slides = [ model_validate(Slide, item) for item in _sequence_items(data.get("slides")) ] claims = [ model_validate(SlideClaim, item) for item in _sequence_items(data.get("claims")) ] visual_assets = [ model_validate(VisualAsset, item) for item in _sequence_items(data.get("visual_assets")) ] layout_specs = [ model_validate(LayoutSpec, item) for item in _sequence_items(data.get("layout_specs")) ] return { "objectives": objectives, "source_chunks": source_chunks, "slides": slides, "claims": claims, "visual_assets": visual_assets, "layout_specs": layout_specs, "metadata": {"format": "json", "filename": path.name}, } def _slide_notes_text(slide: Any) -> str | None: try: notes_slide = slide.notes_slide text_frame = getattr(notes_slide, "notes_text_frame", None) if text_frame and getattr(text_frame, "text", None): return text_frame.text.strip() except (AttributeError, KeyError, ValueError): return None return None def _import_pptx_draft(path: Path) -> dict[str, Any]: try: from pptx import Presentation except ImportError as exc: raise RuntimeError("python-pptx is required to import PPTX drafts.") from exc presentation = Presentation(str(path)) slides: list[Slide] = [] source_lines: list[str] = [] for index, pptx_slide in enumerate(presentation.slides, start=1): text_runs: list[str] = [] title = None if pptx_slide.shapes.title and pptx_slide.shapes.title.has_text_frame: title = pptx_slide.shapes.title.text.strip() or None for shape in pptx_slide.shapes: if not getattr(shape, "has_text_frame", False): continue text = shape.text.strip() if text: text_runs.append(text) deduped_text = [text for text in text_runs if text != title] bullet_points = [] for text in deduped_text: bullet_points.extend([line.strip() for line in text.splitlines() if line.strip()]) slide_id = f"slide_{index}" notes_text = _slide_notes_text(pptx_slide) source_lines.append("\n".join([item for item in [title, *deduped_text, notes_text] if item])) slides.append( Slide( slide_id=slide_id, slide_number=index, title=title or f"Imported slide {index}", visible_text="\n".join(deduped_text), bullet_points=bullet_points, pedagogical_role=PedagogicalRole.UNKNOWN, speaker_notes=( SpeakerNotes(slide_id=slide_id, notes_text=notes_text) if notes_text else None ), ) ) return { "objectives": {}, "source_chunks": {"imported_pptx_text": "\n\n".join(item for item in source_lines if item)}, "slides": slides, "claims": [], "visual_assets": [], "layout_specs": [], "metadata": {"format": "pptx", "filename": path.name}, } def _import_pdf_draft(path: Path) -> dict[str, Any]: text = _read_pdf_text(path) sections = _sections_from_outline_text(text) objectives = _derive_objectives_from_sections(sections) if sections else {} slides = _build_slides_from_sections(sections, list(objectives)) if sections else [] return { "objectives": objectives, "source_chunks": {"imported_pdf_text": text} if text else {}, "slides": slides, "claims": [], "visual_assets": [], "layout_specs": [], "metadata": {"format": "pdf", "filename": path.name}, } def _import_image_draft(path: Path) -> dict[str, Any]: slide_id = "slide_1" title = _title_from_text(path.stem.replace("_", " ").replace("-", " "), "Uploaded draft image") slide = Slide( slide_id=slide_id, slide_number=1, title=title, visible_text="Uploaded image draft. Add or revise slide text before final export.", bullet_points=["Review image content", "Add source-grounded narration", "Confirm final layout"], pedagogical_role=PedagogicalRole.UNKNOWN, requires_visual=True, speaker_notes=SpeakerNotes( slide_id=slide_id, notes_text="Use the uploaded image as the draft reference and refine text manually or with AI.", instructor_intent="Preserve the user's visual draft while making the slide teachable.", estimated_teaching_time_seconds=180, ), ) visual_asset = VisualAsset( asset_id="asset_1", slide_id=slide_id, asset_type="screenshot", path_or_url=str(path), purpose="instructional", alt_text=f"Uploaded draft image for {title}", source="uploaded", license_status="user_provided", approved_for_export=True, ) layout = LayoutSpec( slide_id=slide_id, layout_id="title_bullets_visual", approved_template_id="default_course_template", slot_assignments={"title": title, "bullets": slide.bullet_points, "visual": visual_asset.asset_id}, ) return { "objectives": {"obj_1": f"Explain {title[0].lower() + title[1:] if title else 'the uploaded draft'}."}, "source_chunks": {"uploaded_image_reference": f"Visual draft uploaded from {path.name}."}, "slides": [slide], "claims": [], "visual_assets": [visual_asset], "layout_specs": [layout], "metadata": {"format": "image", "filename": path.name}, } def import_existing_draft(file_path: Any) -> dict[str, Any]: paths = _normalize_file_paths(file_path) if not paths: return {} path = paths[0] suffix = path.suffix.lower() if suffix == ".json": return _import_json_draft(path) if suffix == ".pptx": return _import_pptx_draft(path) if suffix == ".pdf": return _import_pdf_draft(path) if suffix in IMAGE_DRAFT_EXTENSIONS: return _import_image_draft(path) raise ValueError("Existing draft upload must be a .json, .pptx, .pdf, .png, .jpg, .jpeg, or .webp file.") def _apply_imported_draft(state: PipelineState, draft: dict[str, Any], file_hash: str) -> None: if draft.get("objectives"): state.objectives = dict(draft["objectives"]) if draft.get("source_chunks"): state.source_chunks.update(dict(draft["source_chunks"])) for slide in draft.get("slides", []): state.slides[slide.slide_id] = slide for claim in draft.get("claims", []): state.claims[claim.claim_id] = claim for asset in draft.get("visual_assets", []): state.visual_assets[asset.asset_id] = asset for layout in draft.get("layout_specs", []): state.layout_specs[layout.slide_id] = layout imported_content = { "slides": [model_to_dict(slide) for slide in draft.get("slides", [])], "claims": [model_to_dict(claim) for claim in draft.get("claims", [])], "visual_assets": [model_to_dict(asset) for asset in draft.get("visual_assets", [])], "layout_specs": [model_to_dict(layout) for layout in draft.get("layout_specs", [])], "metadata": draft.get("metadata", {}), } created_stage_ids: list[str] = [] if draft.get("objectives") or draft.get("source_chunks"): create_artifact_version( state, "source_extraction_objective_mapping", { "objectives": state.objectives, "source_chunks": state.source_chunks, "import_metadata": draft.get("metadata", {}), }, created_by="human", status=ArtifactStatus.CANDIDATE, mark_downstream_stale=False, ) created_stage_ids.append("source_extraction_objective_mapping") if draft.get("slides"): create_artifact_version( state, "slide_outline_order", imported_content, created_by="human", status=ArtifactStatus.CANDIDATE, mark_downstream_stale=False, ) create_artifact_version( state, "title_generation", {"titles": {slide.slide_id: slide.title for slide in draft["slides"]}}, created_by="human", status=ArtifactStatus.CANDIDATE, mark_downstream_stale=False, ) create_artifact_version( state, "text_generation", imported_content, created_by="human", status=ArtifactStatus.CANDIDATE, mark_downstream_stale=False, ) created_stage_ids.extend(["slide_outline_order", "title_generation", "text_generation"]) if draft.get("visual_assets"): create_artifact_version( state, "image_visual_asset_generation", imported_content, created_by="human", status=ArtifactStatus.CANDIDATE, mark_downstream_stale=False, ) created_stage_ids.append("image_visual_asset_generation") if draft.get("layout_specs"): create_artifact_version( state, "aesthetic_ordering_visual_composition", imported_content, created_by="human", status=ArtifactStatus.CANDIDATE, mark_downstream_stale=False, ) created_stage_ids.append("aesthetic_ordering_visual_composition") state.draft_upload_metadata = { **draft.get("metadata", {}), "content_hash": file_hash, "created_stage_ids": created_stage_ids, "imported_at": now_iso(), } record_audit( state, "draft_imported", metadata={ "filename": state.draft_upload_metadata.get("filename"), "format": state.draft_upload_metadata.get("format"), "created_stage_ids": created_stage_ids, }, ) def apply_setup_material_inputs( state: PipelineState, *, material_files: Any = None, source_text: str | None = None, outline_text: str | None = None, start_mode: str = "instructions_file", draft_file: Any = None, ) -> PipelineState: state.start_mode = _normalize_start_mode(start_mode) # type: ignore[assignment] if state.start_mode in {"instructions_file", "start_from_zero"}: uploaded_chunks, parsed_materials, unsupported_materials = extract_uploaded_source_chunks(material_files) state.source_chunks = uploaded_chunks state.uploaded_materials = parsed_materials state.unsupported_materials = unsupported_materials if source_text and source_text.strip(): state.source_chunks = {"pasted_instructions": source_text.strip(), **state.source_chunks} if state.start_mode == "input_outline": state.uploaded_materials = [] state.unsupported_materials = [] source_chunks: dict[str, str] = {} if outline_text and outline_text.strip(): source_chunks["pasted_outline"] = outline_text.strip() if source_text and source_text.strip(): source_chunks["pasted_context"] = source_text.strip() state.source_chunks = source_chunks if state.start_mode == "import_existing_draft": draft_paths = _normalize_file_paths(draft_file) if draft_paths: file_hash = _file_content_hash(draft_paths[0]) if state.draft_upload_metadata.get("content_hash") != file_hash: draft = import_existing_draft(draft_paths[0]) _apply_imported_draft(state, draft, file_hash) return state 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 = None, outline_text: str | None = None, material_files: Any = None, start_mode: str = "instructions_file", draft_file: Any = 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 apply_setup_material_inputs( state, material_files=material_files, source_text=source_text, outline_text=outline_text, start_mode=start_mode, draft_file=draft_file, ) 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." } outline_text = _outline_text_from_state(state) if not state.objectives and outline_text: sections = _sections_from_outline_text(outline_text) if sections: state.objectives = _derive_objectives_from_sections(sections) 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 outline_text = _outline_text_from_state(state) if outline_text: sections = _sections_from_outline_text(outline_text) if sections: slides = _build_slides_from_sections(sections, list(state.objectives)) state.slides = {slide.slide_id: slide for slide in 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 _material_snapshot(materials: list[dict[str, Any]]) -> list[dict[str, Any]]: return [ { key: material.get(key) for key in ("filename", "extension", "chunk_id", "parsed", "reason") if material.get(key) not in (None, "") } for material in materials ] def _setup_snapshot_content(state: PipelineState) -> dict[str, Any]: draft_summary = { key: state.draft_upload_metadata.get(key) for key in ("filename", "format", "content_hash", "created_stage_ids", "imported_at") if state.draft_upload_metadata.get(key) not in (None, "") } return { "deck_title": state.deck_title, "start_mode": state.start_mode, "source_url": state.source_url, "template_url": state.template_url, "dry_run": state.dry_run, "production_export_requested": state.production_export_requested, "output_folder_id": state.output_folder_id, "mutation_target_url": state.mutation_target_url, "input_summary": { "source_chunk_count": len(state.source_chunks), "uploaded_material_count": len(state.uploaded_materials), "unsupported_material_count": len(state.unsupported_materials), "objective_count": len(state.objectives), "draft_imported": bool(state.draft_upload_metadata), }, "uploaded_materials": _material_snapshot(state.uploaded_materials), "unsupported_materials": _material_snapshot(state.unsupported_materials), "draft_upload": draft_summary, } 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 = _setup_snapshot_content(state) 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) generate_suggested_fixes(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, ) parent_id = artifact.parent_artifact_version_ids[0] if artifact.parent_artifact_version_ids else None if parent_id: compare_artifact_versions_semantically(parent_id, artifact.artifact_version_id, state) generate_suggested_fixes(state) return state, f"Saved human edits as candidate {artifact.artifact_version_id}." def improve_with_ai( state: PipelineState, stage_id: str, constraints: RevisionConstraints | None = None, ) -> 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 critique-before-improve for {STAGE_LABELS[stage_id]}", input_artifact_version_ids=input_ids, ) critique = critique_artifact_for_improvement(stage_id, state, constraints) change_set = create_proposed_change_set( stage_id, state, critique=critique, constraints=constraints, ) prompt_run.settings["critique_id"] = critique.critique_id prompt_run.settings["change_set_id"] = change_set.change_set_id return ( state, "Created critique " f"{critique.critique_id} and proposed change set {change_set.change_set_id}; " "no artifact was mutated.", ) def approve_and_continue( state: PipelineState, stage_id: str, reviewer_name: str = "human_reviewer", reviewer_role: ReviewRole | str | None = None, ) -> 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." role = ReviewRole(reviewer_role) if isinstance(reviewer_role, str) and reviewer_role else reviewer_role approval = approve_current_artifact(state, stage_id, reviewer_name, role) 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:] ] def review_role_choices() -> list[tuple[str, str]]: return [(config.label, role.value) for role, config in REVIEW_MODE_DEFAULTS.items()] def review_mode_summary_text(state: PipelineState, role: str | None) -> str: return role_review_mode_summary(role, state) def review_queue_table( state: PipelineState, role: str | None = None, stage_id: str | None = None, severity: str | None = None, status: str | None = None, issue_type: str | None = None, assigned_to: str | None = None, ) -> list[list[Any]]: queue = build_review_queue( state, role=role, stage_id=stage_id, severity=severity, status=status, issue_type=issue_type, assigned_to=assigned_to or None, ) return [ [ item.severity.value, item.priority, item.status.value, item.stage_id or "", item.slide_id or "", item.assigned_role.value if item.assigned_role else "", item.issue_type.value, item.message, item.suggested_fix_summary or "", item.assigned_to or "", item.created_at, item.issue_id, ] for item in queue ] def suggested_fixes_table(state: PipelineState) -> list[list[Any]]: if state.issues and not state.suggested_fixes: generate_suggested_fixes(state) return [ [ fix.fix_id, fix.fix_type, fix.stage_id or "", ", ".join(fix.slide_ids), ", ".join(fix.issue_ids), fix.risk_level, fix.description, ] for fix in state.suggested_fixes.values() ] def proposed_change_sets_table(state: PipelineState) -> list[list[Any]]: return [ [ change_set.change_set_id, change_set.stage_id, change_set.status, change_set.artifact_version_id or "", len(change_set.changes), change_set.summary or "", change_set.created_at, ] for change_set in state.proposed_change_sets.values() ] def proposed_changes_table(state: PipelineState, change_set_id: str | None = None) -> list[list[Any]]: change_sets = list(state.proposed_change_sets.values()) if change_set_id: change_sets = [ change_set for change_set in change_sets if change_set.change_set_id == change_set_id ] elif change_sets: change_sets = [change_sets[-1]] rows: list[list[Any]] = [] for change_set in change_sets: for change in change_set.changes: rows.append( [ change_set.change_set_id, change.change_id, change.target_type, change.target_id, change.field_path or "", json.dumps(change.before, sort_keys=True, default=str), json.dumps(change.after, sort_keys=True, default=str), change.rationale or "", ", ".join(change.issue_ids), change.risk_level, ] ) return rows def latest_critique_text(state: PipelineState) -> str: if not state.critiques: return "" critique = next(reversed(state.critiques.values())) return json.dumps(model_to_dict(critique), indent=2, sort_keys=True) def latest_semantic_diff_text(state: PipelineState) -> str: if not state.version_comparisons: return "" comparison = next(reversed(state.version_comparisons.values())) return json.dumps(model_to_dict(comparison), indent=2, sort_keys=True) def prompt_runs_table(state: PipelineState, stage_id: str | None = "all") -> list[list[Any]]: runs = list_prompt_runs_for_stage(stage_id or "all", state) return [ [ run.prompt_run_id, run.created_at, run.stage_id, run.provider, run.model_name or "", run.prompt_template_id or "", ", ".join(run.input_artifact_version_ids), run.output_artifact_version_id or "", json.dumps(run.settings, sort_keys=True), ] for run in runs ] def prompt_run_detail_text(state: PipelineState, prompt_run_id: str | None = None) -> str: run = state.prompt_runs.get(prompt_run_id or "") if run is None and state.prompt_runs: run = next(reversed(state.prompt_runs.values())) return json.dumps(model_to_dict(run), indent=2, sort_keys=True) if run else "" def reviewer_metrics_table(state: PipelineState) -> list[list[Any]]: metrics = compute_reviewer_productivity_metrics(state) return [ ["total_issues", metrics.total_issues], ["open_issues", metrics.open_issues], ["resolved_issues", metrics.resolved_issues], ["waived_issues", metrics.waived_issues], ["blocker_count", metrics.blocker_count], ["major_count", metrics.major_count], ["issues_by_stage", json.dumps(metrics.issues_by_stage, sort_keys=True)], ["issues_by_role", json.dumps(metrics.issues_by_role, sort_keys=True)], ["average_score_by_stage", json.dumps(metrics.average_score_by_stage, sort_keys=True)], ["candidate_versions_created", metrics.candidate_versions_created], ["approvals_invalidated", metrics.approvals_invalidated], ["stale_events", metrics.stale_events], ] def triage_issue_action( state: PipelineState, issue_id: str, action: str, reviewer_name: str, note: str, assigned_role: str | None = None, assigned_to: str | None = None, ) -> tuple[PipelineState, str]: if not issue_id: return state, "Select an issue ID first." try: if action == "assign": assign_issue(issue_id, assigned_role, assigned_to, state) return state, f"Assigned {issue_id}." status_map = { "acknowledge": IssueStatus.ACKNOWLEDGED, "in_progress": IssueStatus.IN_PROGRESS, "resolve": IssueStatus.RESOLVED, "waive": IssueStatus.WAIVED, "wont_fix": IssueStatus.WONT_FIX, "reopen": IssueStatus.OPEN, } update_issue_status( issue_id, status_map[action], state, reviewer_name=reviewer_name or "human_reviewer", note=note, ) return state, f"Updated {issue_id} to {status_map[action].value}." except (KeyError, ValueError) as exc: return state, f"Issue triage failed: {exc}" def generate_suggested_fix_action(state: PipelineState, issue_id: str) -> tuple[PipelineState, str]: if not issue_id: return state, "Select an issue ID first." try: fix = generate_suggested_fix_for_issue(issue_id, state) return state, f"Generated suggested fix {fix.fix_id}." except ValueError as exc: return state, f"Suggested fix failed: {exc}" def apply_change_set_action( state: PipelineState, change_set_id: str, selected_change_ids_csv: str | None, reviewer_name: str, ) -> tuple[PipelineState, str]: if not change_set_id: if not state.proposed_change_sets: return state, "Create a proposed change set first." change_set_id = next(reversed(state.proposed_change_sets)) selected = [ item.strip() for item in (selected_change_ids_csv or "").split(",") if item.strip() ] try: before_count = len(state.artifacts) apply_proposed_change_set( change_set_id, state, selected_change_ids=selected or None, reviewer_name=reviewer_name or "human_reviewer", ) after_count = len(state.artifacts) message = ( f"Applied {change_set_id} and created a candidate artifact." if after_count > before_count else f"{change_set_id} was not applied." ) return state, message except ValueError as exc: return state, f"Apply failed: {exc}" def reject_change_set_action( state: PipelineState, change_set_id: str, reviewer_name: str, reason: str, ) -> tuple[PipelineState, str]: if not change_set_id: if not state.proposed_change_sets: return state, "Create a proposed change set first." change_set_id = next(reversed(state.proposed_change_sets)) try: reject_proposed_change_set( change_set_id, state, reviewer_name=reviewer_name or "human_reviewer", reason=reason or None, ) return state, f"Rejected {change_set_id}." except ValueError as exc: return state, f"Reject failed: {exc}" def targeted_slide_improvement_action( state: PipelineState, stage_id: str, slide_ids_csv: str, action: str, constraints: RevisionConstraints, ) -> tuple[PipelineState, str]: slide_ids = [item.strip() for item in slide_ids_csv.split(",") if item.strip()] if not slide_ids: return state, "Enter one or more slide IDs." change_set = improve_selected_slides( stage_id, slide_ids, state, constraints=constraints, action=action, # type: ignore[arg-type] ) return state, f"Created targeted change set {change_set.change_set_id}." def compare_latest_versions_action(state: PipelineState, stage_id: str) -> tuple[PipelineState, str]: versions = state.stage_artifact_versions.get(stage_id, []) if len(versions) < 2: return state, "At least two artifact versions are required for semantic comparison." comparison = compare_artifact_versions_semantically(versions[-2], versions[-1], state) return state, comparison.summary or f"Created comparison {comparison.comparison_id}." def create_review_packet_action( state: PipelineState, role: str | None, stage_ids_csv: str, slide_ids_csv: str, packet_format: str, ) -> tuple[PipelineState, str]: stage_ids = [item.strip() for item in stage_ids_csv.split(",") if item.strip()] slide_ids = [item.strip() for item in slide_ids_csv.split(",") if item.strip()] packet = create_review_packet( state, role=role, stage_ids=stage_ids, slide_ids=slide_ids, format="json" if packet_format == "json" else "markdown", ) return state, f"Created {packet.format} review packet: {packet.path}" def restore_candidate_action( state: PipelineState, artifact_version_id: str, reviewer_name: str, ) -> tuple[PipelineState, str]: if not artifact_version_id: return state, "Enter an artifact version ID to restore." try: restore_artifact_version_as_candidate( artifact_version_id, state, reviewer_name=reviewer_name or "human_reviewer", ) return state, f"Restored {artifact_version_id} as a new candidate." except ValueError as exc: return state, f"Restore failed: {exc}" def run_generate_and_grade_for_current_stage( stage_id: str, state: PipelineState, ) -> PipelineState: record_audit(state, "fast_path_generate_grade_started", stage_id=stage_id) generate_stage(state, stage_id) grade_stage_action(state, stage_id) stop_reason = None if has_unresolved_blockers(stage_id, state): stop_reason = "Stage has blockers." elif state.stages[stage_id].score is not None and state.stages[stage_id].score < PASSING_SCORE: stop_reason = "Stage score is below threshold." elif stage_id in APPROVAL_REQUIRED_STAGE_IDS: stop_reason = "Human approval is required." record_audit( state, "fast_path_generate_grade_stopped", stage_id=stage_id, reason=stop_reason or "Current stage generated and graded.", ) return state def run_draft_until_next_required_approval( start_stage_id: str, state: PipelineState, max_stages: int = 3, ) -> PipelineState: if start_stage_id not in STAGE_IDS: raise ValueError(f"Unknown stage: {start_stage_id}") start_index = STAGE_IDS.index(start_stage_id) for stage_id in STAGE_IDS[start_index : start_index + max_stages]: run_generate_and_grade_for_current_stage(stage_id, state) stage = state.stages[stage_id] if has_unresolved_blockers(stage_id, state): break if stage.score is None or stage.score < PASSING_SCORE: break if stage_id in APPROVAL_REQUIRED_STAGE_IDS: break return state