"""NEXUS 6-step agent loop — Manus-style: analyze → search → plan → generate → assemble → save.""" from __future__ import annotations import asyncio import json import logging import re from datetime import datetime, timezone from typing import Any, Awaitable, Callable from sqlalchemy import select from agent.deck_repair import repair_for_validator from agent.layouts_registry import ( CANONICAL_LAYOUTS, FALLBACK_LAYOUT, normalize_layout, ) from agent.memory import AgentMemory from agent.planner import Planner from agent.prompt_intent import extract_slide_count from agent.prompts import nexus_system_prompt, slides_user_message from agent.prompts import CRITIC_SYSTEM_PROMPT, critic_user_message from database.connection import SessionLocal from database.models import SlideDeck, Task from services.claude_service import ClaudeService from services.lifecycle_service import JobCancelled from services.search_service import SearchService logger = logging.getLogger("nexus.agent.loop") ProgressCallback = Callable[[str, float, str], Awaitable[None]] # Valid slide layouts come from the canonical registry. Do NOT reintroduce a # hardcoded literal here — it will diverge from the frontend renderer. _VALID_LAYOUTS = frozenset(CANONICAL_LAYOUTS) class NexusAgentLoop: """Orchestrates the full slide generation pipeline for a single task.""" def __init__( self, claude: ClaudeService | None = None, search: SearchService | None = None, planner: Planner | None = None, ) -> None: self.claude = claude or ClaudeService() self.search = search or SearchService() self.planner = planner or Planner(claude=self.claude) async def _ai_call( self, *, role: str, system: str, user: str, max_tokens: int, temperature: float = 0.7, ) -> tuple[str, int, float]: """Phase 6W: route via ``complete_for_role`` when available, else fall back to ``complete`` so ClaudeService stand-ins/fakes that don't implement role routing keep working unchanged.""" if hasattr(self.claude, "complete_for_role"): return await self.claude.complete_for_role( role=role, system=system, user=user, max_tokens=max_tokens, temperature=temperature, ) return await self.claude.complete( system=system, user=user, max_tokens=max_tokens, temperature=temperature ) # ── Phase 6W: research compression (summarize role) ────────────────── # Threshold above which raw research is summarized before reaching the # planner / writer. Below this we keep full text to avoid information # loss on small decks. ~10k chars ≈ ~2.5k tokens. _RESEARCH_SUMMARIZE_THRESHOLD = 10_000 async def _summarize_long_research( self, topic: str, research: str ) -> tuple[str, int, float]: """Compress harvested research via the ``summarize`` role when long. Returns ``(text, tokens, cost)``. If research is short or summarize fails, returns the original text unchanged with zero tokens/cost. Source metadata is preserved separately by the caller; this only compresses the narrative blob. """ if not research or len(research) < self._RESEARCH_SUMMARIZE_THRESHOLD: return research, 0, 0.0 try: text, tokens, cost = await self._ai_call( role="summarize", system=( "You compress research notes into dense, fact-preserving " "summaries. Keep all numbers, dates, names, and citations. " "Drop filler, marketing language, and duplicates. Plain text only." ), user=( f"Topic: {topic}\n\n" f"Compress the following research to <= 1500 words while " f"preserving every concrete figure, date, and source mention.\n\n" f"---\n{research}\n---" ), max_tokens=2048, temperature=0.2, ) if text and len(text) < len(research): logger.info( "loop.research_summarized", extra={"orig_chars": len(research), "new_chars": len(text)}, ) return text, tokens, cost except Exception as exc: logger.warning("loop.summarize_failed", extra={"err": str(exc)}) return research, 0, 0.0 async def _json_fix_retry( self, raw_text: str, *, expected: str = "array" ) -> tuple[str, int, float]: """Phase 6W: ask the ``json_fix`` role to repair malformed model output. Used when ``_parse_slides_array``/``_parse_single_slide`` fails on the primary writer output. Returns ``(repaired_text, tokens, cost)``; the caller is responsible for re-parsing. On any error returns the input text unchanged with zero tokens/cost so callers can fall through to the existing deterministic fallback path. """ if not raw_text or not raw_text.strip(): return raw_text, 0, 0.0 target = "JSON array of slide objects" if expected == "array" else "single JSON slide object" try: text, tokens, cost = await self._ai_call( role="json_fix", system=( "You repair malformed JSON. Output ONLY valid JSON matching " "the requested shape. No prose, no code fences, no commentary." ), user=( f"The following text was supposed to be a {target} but failed to parse. " f"Repair it and return ONLY the valid JSON.\n\n---\n{raw_text}\n---" ), max_tokens=4096, temperature=0.0, ) return text or raw_text, tokens, cost except Exception as exc: logger.warning("loop.json_fix_failed", extra={"err": str(exc)}) return raw_text, 0, 0.0 async def run( self, task_id: str, topic: str, slide_count: int, theme: str, search_web: bool, on_progress: ProgressCallback, min_sources: int = 0, ) -> dict[str, Any]: memory = AgentMemory(task_id) total_tokens = 0 total_cost = 0.0 model_used = self.claude.active_model research_sources: list[dict[str, Any]] = [] # Resolved at strategy step; used to build a mode-specific system prompt. _deck_type = "overview" _mood = "neutral" # Phase 6AD: per-slide narrative beats. Computed once after # strategy is built; threaded into ``attach_slide_intent`` and # ``recommend_layouts`` so all downstream consumers see the # same dramatic shape. _beats: list[str] = [] # Phase 6U: honour explicit slide-count requests embedded in the # prompt ("Produce a 12-slide ..."). This overrides the API # default of 8 — the user told us how many slides they want, so # the planner must respect that instead of the field default. hinted = extract_slide_count(topic) if hinted is not None and hinted != slide_count: logger.info( "loop.slide_count_override topic_hint=%d field=%d", hinted, slide_count ) slide_count = hinted try: # 1 — ANALYZE await on_progress("Analyzing your topic...", 8.0, "analyze") await self._mark_running(task_id, "analyze", 8.0) # Phase 6AM-Grounding: topic expansion. # Short/vague prompts ("ai", "create ppt on ai") cause the # search step to return disambiguation noise (Wikipedia # stubs about unrelated subjects). We expand into 3-5 # specific queries BEFORE search so the harvester has real # angles to ground on. The expander degrades to [topic] # under env kill switch, LLM failure, or already-specific # input \u2014 the pipeline keeps working unchanged in those # cases. Tokens are charged to ``total_tokens`` so the # frontend cost panel stays accurate. search_queries: list[str] = [topic] try: from agent.topic_expander import expand_topic, canonicalize_topic expanded, exp_tokens, exp_cost = await expand_topic( topic, ai_call=self._ai_call, max_queries=5 ) if expanded: search_queries = expanded total_tokens += exp_tokens total_cost += exp_cost # Replace the topic the rest of the pipeline sees with # the canonicalized form so meta-verbs ("create ppt on") # don't leak into the planner / writer / image prompts. canon = canonicalize_topic(topic) if canon: topic = canon if len(search_queries) > 1: await on_progress( f"Expanded to {len(search_queries)} search angles.", 12.0, "analyze", event="design_decision", decision="topic_expanded", value={ "queries": search_queries, "tokens": exp_tokens, }, ) except Exception as exc: logger.warning("loop.topic_expander_failed", extra={"err": str(exc)}) # 2 — SEARCH research = "" # Phase 6AN-JsonImport: when the task was created by the # POST /api/import/json endpoint, the user-supplied data has # already been written to the memory dir as # ``seed_research.json`` (or ``seed_research.txt``). Use it # as the research blob and skip the web-search step entirely # \u2014 the generator should ground on the user's data, not # Wikipedia. ``search_web`` is set to False on those tasks, # so this branch is only reachable when there is in fact a # seed file to read; the legacy path is untouched. seed_research_text = "" try: from pathlib import Path as _Path _seed_root = memory.root # type: ignore[attr-defined] _seed_json = _seed_root / "seed_research.json" _seed_txt = _seed_root / "seed_research.txt" if _seed_json.exists(): import json as _json seed_research_text = _json.dumps( _json.loads(_seed_json.read_text(encoding="utf-8")), ensure_ascii=False, indent=2, ) elif _seed_txt.exists(): seed_research_text = _seed_txt.read_text(encoding="utf-8") except Exception as exc: # pragma: no cover - defensive logger.warning("loop.seed_research_read_failed", extra={"err": str(exc)}) seed_research_text = "" if seed_research_text.strip(): research = seed_research_text[:20_000] await on_progress( "Grounding on user-supplied data.", 18.0, "search", event="design_decision", decision="seed_research", value={"chars": len(research)}, rationale=( "Skipped web search; the deck is generated from the " "JSON payload provided with the import." ), ) elif search_web: await on_progress("Researching topic on the web...", 18.0, "search") await self._mark_running(task_id, "search", 18.0) try: # Phase 6AM-Grounding: when expansion produced # multiple queries, run each through the search # backend in sequence and merge results. Dedupe by # URL. The harvest path's built-in riders still # apply for the FIRST query when min_sources is set. if len(search_queries) > 1: seen_urls: set[str] = set() merged_sources: list[dict[str, Any]] = [] summary_parts: list[str] = [] for qi, q in enumerate(search_queries): try: if ( qi == 0 and min_sources and hasattr(self.search, "harvest") ): sub_research, sub_sources = ( await self.search.harvest( q, target_min=min_sources, max_total=8, ) ) else: sub_research, sub_sources = ( await self.search.search(q, max_results=4) ) except Exception as exc: logger.warning( "loop.search_query_failed", extra={"q": q[:80], "err": str(exc)}, ) continue if isinstance(sub_research, str) and sub_research.strip(): summary_parts.append(sub_research.strip()) for s in sub_sources or []: if not isinstance(s, dict): continue url = str(s.get("url") or "").strip() key = url or str(s) if key in seen_urls: continue seen_urls.add(key) merged_sources.append(s) if len(merged_sources) >= 12: break if len(merged_sources) >= 12: break research = "\n\n".join(summary_parts[:3]).strip() search_sources = merged_sources elif min_sources and hasattr(self.search, "harvest"): research, search_sources = await self.search.harvest( topic, target_min=min_sources, max_total=12 ) else: research, search_sources = await self.search.search( topic, max_results=6 ) if isinstance(search_sources, list): research_sources = list(search_sources) except Exception as exc: logger.warning("loop.search_failed", extra={"err": str(exc)}) research = "" memory.write_research(research) # Emit per-source events so the frontend can show live source cards. for src in research_sources[:8]: src_title = (src.get("title") or src.get("url") or "Source")[:100] src_url = src.get("url") or "" await on_progress(src_title, 19.0, "search", event="source_found", url=src_url) # Phase 6W: compress very long research before it reaches planner # and writer. Source metadata stays intact in ``research_sources``; # only the narrative blob is summarized. if research: research, _stok, _scost = await self._summarize_long_research( topic, research ) total_tokens += _stok total_cost += _scost # Emit research snippet for live intel panel. if research: await on_progress(research[:800], 23.5, "search", event="research_note") # 2b — STRATEGY (Phase 6V) # Build a structured deck-strategy artifact from the topic + # harvested research + art-direction. The planner and the # batch-generator now read from this artifact instead of just # the raw research blob, so different deck types produce # different layout orders, story arcs, and visual directions. await on_progress("Building deck strategy...", 24.0, "strategy") try: from agent.art_direction import infer_art_direction from agent.deck_strategy import build_deck_strategy art = infer_art_direction(topic, theme) strategy = build_deck_strategy( topic=topic, slide_count=slide_count, art_direction=art, research=research, research_sources=research_sources, ) memory.write_artifact("strategy.json", strategy.to_dict()) _deck_type = strategy.deck_type _mood = art.mood # Narrate design decisions so the user can see the AI's # reasoning. These events are surface-only (no schema impact). try: await on_progress( f"Deck type: {strategy.deck_type}", 24.5, "strategy", event="design_decision", decision="deck_type", value=strategy.deck_type, rationale=(strategy.thesis or "")[:240], ) await on_progress( f"Mood: {art.mood}", 24.7, "strategy", event="design_decision", decision="mood", value=art.mood, rationale=art.rationale, category=art.category, ) if strategy.story_arc: await on_progress( f"Story arc: {' → '.join(strategy.story_arc[:5])}", 24.9, "strategy", event="design_decision", decision="story_arc", value=list(strategy.story_arc), rationale=f"{strategy.deck_type} decks pace as: {' → '.join(strategy.story_arc)}", ) if strategy.layout_recipe: await on_progress( f"Layout recipe: {', '.join(strategy.layout_recipe[:6])}", 25.1, "strategy", event="design_decision", decision="layout_recipe", value=list(strategy.layout_recipe), rationale=f"Tone: {strategy.tone}; visual: {strategy.visual_direction}", ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.design_decision_emit_failed", extra={"err": str(exc)}) # Phase 6AD: derive narrative beats from the strategy # arc + research signals. Pure-Python; never raises. try: from agent.narrative_beats import derive_beats _beats = derive_beats( slide_count=slide_count, story_arc=list(strategy.story_arc) if strategy else None, research=research, ) if _beats: memory.write_artifact("narrative_beats.json", {"beats": _beats}) await on_progress( f"Narrative beats: {' → '.join(_beats[:7])}", 25.3, "strategy", event="design_decision", decision="narrative_beats", value=list(_beats), rationale=( "Documentary beat sequence derived from the deck strategy " "and research signals; drives per-slide intent and section " "divider placement." ), ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.narrative_beats_failed", extra={"err": str(exc)}) _beats = [] except Exception as exc: # pragma: no cover - defensive logger.warning("loop.strategy_failed", extra={"err": str(exc)}) strategy = None await on_progress( "Research strategy ready.", 26.0, "strategy", ) # 2c — NARRATIVE SYNTHESIS (Phase 6AN-Story) # Write the deck's prose-first ground truth before we plan # slide titles. One LLM call produces a ~500-word editor's # brief mapped to the strategy's story arc; both the planner # and the writer then consume it so every slide condenses a # paragraph of one coherent story instead of being invented # in isolation. Degrades to None on kill switch or failure; # the legacy outline-only path keeps working. narrative = None try: from agent.narrative_synthesizer import synthesize_narrative key_facts: list[str] = [] audience = "" tone = "" thesis_hint = "" story_arc_list: list[str] = [] if strategy is not None: key_facts = list(getattr(strategy, "key_facts", []) or []) audience = getattr(strategy, "audience", "") or "" tone = getattr(strategy, "tone", "") or "" thesis_hint = getattr(strategy, "thesis", "") or "" story_arc_list = list(getattr(strategy, "story_arc", []) or []) draft, n_tokens, n_cost = await synthesize_narrative( topic=topic, research=research, deck_type=_deck_type, audience=audience, tone=tone, thesis_hint=thesis_hint, story_arc=story_arc_list, key_facts=key_facts, slide_count=slide_count, ai_call=self._ai_call, ) total_tokens += n_tokens total_cost += n_cost if not draft.is_empty: narrative = draft memory.write_artifact( "narrative.json", { "thesis": draft.thesis, "sections": [ {"beat": s.beat, "heading": s.heading, "body": s.body} for s in draft.sections ], }, ) try: await on_progress( (draft.thesis or "Narrative drafted.")[:240], 27.0, "strategy", event="design_decision", decision="narrative", value={ "thesis": draft.thesis, "sections": [ {"beat": s.beat, "heading": s.heading} for s in draft.sections ], }, rationale=( "Story-first draft written before the slide " "skeleton. Each slide condenses one section." ), ) except Exception as exc: # pragma: no cover - defensive logger.warning( "loop.narrative_emit_failed", extra={"err": str(exc)} ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.narrative_failed", extra={"err": str(exc)}) narrative = None # 3 — PLAN await on_progress("Planning slide structure...", 28.0, "plan") await self._mark_running(task_id, "plan", 28.0) outline, p_tokens, p_cost = await self.planner.plan( topic, slide_count, research, strategy=strategy, narrative=narrative ) total_tokens += p_tokens total_cost += p_cost memory.write_outline(outline) memory.write_todo(outline) # Emit outline so the frontend can show the planned slide structure. if outline: _outline_titles = [ {"i": i + 1, "layout": p.get("layout", ""), "title": p.get("title", "")} for i, p in enumerate(outline) ] await on_progress( "Slide structure planned", 29.0, "plan", event="outline_ready", outline=_outline_titles, ) # 4 — GENERATE slides: list[dict[str, Any]] = [] try: slides, g_tokens, g_cost = await self._generate_all_at_once( topic, slide_count, research, outline, on_progress, memory, strategy=strategy, deck_type=_deck_type, mood=_mood, narrative=narrative, ) total_tokens += g_tokens total_cost += g_cost except Exception as exc: logger.warning( "loop.batch_generate_failed_falling_back", extra={"err": str(exc)} ) slides = [] if not slides: slides, g_tokens, g_cost = await self._generate_per_slide( topic, research, outline, on_progress, memory, deck_type=_deck_type, mood=_mood, ) total_tokens += g_tokens total_cost += g_cost # 5 — ASSEMBLE await on_progress("Finalizing presentation...", 90.0, "assemble") await self._mark_running(task_id, "assemble", 90.0) slides = self._normalize_slides(slides, slide_count, topic) # Phase 4: attach research sources to source-bearing layouts. # Advisory only — never invents sources, never fails generation. try: from agent.source_grounding import attach_research_sources_to_deck slides = attach_research_sources_to_deck(slides, research_sources) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.attach_sources_failed", extra={"err": str(exc)}) # Phase 6AB: attach per-slide intent metadata. Pure function; # never invents content; renderer ignores it if unrecognised. try: from agent.slide_intent import attach_slide_intent slides = attach_slide_intent( slides, story_arc=list(strategy.story_arc) if strategy else None, strategy_tone=strategy.tone if strategy else None, art_mood=_mood, beats=_beats if _beats else None, ) # Surface the per-slide intent rhythm in the AI-reasoning # panel so the user can see the narrative pacing decision. try: rhythm = [ { "i": i + 1, "role": (s.get("intent") or {}).get("narrative_role"), "density": (s.get("intent") or {}).get("density"), } for i, s in enumerate(slides) if isinstance(s, dict) ] if rhythm: await on_progress( f"Pacing: {' · '.join(r['role'] or '?' for r in rhythm[:8])}", 91.0, "assemble", event="design_decision", decision="intent_rhythm", value=rhythm, rationale=f"Per-slide narrative role and density across {len(rhythm)} slides.", ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.intent_rhythm_emit_failed", extra={"err": str(exc)}) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.attach_intent_failed", extra={"err": str(exc)}) # Phase 6AA: deterministic layout upgrader. Uses the intent # metadata attached above plus the slide's own content shape # to promote stats→bigstat / bullets→section_divider when # the upgrade is unambiguous. Never replaces planner choices # otherwise. Re-normalises so the new layouts pick up their # canonical field shape, then re-attaches intent so the # rhythm event reflects the upgrades. try: from agent.layout_recommender import ( recommend_layouts, enforce_hero, insert_section_dividers, ) slides, _layout_upgrades = recommend_layouts(slides) # Phase 6AI-B1 — guarantee at least one hero moment. _slides_after_hero, _hero_upgrades = enforce_hero(slides) if _hero_upgrades: slides = _slides_after_hero _layout_upgrades = list(_layout_upgrades) + list(_hero_upgrades) # Phase 6AI-B6 — chapter break for long decks. _slides_after_div, _div_upgrades = insert_section_dividers(slides) if _div_upgrades: slides = _slides_after_div _layout_upgrades = list(_layout_upgrades) + list(_div_upgrades) if _layout_upgrades: slides = self._normalize_slides(slides, slide_count, topic) try: from agent.slide_intent import attach_slide_intent slides = attach_slide_intent( slides, story_arc=list(strategy.story_arc) if strategy else None, strategy_tone=strategy.tone if strategy else None, art_mood=_mood, ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.attach_intent_failed", extra={"err": str(exc)}) # One design_decision per upgrade so the AI-reasoning # panel shows the audit trail. for u in _layout_upgrades: try: await on_progress( f"Slide {u['slide_index']}: {u['from']} → {u['to']}", 91.5, "assemble", event="design_decision", decision="layout_upgrade", value=u, rationale=u.get("reason") or "", ) except Exception as exc: # pragma: no cover - defensive logger.warning( "loop.layout_upgrade_emit_failed", extra={"err": str(exc)}, ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.layout_recommender_failed", extra={"err": str(exc)}) # 5b — CRITIC (rewrite weak slides for Manus-level specificity) try: slides, c_tokens, c_cost = await self._critique_and_rewrite( topic, research, slides, on_progress ) total_tokens += c_tokens total_cost += c_cost slides = self._normalize_slides(slides, slide_count, topic) # Re-attach: critic may have rewritten chart/stats slides. try: from agent.source_grounding import attach_research_sources_to_deck slides = attach_research_sources_to_deck(slides, research_sources) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.attach_sources_failed", extra={"err": str(exc)}) # Re-attach intent — critic may have changed layout / density. try: from agent.slide_intent import attach_slide_intent slides = attach_slide_intent( slides, story_arc=list(strategy.story_arc) if strategy else None, strategy_tone=strategy.tone if strategy else None, art_mood=_mood, ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.attach_intent_failed", extra={"err": str(exc)}) except Exception as exc: logger.warning("loop.critic_failed", extra={"err": str(exc)}) # 5c — IMAGES (hero visual per slide via Pollinations) try: slides, i_tokens, i_cost = await self._add_hero_images( topic, slides, on_progress ) total_tokens += i_tokens total_cost += i_cost except Exception as exc: logger.warning("loop.images_failed", extra={"err": str(exc)}) # 6 — SAVE await on_progress("Saving your slides...", 96.0, "save") # Phase 6U: final pass that fills layout-local defaults # (title.subtitle/eyebrow, closing.subtitle/cta, chart.subtitle, # quote.attribution, chart_data.unit/source) so the deck # satisfies ``validate_deck`` before persisting. The leading # cause of the 6T ``deck_quality_ok=1/11`` regression was the # safety-net inside ``_normalize_slides`` that pins # ``out[0].layout='title'`` and ``out[-1].layout='closing'`` # without seeding the required fields for those layouts. The # repair pass closes that gap without inventing content. try: slides = repair_for_validator(slides) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.deck_repair_failed", extra={"err": str(exc)}) # Phase 6AJ: deterministic editorial polish. # Strengthens titles, drops hedging openers / generic phrases # in bullets, and adds short narrative transitions where # consecutive slides have an ``intent.narrative_role``. # Pure / offline / additive — never raises, never removes # numbers, years, or proper nouns. Runs after repair so the # rewritten copy still satisfies the validator contract. try: from agent.editorial_pass import apply_editorial_pass slides, editorial_summary = apply_editorial_pass(slides) try: await on_progress( f"Edited {editorial_summary.get('headline_rewrites', 0)} titles, " f"{editorial_summary.get('bullet_rewrites', 0)} bullets; " f"added {editorial_summary.get('transitions_added', 0)} transitions.", 96.5, "save", event="editorial_pass", editorial_summary=editorial_summary, ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.editorial_event_failed", extra={"err": str(exc)}) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.editorial_pass_failed", extra={"err": str(exc)}) # Phase 6AL-Voice: deterministic editorial-voice pass. # Rewrites category-label headlines ("Problem Statement", # "Traction Metrics", "Investment Ask") into authored # sentences derived from existing slide content, drops # filler subtitles ("A Fundable Story", "Join Our Mission"), # scrubs debug-grade transitions ("Setting the stage:", # "What the data shows:"), and rewrites button-shaped # closings into declarative titles. Pure / offline / # deterministic / additive — no LLM, no network, never # raises, exporter-safe. Kill-switch: # NEXUS_DISABLE_VOICE_PASS=true. try: from agent.voice_pass import apply_voice_pass slides, voice_summary = apply_voice_pass(slides) if ( voice_summary.get("headline_rewrites", 0) + voice_summary.get("subtitle_kills", 0) + voice_summary.get("transition_scrubs", 0) + voice_summary.get("closing_rewrites", 0) > 0 ): try: await on_progress( f"Voice pass: " f"{voice_summary.get('headline_rewrites', 0)} headline(s), " f"{voice_summary.get('subtitle_kills', 0)} filler subtitle(s), " f"{voice_summary.get('transition_scrubs', 0)} transition(s), " f"{voice_summary.get('closing_rewrites', 0)} closing(s) rewritten.", 96.6, "save", event="design_decision", decision="voice_pass", value=voice_summary, rationale="Replaced category-label headlines and filler copy with authored deck voice.", ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.voice_event_failed", extra={"err": str(exc)}) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.voice_pass_failed", extra={"err": str(exc)}) # Phase 6AK: mark cinematic hero moments. Writes is_hero=True # on the first bigstat / section_divider / attributed quote # so the renderer can promote them into full-bleed cinematic # variants. Pure / offline / additive — renderer-only flag, # ignored by exporters. try: from agent.cinematic_marker import mark_hero_moments slides, hero_summary = mark_hero_moments(slides) if hero_summary.get("total", 0) > 0: try: await on_progress( f"Marked {hero_summary.get('total', 0)} cinematic moment(s).", 96.8, "save", event="design_decision", decision="cinematic_hero", value=hero_summary, rationale="Promoted hero slides for full-bleed cinematic rendering.", ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.cinematic_event_failed", extra={"err": str(exc)}) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.cinematic_marker_failed", extra={"err": str(exc)}) # Phase 6AF: deterministic claim-level citation attach. # Runs after repair so citations target the final, validator- # accepted slides. Pure / offline / additive — never raises, # never mutates ``sources``, and only writes ``slide["citations"]``. try: from agent.citation_attach import attach_citations_to_deck slides, citation_summary = attach_citations_to_deck(slides) try: await on_progress( f"Cited {citation_summary.get('supported', 0)}/" f"{citation_summary.get('total_claims', 0)} claims " f"across {citation_summary.get('slides_with_citations', 0)} slides.", 97.0, "save", event="citation_checked", citation_summary=citation_summary, ) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.citation_event_failed", extra={"err": str(exc)}) except Exception as exc: # pragma: no cover - defensive logger.warning("loop.citation_attach_failed", extra={"err": str(exc)}) await self._save_deck(task_id, slides, theme, total_tokens, total_cost, model_used) await on_progress("Done! Your slides are ready.", 100.0, "done", status="done") return {"slides": slides, "tokens": total_tokens, "cost_usd": total_cost} except JobCancelled as exc: # Phase 6Q: graceful cancel. Worker top-level finalizes the # row via ``lifecycle_service.mark_cancelled`` and the SSE # stream closes once the ``cancelled`` event is published. logger.info( "loop.cancelled", extra={"task_id": task_id, "stage": str(exc) or "unknown"}, ) await on_progress( "Generation cancelled by user.", 100.0, "cancelled", status="cancelled", ) raise except Exception as exc: logger.exception("loop.failed", extra={"task_id": task_id}) await self._mark_failed(task_id, str(exc)) await on_progress(f"Generation failed: {exc}", 100.0, "failed", status="failed", error=str(exc)) raise # ── generation strategies ───────────────────────────────────────────── async def _generate_all_at_once( self, topic: str, slide_count: int, research: str, outline: list[dict[str, Any]], on_progress: ProgressCallback, memory: AgentMemory, *, strategy: Any | None = None, deck_type: str = "overview", mood: str = "neutral", narrative: Any | None = None, ) -> tuple[list[dict[str, Any]], int, float]: await on_progress(f"Writing slides 1 of {slide_count}...", 35.0, "generate") outline_text = "\n".join( f"{i + 1}. ({p['layout']}) {p['title']} — {p['intent']}" for i, p in enumerate(outline) ) user_msg = slides_user_message( topic, slide_count, research, outline_text, strategy=strategy, narrative=narrative, ) text, tokens, cost = await self._ai_call( role="writer", system=nexus_system_prompt(deck_type, mood), user=user_msg, max_tokens=8096, ) slides = self._parse_slides_array(text) if not slides: # Phase 6W: try a single json_fix repair before giving up. repaired, rtok, rcost = await self._json_fix_retry(text, expected="array") tokens += rtok cost += rcost if repaired and repaired != text: slides = self._parse_slides_array(repaired) if not slides: return [], tokens, cost for i, slide in enumerate(slides[:slide_count]): memory.write_slide(i, slide) memory.mark_todo_done(i) pct = 35.0 + (i + 1) * (50.0 / slide_count) await on_progress( f"Writing slide {i + 1} of {slide_count}...", min(pct, 85.0), "generate", event="slide", slide_index=i, slide_total=slide_count, slide=slide, ) return slides[:slide_count], tokens, cost async def _generate_per_slide( self, topic: str, research: str, outline: list[dict[str, Any]], on_progress: ProgressCallback, memory: AgentMemory, *, deck_type: str = "overview", mood: str = "neutral", ) -> tuple[list[dict[str, Any]], int, float]: from agent.prompts import single_slide_user_message slides: list[dict[str, Any]] = [] total_tokens = 0 total_cost = 0.0 n = len(outline) # OpenManus-style stuck detection: track raw responses to avoid the # LLM repeating identical bad output across attempts. seen_responses: set[str] = set() MAX_ATTEMPTS = 2 for i, plan in enumerate(outline): pct = 35.0 + (i + 1) * (50.0 / max(n, 1)) await on_progress( f"Writing slide {i + 1} of {n}...", min(pct, 85.0), "generate" ) base_user = single_slide_user_message(topic, research, plan, i, n) slide = None for attempt in range(MAX_ATTEMPTS): # Perturb prompt + temperature on retry to break repeat loops. user_msg = base_user temp = 0.7 if attempt > 0: user_msg = ( base_user + "\n\nThe previous attempt produced unusable output. " "Try a different angle, use specific numbers, and return ONLY valid JSON." ) temp = 0.95 try: text, tokens, cost = await self._ai_call( role="writer", system=nexus_system_prompt(deck_type, mood), user=user_msg, max_tokens=2048, temperature=temp, ) total_tokens += tokens total_cost += cost fingerprint = (text or "").strip()[:200] if fingerprint and fingerprint in seen_responses: logger.info( "loop.stuck_detected_retrying", extra={"i": i, "attempt": attempt} ) continue if fingerprint: seen_responses.add(fingerprint) slide = self._parse_single_slide(text) if slide is not None: break # Phase 6W: one json_fix retry before next attempt/fallback. repaired, rtok, rcost = await self._json_fix_retry( text, expected="object" ) total_tokens += rtok total_cost += rcost if repaired and repaired != text: slide = self._parse_single_slide(repaired) if slide is not None: break except Exception as exc: logger.warning( "loop.single_slide_failed", extra={"i": i, "attempt": attempt, "err": str(exc)}, ) if slide is None: slide = self._fallback_slide(plan, i, n, topic) if slide is None: slide = self._fallback_slide(plan, i, n, topic) slides.append(slide) memory.write_slide(i, slide) memory.mark_todo_done(i) await on_progress( f"Wrote slide {i + 1} of {n}.", min(35.0 + (i + 1) * (50.0 / max(n, 1)), 85.0), "generate", event="slide", slide_index=i, slide_total=n, slide=slide, ) return slides, total_tokens, total_cost # ── critic pass ─────────────────────────────────────────────────────── _BLAND_PHRASES = ( "enhanced productivity", "improved efficiency", "improved decision", "data-driven insight", "personalized customer", "increased accuracy", "leverage synergies", "streamline operations", "key insight", "drive value", "cutting-edge", "state-of-the-art", "next-generation", "innovative solution", "seamless integration", "unlock potential", "transform industries", "revolutionize", ) @classmethod def _is_weak(cls, slide: dict[str, Any]) -> bool: """Heuristic: bland phrases or no concrete data → needs critic rewrite.""" layout = slide.get("layout") if layout in ("title", "closing"): return False # Don't waste tokens rewriting titles/closings. # Collect all visible text from the slide. chunks: list[str] = [str(slide.get("title") or "")] if layout == "bullets": chunks.extend(str(b) for b in (slide.get("bullets") or [])) elif layout == "two-col": for c in slide.get("columns") or []: if isinstance(c, dict): chunks.append(str(c.get("heading") or "")) chunks.append(str(c.get("body") or "")) elif layout == "stats": for s in slide.get("stats") or []: if isinstance(s, dict): chunks.append(str(s.get("value") or "")) chunks.append(str(s.get("label") or "")) elif layout == "quote": chunks.append(str(slide.get("quote") or "")) chunks.append(str(slide.get("attribution") or "")) text = " ".join(chunks).lower() if not text.strip(): return True # Bland phrase detection. for phrase in cls._BLAND_PHRASES: if phrase in text: return True # Stats slide must have real numeric values. if layout == "stats": for s in slide.get("stats") or []: val = str(s.get("value") if isinstance(s, dict) else "") if not re.search(r"\d", val): return True # Bullets/two-col: weak if NO bullet/body has any digit, year, or proper noun. if layout in ("bullets", "two-col"): joined = " ".join(chunks[1:]) # exclude title has_signal = bool( re.search(r"\d", joined) or re.search(r"\b(19|20)\d{2}\b", joined) # year or re.search(r"\b[A-Z][a-z]+(?:\s[A-Z][a-z]+)+", joined) # proper noun pair ) if not has_signal: return True return False async def _critique_and_rewrite( self, topic: str, research: str, slides: list[dict[str, Any]], on_progress: ProgressCallback, ) -> tuple[list[dict[str, Any]], int, float]: """For each weak slide, ask the LLM to rewrite it with specifics.""" weak_indices = [i for i, s in enumerate(slides) if self._is_weak(s)] if not weak_indices: logger.info("critic.all_strong", extra={"n": len(slides)}) return slides, 0, 0.0 await on_progress( f"Refining {len(weak_indices)} slide(s)...", 92.0, "critique" ) logger.info( "critic.rewriting", extra={"weak": weak_indices, "total": len(slides)} ) total_tokens = 0 total_cost = 0.0 out = list(slides) async def _rewrite_one(idx: int) -> None: nonlocal total_tokens, total_cost original = out[idx] try: text, tokens, cost = await self._ai_call( role="critic", system=CRITIC_SYSTEM_PROMPT, user=critic_user_message(topic, research, original), max_tokens=1024, ) total_tokens += tokens total_cost += cost rewritten = self._parse_single_slide(text) if rewritten and rewritten.get("layout") == original.get("layout"): rewritten["id"] = original.get("id") out[idx] = rewritten # Push the refined slide to the live preview. try: await on_progress( f"Refined slide {idx + 1}.", 93.0, "critique", event="slide", slide_index=idx, slide_total=len(out), slide=rewritten, ) except Exception: pass except Exception as exc: logger.warning( "critic.rewrite_failed", extra={"i": idx, "err": str(exc)} ) # Rewrite up to 4 slides in parallel to keep total time reasonable. sem = asyncio.Semaphore(4) async def _bound(idx: int) -> None: async with sem: await _rewrite_one(idx) await asyncio.gather(*(_bound(i) for i in weak_indices)) return out, total_tokens, total_cost # ── hero images ──────────────────────────────────────────────────────── # Phase 6AL-Visuals: image placement inversion. # Image-bearing layouts now include the hero moments (quote, bigstat, # section_divider). Stats / chart / table / comparison / timeline stay # image-free because they're data-dense. The renderer is responsible # for full-bleed vs right-panel composition; this method only attaches. _IMAGE_LAYOUTS = { "title", "bullets", "two-col", "closing", "quote", "bigstat", "section_divider", } # Phase 6AL-Visuals: banned vocabulary that turns Pollinations into a # stock-photo machine. We rewrite anything matching one of these words # out of LLM-generated prompts before sending them downstream. _IMAGE_BANNED_WORDS = ( "calming", "serene", "peaceful", "tranquil", "growing", "vibrant", "modern", "professional", "corporate", "beautiful", "stunning", "amazing", "perfect", "wonderful", "abstract", "minimal", "minimalist", "clean", "simple", "background", "wallpaper", "stock", "generic", "businessman", "businesswoman", "team meeting", "handshake", "lightbulb", "rocket", "graph going up", "thumbs up", ) # Phase 6AL-Visuals: deterministic cinematic suffix appended to every # image prompt. Forces editorial / documentary photography feel and # blocks Pollinations' default "AI slop" aesthetic. Kept additive so # weak LLM output still produces atmospheric imagery. _IMAGE_CINEMATIC_SUFFIX = ( "shot on 35mm film, shallow depth of field, golden hour lighting, " "editorial photography, atmospheric, cinematic composition, " "muted color grading, no text, no logos, no watermarks" ) @classmethod def _scrub_image_prompt(cls, raw: str) -> str: """Strip banned vocabulary; collapse whitespace. Never returns empty.""" cleaned = (raw or "").strip() if not cleaned: return "" lowered = cleaned.lower() for word in cls._IMAGE_BANNED_WORDS: if word in lowered: # Word-boundary replacement so "abstract" doesn't nuke # "extraction"; case-insensitive. cleaned = re.sub( rf"\b{re.escape(word)}\b", "", cleaned, flags=re.IGNORECASE ) cleaned = re.sub(r"\s+", " ", cleaned).strip(" ,.;:-") return cleaned @classmethod def _direct_image_prompt(cls, subject: str) -> str: """Compose subject + cinematic suffix into the final Pollinations brief.""" subj = cls._scrub_image_prompt(subject) if not subj: subj = "documentary scene" return f"{subj}, {cls._IMAGE_CINEMATIC_SUFFIX}" async def _add_hero_images( self, topic: str, slides: list[dict[str, Any]], on_progress: ProgressCallback, ) -> tuple[list[dict[str, Any]], int, float]: """Attach a Pollinations image URL to slides that benefit from a visual.""" from services.image_service import pollinations_url target_indices = [ i for i, s in enumerate(slides) if s.get("layout") in self._IMAGE_LAYOUTS ] if not target_indices: return slides, 0, 0.0 await on_progress("Generating slide imagery...", 94.0, "images") # ONE LLM call to generate cinematic visual subjects for every # target slide. The model is asked for SUBJECT + LIGHTING + LENS # + MOOD — never style adjectives. We append the editorial suffix # deterministically so the final prompt is always cinematic even # if the LLM output is weak. prompts: list[str] = [] tokens = 0 cost = 0.0 try: indexed = "\n".join( f"{i + 1}. layout={slides[i].get('layout')} title={slides[i].get('title','')}" for i in target_indices ) user = ( f"Topic: {topic}\n\n" f"Slides needing a cinematic image:\n{indexed}\n\n" f"You are an art director briefing a documentary photographer.\n" f"For EACH slide, write ONE concrete visual brief (max 22 words) " f"specifying a SUBJECT, a LIGHTING condition, and a CAMERA framing.\n\n" f"Examples of GOOD briefs:\n" f"- 'rain-slicked harbor at dawn, fishermen pulling nets, long lens, side light'\n" f"- 'half-empty warehouse with single forklift, fluorescent overhead, wide angle, deep shadow'\n" f"- 'analyst's desk at night, blue monitor glow on face, 50mm portrait, shallow focus'\n\n" f"DO NOT use words like: calming, serene, vibrant, modern, professional, " f"abstract, beautiful, stunning, background. No stock-photo cliches " f"(handshakes, lightbulbs, arrows pointing up, team meetings).\n" f"DO use specific subjects, real places, named lighting, real lens choices.\n\n" f"Return ONLY a JSON array of strings in slide order. No prose." ) text, tokens, cost = await self._ai_call( role="vision", system=( "You are an art director writing cinematic image briefs. " "You return only a JSON array of strings, each describing " "a concrete subject, lighting, and lens." ), user=user, max_tokens=700, temperature=0.5, ) cleaned = self._strip_fences(text) match = re.search(r"\[[\s\S]*\]", cleaned) if match: cleaned = match.group(0) data = json.loads(cleaned) if isinstance(data, list): prompts = [str(p) for p in data] except Exception as exc: logger.warning("images.prompt_gen_failed", extra={"err": str(exc)}) prompts = [] # Fallback: if the LLM failed or under-delivered, synthesize a # subject from the slide title. The cinematic suffix is appended # by _direct_image_prompt regardless of source. if len(prompts) < len(target_indices): for i in target_indices[len(prompts):]: t = slides[i].get("title") or topic prompts.append(f"documentary scene illustrating {t}, low key lighting, 35mm") for slot, idx in enumerate(target_indices): raw = prompts[slot] if slot < len(prompts) else topic final_prompt = self._direct_image_prompt(raw) slides[idx]["image_url"] = pollinations_url(final_prompt, seed=idx + 1) slides[idx]["image_prompt"] = final_prompt logger.info( "images.attached", extra={"count": len(target_indices), "total": len(slides)} ) return slides, tokens, cost @staticmethod def _strip_fences(text: str) -> str: cleaned = (text or "").strip() cleaned = re.sub(r"^```(?:json)?\s*", "", cleaned) cleaned = re.sub(r"\s*```$", "", cleaned) return cleaned.strip() @classmethod def _parse_slides_array(cls, text: str) -> list[dict[str, Any]]: cleaned = cls._strip_fences(text) match = re.search(r"\[\s*[\s\S]*\]", cleaned) if match: cleaned = match.group(0) try: data = json.loads(cleaned) except json.JSONDecodeError: return [] if not isinstance(data, list): return [] return [s for s in data if isinstance(s, dict)] @classmethod def _parse_single_slide(cls, text: str) -> dict[str, Any] | None: cleaned = cls._strip_fences(text) match = re.search(r"\{[\s\S]*\}", cleaned) if match: cleaned = match.group(0) try: data = json.loads(cleaned) except json.JSONDecodeError: return None return data if isinstance(data, dict) else None @classmethod def _normalize_slides( cls, slides: list[dict[str, Any]], slide_count: int, topic: str ) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] for i, raw in enumerate(slides): # Resolve via the canonical registry so aliases (when added) # and unknown values both go through one path. layout = normalize_layout(raw.get("layout")) if layout not in _VALID_LAYOUTS: layout = FALLBACK_LAYOUT base = { "id": f"slide-{i:03d}", "layout": layout, "title": str(raw.get("title") or "").strip(), } if layout == "title": base.update( { "subtitle": str(raw.get("subtitle") or "").strip(), "eyebrow": str(raw.get("eyebrow") or "Presentation").strip(), } ) elif layout == "bullets": bullets = raw.get("bullets") or [] if not isinstance(bullets, list): bullets = [] base["bullets"] = [str(b).strip() for b in bullets if str(b).strip()][:4] elif layout == "two-col": cols = raw.get("columns") or [] if not isinstance(cols, list): cols = [] base["columns"] = [ { "heading": str(c.get("heading") or "").strip(), "body": str(c.get("body") or "").strip(), } for c in cols[:2] if isinstance(c, dict) ] elif layout == "quote": base["quote"] = str(raw.get("quote") or raw.get("title") or "").strip() base["attribution"] = str(raw.get("attribution") or "").strip() elif layout == "stats": stats = raw.get("stats") or [] if not isinstance(stats, list): stats = [] base["stats"] = [ { "value": str(s.get("value") or "").strip(), "label": str(s.get("label") or "").strip(), } for s in stats[:3] if isinstance(s, dict) ] elif layout == "chart": cd = raw.get("chart_data") or {} if not isinstance(cd, dict): cd = {} labels = cd.get("labels") or [] values = cd.get("values") or [] if not isinstance(labels, list): labels = [] if not isinstance(values, list): values = [] # Coerce values to numbers; pair length to labels. numeric: list[float] = [] for v in values: try: numeric.append(float(str(v).replace(",", "").replace("$", ""))) except (TypeError, ValueError): numeric.append(0.0) pairs = min(len(labels), len(numeric)) base["chart_type"] = ( str(raw.get("chart_type") or "bar").strip().lower() or "bar" ) if base["chart_type"] not in {"bar", "line", "doughnut"}: base["chart_type"] = "bar" base["chart_data"] = { "labels": [str(x).strip() for x in labels[:pairs]], "values": numeric[:pairs], "unit": str(cd.get("unit") or "").strip(), "source": str(cd.get("source") or "").strip(), } base["subtitle"] = str(raw.get("subtitle") or "").strip() elif layout == "closing": base["subtitle"] = str(raw.get("subtitle") or "").strip() base["cta"] = str(raw.get("cta") or "Thank you").strip() elif layout == "bigstat": # Phase 6AA — preserve value / label / subtitle. If the # source slide carries a stats[] array (because the # recommender upgraded it), fall back to its first stat. stats_first = None stats = raw.get("stats") if isinstance(stats, list) and stats and isinstance(stats[0], dict): stats_first = stats[0] base["value"] = str( raw.get("value") or (stats_first.get("value") if stats_first else "") or "" ).strip() base["label"] = str( raw.get("label") or (stats_first.get("label") if stats_first else "") or "" ).strip() base["subtitle"] = str(raw.get("subtitle") or "").strip() elif layout == "section_divider": # Phase 6AA — typography-only block. Eyebrow/subtitle # default to empty so the validator accepts the slide. base["eyebrow"] = str(raw.get("eyebrow") or "").strip() base["subtitle"] = str(raw.get("subtitle") or "").strip() elif layout == "timeline": # Phase 6AC — chronology of dated events. Up to 6 events # are kept; each event is sanitized into {date, label}. events = raw.get("events") or [] if not isinstance(events, list): events = [] base["events"] = [ { "date": str(e.get("date") or "").strip(), "label": str(e.get("label") or "").strip(), } for e in events[:6] if isinstance(e, dict) ] base["subtitle"] = str(raw.get("subtitle") or "").strip() elif layout == "comparison": # Phase 6AC — side-by-side comparison. Both sides are # sanitized; missing sides default to empty heading/body # so the deck-repair pass + validator surface the gap. left = raw.get("left") or {} right = raw.get("right") or {} if not isinstance(left, dict): left = {} if not isinstance(right, dict): right = {} base["left"] = { "heading": str(left.get("heading") or "").strip(), "body": str(left.get("body") or "").strip(), } base["right"] = { "heading": str(right.get("heading") or "").strip(), "body": str(right.get("body") or "").strip(), } base["subtitle"] = str(raw.get("subtitle") or "").strip() out.append(base) # pad / trim to slide_count while len(out) < slide_count: out.append( cls._fallback_slide( {"layout": "bullets", "title": f"Section {len(out) + 1}", "intent": ""}, len(out), slide_count, topic, ) ) out = out[:slide_count] if out: out[0]["layout"] = "title" out[-1]["layout"] = "closing" # Safety net: ensure at least ONE chart slide. If the model produced # none, convert a stats slide (numeric values present) into a chart. has_chart = any(s.get("layout") == "chart" for s in out) if not has_chart: for s in out[1:-1]: if s.get("layout") != "stats": continue stats = s.get("stats") or [] labels: list[str] = [] values: list[float] = [] for item in stats: raw_v = str(item.get("value", "")) cleaned = raw_v.replace("$", "").replace(",", "").replace("%", "").strip() # take leading numeric portion num = "" for ch in cleaned: if ch.isdigit() or ch in {".", "-"}: num += ch else: break try: v = float(num) except ValueError: continue labels.append(str(item.get("label", "")).strip()[:24] or f"Item {len(values) + 1}") values.append(v) if len(values) >= 2: s["layout"] = "chart" s["chart_type"] = "bar" s["chart_data"] = { "labels": labels, "values": values, "unit": "", "source": s.get("source", ""), } # P1-2: chart slides require a slide-level ``subtitle`` # per the tightened schema. The safety-net previously # produced chart slides without it, generating a # self-inflicted validation warning on every deck. Carry # forward whatever the source stats slide had, default "". s["subtitle"] = s.get("subtitle", "") s.pop("stats", None) break # Phase 1C: non-repairing deck-quality telemetry. Build a # structured ``DeckQualityReport`` from the normalized deck and # log per-slide failures + a deck-level summary. We do NOT # mutate ``out`` and we do NOT reject slides — this is # observability only until a repair pipeline lands later. try: from agent.deck_quality import build_deck_quality_report # local import to avoid app/db cycles report = build_deck_quality_report(out) for record in report.errors: logger.warning( "loop.slide_validation_failed slide=%d layout=%s path=%s code=%s message=%s", record["slide_index"], record["layout"], record["path"], record["code"], record["message"], ) logger.info( "loop.deck_quality_report ok=%s slide_count=%d valid=%d invalid=%d repairs_needed=%d", report.ok, report.slide_count, report.valid_count, report.invalid_count, len(report.repair_actions), ) except Exception as exc: # pragma: no cover — telemetry must never break generation logger.warning("loop.slide_validation_telemetry_failed err=%s", exc) return out @staticmethod def _fallback_slide(plan: dict, index: int, total: int, topic: str) -> dict[str, Any]: layout = plan.get("layout", "bullets") title = plan.get("title") or f"Slide {index + 1}" if layout == "title": return {"id": f"slide-{index:03d}", "layout": "title", "title": topic, "subtitle": title, "eyebrow": "Presentation"} if layout == "closing": return {"id": f"slide-{index:03d}", "layout": "closing", "title": title, "subtitle": "", "cta": "Thank you"} return { "id": f"slide-{index:03d}", "layout": "bullets", "title": title, "bullets": [ "Key insight one.", "Key insight two.", "Key insight three.", ], } # ── DB writes ───────────────────────────────────────────────────────── async def _mark_running(self, task_id: str, step: str, pct: float) -> None: async with SessionLocal() as session: res = await session.execute(select(Task).where(Task.id == task_id)) task = res.scalar_one_or_none() if task is None: return # Phase 6Q: cancel checkpoint. The lifecycle endpoint flips # ``Task.status`` to ``cancelling``; the loop honours it at # every stage transition and exits via :class:`JobCancelled`. if task.status == "cancelling": raise JobCancelled(step) task.status = "running" task.current_step = step task.progress_pct = pct session.add(task) await session.commit() async def _mark_failed(self, task_id: str, err: str) -> None: async with SessionLocal() as session: res = await session.execute(select(Task).where(Task.id == task_id)) task = res.scalar_one_or_none() if task is None: return task.status = "failed" task.current_step = "failed" task.progress_pct = 100.0 task.error_msg = err[:2000] task.completed_at = datetime.now(timezone.utc) session.add(task) await session.commit() async def _save_deck( self, task_id: str, slides: list[dict[str, Any]], theme: str, tokens: int, cost: float, model_used: str, ) -> None: async with SessionLocal() as session: existing = await session.execute( select(SlideDeck).where(SlideDeck.task_id == task_id) ) deck = existing.scalar_one_or_none() if deck is None: deck = SlideDeck( task_id=task_id, slide_data=slides, theme=theme, slide_count=len(slides), ) session.add(deck) else: deck.slide_data = slides deck.theme = theme deck.slide_count = len(slides) session.add(deck) res = await session.execute(select(Task).where(Task.id == task_id)) task = res.scalar_one_or_none() if task is not None: task.status = "done" task.current_step = "done" task.progress_pct = 100.0 task.tokens_used = tokens task.cost_usd = round(cost, 6) task.model_used = model_used task.completed_at = datetime.now(timezone.utc) session.add(task) await session.commit()