""" BookEngine ========== Top-level orchestrator for the Book Engine. Sits **parallel** to ``ChatOrchestrator`` (i.e. it is **not** a ``BaseCapability``) and is the single public entry point used by the API router, CLI, and SDK. Lifecycle --------- :: create_book(...) → BookProposal (Stage 1, requires user confirm) confirm_proposal(...) → Spine (Stage 2, requires user confirm) confirm_spine(...) → page shells + queued compilation compile_page(book, page) → Page (Stage 3-4, drives BookCompiler) list_books(), load_book(), delete_book(), resume_book() Compilation queue ----------------- For each book a per-book ``asyncio.Queue`` schedules pages with priority: - Highest priority: page the user just opened (handled inline). - Background: remaining unfinished pages, processed by a single worker. The engine emits all progress over a ``StreamBus`` (wrapped by ``BookStream``) which the WebSocket router fans out to clients. """ from __future__ import annotations import asyncio from dataclasses import dataclass, field import logging import time from typing import Any from deeptutor.core.stream_bus import StreamBus from .agents.ideation_agent import IdeationAgent from .agents.source_explorer import SourceExplorer from .agents.spine_synthesizer import SpineSynthesizer from .compiler import BookCompiler, CompilerOptions from .inputs import IdeationContext, build_book_inputs from .models import ( Block, BlockStatus, BlockType, Book, BookInputs, BookProposal, BookStatus, Chapter, ContentType, ExplorationReport, Page, PageLink, PageStatus, Progress, QuizAttempt, Spine, ) from .storage import BookStorage, get_book_storage from .streaming import ( STAGE_COMPILATION, STAGE_CRITIQUE, STAGE_EXPLORATION, STAGE_IDEATION, STAGE_OVERVIEW, STAGE_SPINE, STAGE_SYNTHESIS, BookStream, ) logger = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────────────── # Per-book runtime state (queues, workers) # ───────────────────────────────────────────────────────────────────────────── @dataclass class _BookRuntime: """In-process per-book scheduling state.""" queue: asyncio.Queue[str] = field(default_factory=asyncio.Queue) queued: set[str] = field(default_factory=set) worker: asyncio.Task[None] | None = None lock: asyncio.Lock = field(default_factory=asyncio.Lock) stream: BookStream | None = None # default stream for background work # ───────────────────────────────────────────────────────────────────────────── # Engine # ───────────────────────────────────────────────────────────────────────────── class BookEngine: """Process-wide orchestrator for the Book Engine.""" def __init__( self, *, storage: BookStorage | None = None, compiler_options: CompilerOptions | None = None, ) -> None: self.storage = storage or get_book_storage() self.compiler = BookCompiler( storage=self.storage, options=compiler_options or CompilerOptions(phase=2), ) self._runtimes: dict[str, _BookRuntime] = {} self._global_lock = asyncio.Lock() # ── Discovery / lifecycle ──────────────────────────────────────────── def list_books(self) -> list[Book]: books: list[Book] = [] for book_id in self.storage.list_book_ids(): book = self.storage.load_book(book_id) if book is not None: books.append(book) books.sort(key=lambda b: b.updated_at, reverse=True) return books def load_book(self, book_id: str) -> Book | None: return self.storage.load_book(book_id) def load_spine(self, book_id: str) -> Spine | None: return self.storage.load_spine(book_id) def list_pages(self, book_id: str) -> list[Page]: return self.storage.list_pages(book_id) def load_page(self, book_id: str, page_id: str) -> Page | None: return self.storage.load_page(book_id, page_id) def load_progress(self, book_id: str) -> Progress: progress = self.storage.load_progress(book_id) if progress is None: progress = Progress(book_id=book_id) self.storage.save_progress(progress) return progress def delete_book(self, book_id: str) -> bool: runtime = self._runtimes.pop(book_id, None) if runtime and runtime.worker and not runtime.worker.done(): runtime.worker.cancel() return self.storage.delete_book(book_id) def set_page_chat_session(self, *, book_id: str, page_id: str, session_id: str) -> Book | None: """Persist the chat session associated with a specific book page.""" book = self.storage.load_book(book_id) page = self.storage.load_page(book_id, page_id) clean_session_id = (session_id or "").strip() if book is None or page is None or not clean_session_id: return None metadata = dict(book.metadata or {}) mapping = metadata.get("page_chat_sessions") if not isinstance(mapping, dict): mapping = {} mapping[str(page_id)] = clean_session_id metadata["page_chat_sessions"] = mapping book.metadata = metadata book.updated_at = time.time() self.storage.save_book(book) self.storage.append_log( book_id, f"page chat session mapped ({page_id} → {clean_session_id})", op="page_chat", ) return book @staticmethod def _reset_page_for_force_compile(page: Page) -> None: """Reset generated block outputs while preserving user-authored notes.""" for block in page.blocks: if block.type == BlockType.USER_NOTE: continue preserved_metadata = { key: value for key, value in (block.metadata or {}).items() if key in {"transition_in", "deep_dive_page_id"} } block.status = BlockStatus.PENDING block.payload = {} block.error = "" block.source_anchors = [] block.metadata = preserved_metadata block.updated_at = time.time() page.status = PageStatus.PENDING page.error = "" page.updated_at = time.time() # ── Stage 1: Ideation ──────────────────────────────────────────────── async def create_book( self, *, user_intent: str, chat_session_id: str = "", chat_selections: list[dict[str, Any]] | None = None, notebook_refs: list[dict[str, Any]] | None = None, knowledge_bases: list[str] | None = None, question_categories: list[int] | None = None, question_entries: list[int] | None = None, language: str = "en", stream: StreamBus | None = None, ) -> tuple[Book, BookProposal]: """Capture inputs, run IdeationAgent, persist DRAFT book + proposal.""" bus = stream or StreamBus() bstream = BookStream(bus) async with bstream.stage(STAGE_IDEATION): await bstream.progress("Capturing inputs…", stage=STAGE_IDEATION) book_inputs, ideation_ctx = await build_book_inputs( user_intent=user_intent, chat_session_id=chat_session_id, chat_selections=chat_selections, notebook_refs=notebook_refs, knowledge_bases=knowledge_bases, question_categories=question_categories, question_entries=question_entries, language=language, ) await bstream.progress( "Generating book proposal…", stage=STAGE_IDEATION, ) proposal = await self._run_ideation(ideation_ctx, language) book = Book( title=proposal.title, description=proposal.description, status=BookStatus.DRAFT, proposal=proposal, knowledge_bases=book_inputs.knowledge_bases, language=language, chapter_count=proposal.estimated_chapters, ) # Capture baseline KB fingerprints immediately so subsequent drift # checks have something to compare against. Without this the very # first health-check run treats every selected KB as "newly added" # and surfaces a spurious drift warning. try: from .kb_health import fingerprint_kbs if book.knowledge_bases: book.kb_fingerprints = fingerprint_kbs(book.knowledge_bases) except Exception as exc: # noqa: BLE001 logger.debug(f"baseline fingerprint capture skipped: {exc}") self.storage.save_book(book) self.storage.save_inputs(book.id, book_inputs) self.storage.save_progress(Progress(book_id=book.id)) self.storage.append_log( book.id, f"created (status=draft, title='{book.title}')", op="create" ) await bstream.book_event( "proposal_ready", { "book_id": book.id, "title": book.title, "proposal": proposal.model_dump(), }, stage=STAGE_IDEATION, ) return book, proposal async def _run_ideation(self, ctx: IdeationContext, language: str) -> BookProposal: agent = IdeationAgent(language=language) return await agent.process(ideation_context=ctx) # ── Stage 2: Spine ─────────────────────────────────────────────────── async def confirm_proposal( self, *, book_id: str, edited_proposal: BookProposal | None = None, stream: StreamBus | None = None, ) -> tuple[Book, Spine]: """User confirms (and optionally edits) the proposal → run SpineAgent.""" book = self.storage.load_book(book_id) if book is None: raise ValueError(f"Book {book_id} not found") if edited_proposal is not None: book.proposal = edited_proposal book.title = edited_proposal.title or book.title book.description = edited_proposal.description or book.description bus = stream or StreamBus() bstream = BookStream(bus) proposal = book.proposal or BookProposal(title=book.title) inputs = self.storage.load_inputs(book.id) or BookInputs( user_intent=book.title or "", knowledge_bases=list(book.knowledge_bases), language=book.language, ) async with bstream.stage(STAGE_SPINE): # ── Sub-stage 1: Source exploration ────────────────────── exploration: ExplorationReport | None = None try: async with bstream.stage(STAGE_EXPLORATION): await bstream.progress( "Exploring your sources in parallel…", stage=STAGE_EXPLORATION, ) explorer = SourceExplorer(language=book.language) exploration = await explorer.explore( book_id=book.id, proposal=proposal, inputs=inputs, ) self.storage.save_exploration(book.id, exploration) await bstream.book_event( "exploration_ready", { "book_id": book.id, "queries": exploration.queries, "coverage": exploration.coverage, "candidate_concepts": exploration.candidate_concepts, "summary": exploration.summary, }, stage=STAGE_EXPLORATION, ) except Exception as exc: logger.warning(f"SourceExplorer failed for {book.id}: {exc}") exploration = None await bstream.progress( "Source exploration unavailable — falling back to proposal-only spine.", stage=STAGE_EXPLORATION, ) # ── Sub-stage 2: Synthesise spine + concept graph ──────── synthesizer = SpineSynthesizer(language=book.language) async def _on_round(label: str, payload: dict[str, Any]) -> None: stage = STAGE_CRITIQUE if label.startswith("critique") else STAGE_SYNTHESIS # Don't push the full payload; just enough to drive the timeline. summary = { "round": label, "chapter_count": len(payload.get("chapters") or []) if isinstance(payload.get("chapters"), list) else 0, "issue_count": len(payload.get("issues") or []) if isinstance(payload.get("issues"), list) else 0, "verdict": payload.get("verdict") or "", } await bstream.book_event( "spine_round", {"book_id": book.id, **summary}, stage=stage ) async with bstream.stage(STAGE_SYNTHESIS): await bstream.progress("Synthesising spine + concept graph…", stage=STAGE_SYNTHESIS) spine = await synthesizer.synthesize( book_id=book.id, proposal=proposal, exploration=exploration, on_round=_on_round, ) book.chapter_count = len(spine.chapters) book.status = BookStatus.SPINE_READY self.storage.save_book(book) self.storage.save_spine(spine) self.storage.append_log( book.id, f"spine generated ({len(spine.chapters)} chapters, " f"{len(spine.concept_graph.nodes)} concepts)", op="spine", ) await bstream.book_event( "spine_ready", { "book_id": book.id, "chapter_count": len(spine.chapters), "concept_node_count": len(spine.concept_graph.nodes), "concept_edge_count": len(spine.concept_graph.edges), "spine": spine.model_dump(), }, stage=STAGE_SPINE, ) return book, spine # ── Stage 2.5: Overview chapter injection ─────────────────────────── async def _ensure_overview_chapter( self, spine: Spine, book: Book, *, stream: StreamBus | None, ) -> Spine: """Insert an Overview chapter at position 0 (idempotent).""" # Idempotent guard — already injected? first = spine.chapters[0] if spine.chapters else None already = bool( first and ( first.content_type == ContentType.OVERVIEW or (first.__pydantic_extra__ or {}).get("auto_overview") is True ) ) if already: return spine overview_title = "本书导览" if book.language == "zh" else "How to read this book" objectives = ( [ "了解整本书的章节脉络", "掌握各章之间的概念依赖关系", "选择最合适的阅读顺序", ] if book.language == "zh" else [ "See the full chapter map at a glance", "Understand how concepts depend on each other", "Pick the reading path that fits your goals", ] ) overview = Chapter( title=overview_title, learning_objectives=objectives, content_type=ContentType.OVERVIEW, summary=( "Auto-generated overview of the book's concept graph and chapter index." if book.language != "zh" else "自动生成的概念图与章节索引，作为本书的入口。" ), order=0, ) # Mark for idempotency on subsequent runs. overview.__pydantic_extra__ = overview.__pydantic_extra__ or {} overview.__pydantic_extra__["auto_overview"] = True # Re-number existing chapters down by one. for ch in spine.chapters: ch.order = ch.order + 1 spine.chapters = [overview, *spine.chapters] return spine async def _materialize_overview_page( self, spine: Spine, pages: list[Page], book: Book, *, stream: StreamBus | None, ) -> None: """Build the Overview page's blocks deterministically.""" if not spine.chapters: return overview_chapter = spine.chapters[0] if overview_chapter.content_type != ContentType.OVERVIEW: return overview_page = next((p for p in pages if p.chapter_id == overview_chapter.id), None) if overview_page is None or overview_page.status == PageStatus.READY: return zh = book.language == "zh" chapter_index = [ { "id": ch.id, "title": ch.title, "summary": ch.summary, "objectives": list(ch.learning_objectives), "order": ch.order, "content_type": ch.content_type.value, "page_id": (ch.page_ids[0] if ch.page_ids else ""), } for ch in spine.chapters if ch.content_type != ContentType.OVERVIEW ] # 1) Intro text block (pre-rendered, status=READY) intro_md = ( ( f"# {book.title or '本书'}\n\n" f"{(book.proposal.description if book.proposal else '') or ''}\n\n" f"下方的概念图展示了本书 {len(spine.concept_graph.nodes)} 个核心概念以及" f"它们之间的依赖关系；再下方是 {len(chapter_index)} 个章节的入口。" f"你可以按从上到下的顺序阅读，也可以根据自己的兴趣或先验知识选择切入点。" ) if zh else ( f"# {book.title or 'This book'}\n\n" f"{(book.proposal.description if book.proposal else '') or ''}\n\n" f"The diagram below maps the {len(spine.concept_graph.nodes)} core " f"concepts in this book and how they depend on each other. The " f"chapter index that follows lists all {len(chapter_index)} " f"chapters — read top-to-bottom for the recommended path, or jump " f"straight to whatever you're most curious about." ) ) intro_block = Block( type=BlockType.TEXT, status=BlockStatus.READY, title=("如何阅读这本书" if zh else "How to read this book"), params={"role": "overview_intro"}, payload={"content": intro_md, "format": "markdown"}, ) # 2) Concept graph block — render deterministically from .blocks.concept_graph import render_mermaid graph_block = Block( type=BlockType.CONCEPT_GRAPH, status=BlockStatus.READY, title=("概念图" if zh else "Concept map"), params={ "concept_graph": spine.concept_graph.model_dump(), "chapter_index": chapter_index, }, payload={ "render_type": "concept_graph", "code": { "language": "mermaid", "content": render_mermaid(spine.concept_graph), }, "graph": spine.concept_graph.model_dump(), "index": { "chapters": chapter_index, "node_to_chapter": { n.id: n.chapter_id for n in spine.concept_graph.nodes if n.chapter_id }, }, }, metadata={ "node_count": len(spine.concept_graph.nodes), "edge_count": len(spine.concept_graph.edges), }, ) # 3) Chapter index callout — also rendered deterministically index_lines = [] for entry in chapter_index: line = f"- **{entry['title']}**" if entry.get("summary"): line += f" — {entry['summary']}" index_lines.append(line) index_md = ("## 章节索引\n\n" if zh else "## Chapter index\n\n") + "\n".join(index_lines) index_block = Block( type=BlockType.TEXT, status=BlockStatus.READY, title=("章节索引" if zh else "Chapter index"), params={"role": "chapter_index"}, payload={"content": index_md, "format": "markdown"}, ) overview_page.blocks = [intro_block, graph_block, index_block] overview_page.status = PageStatus.READY overview_page.content_type = ContentType.OVERVIEW self.storage.save_page(overview_page) if stream is not None: bstream = BookStream(stream) await bstream.book_event( "overview_ready", { "book_id": book.id, "page_id": overview_page.id, "node_count": len(spine.concept_graph.nodes), "chapter_count": len(chapter_index), }, stage=STAGE_OVERVIEW, ) # ── Stage 3: confirm spine + create page shells ───────────────────── async def confirm_spine( self, *, book_id: str, edited_spine: Spine | None = None, stream: StreamBus | None = None, auto_compile: bool = True, ) -> list[Page]: """User confirms (or edits) the spine → create pending page shells. BookEngine v2: automatically injects an **Overview** chapter at order 0 whose page is fully pre-built (deterministic concept-graph render + intro text + chapter index). The rest of the chapters are queued for normal compilation. """ book = self.storage.load_book(book_id) if book is None: raise ValueError(f"Book {book_id} not found") spine = edited_spine or self.storage.load_spine(book_id) if spine is None: raise ValueError(f"No spine for book {book_id}") if edited_spine is not None: spine.book_id = book_id self.storage.save_spine(spine) # ── Inject Overview chapter (idempotent) ───────────────────── spine = await self._ensure_overview_chapter(spine, book, stream=stream) self.storage.save_spine(spine) existing = {p.chapter_id: p for p in self.storage.list_pages(book_id)} pages: list[Page] = [] for chapter in spine.chapters: page = existing.get(chapter.id) if page is None: page = Page( book_id=book_id, chapter_id=chapter.id, title=chapter.title, learning_objectives=list(chapter.learning_objectives), content_type=chapter.content_type, order=chapter.order, status=PageStatus.PENDING, ) self.storage.save_page(page) chapter.page_ids = [page.id] self.storage.save_spine(spine) pages.append(page) # Build the Overview page eagerly (no LLM, no queue). await self._materialize_overview_page(spine, pages, book, stream=stream) book.page_count = len(pages) book.status = BookStatus.COMPILING if auto_compile else BookStatus.SPINE_READY self.storage.save_book(book) self.storage.append_log( book_id, f"spine confirmed ({len(pages)} page shells, auto_compile={auto_compile})", op="confirm_spine", ) if auto_compile: await self._enqueue_pending_pages(book_id, pages, stream=stream) return pages async def rebuild_book( self, *, book_id: str, stream: StreamBus | None = None, auto_compile: bool = True, ) -> list[Page]: """Regenerate all pages while preserving the confirmed proposal/spine.""" book = self.storage.load_book(book_id) spine = self.storage.load_spine(book_id) if book is None or spine is None: raise ValueError(f"Cannot rebuild book – missing book/spine ({book_id})") runtime = self._runtimes.get(book_id) if runtime is not None: async with runtime.lock: if runtime.worker is not None and not runtime.worker.done(): runtime.worker.cancel() runtime.worker = None runtime.queued.clear() while not runtime.queue.empty(): try: runtime.queue.get_nowait() except asyncio.QueueEmpty: break for page in self.storage.list_pages(book_id): self.storage.delete_page(book_id, page.id) for chapter in spine.chapters: chapter.page_ids = [] self.storage.save_spine(spine) self.storage.save_progress(Progress(book_id=book_id)) book.status = BookStatus.SPINE_READY book.page_count = 0 book.updated_at = time.time() self.storage.save_book(book) self.storage.append_log( book_id, f"rebuild requested (preserve_spine=true, auto_compile={auto_compile})", op="rebuild", ) return await self.confirm_spine( book_id=book_id, edited_spine=spine, stream=stream, auto_compile=auto_compile, ) # ── Stage 3-4: compile a single page (current page) ────────────────── async def compile_page( self, *, book_id: str, page_id: str, stream: StreamBus | None = None, force: bool = False, ) -> Page: """Drive the compiler for one page (used when a user opens it).""" book = self.storage.load_book(book_id) spine = self.storage.load_spine(book_id) page = self.storage.load_page(book_id, page_id) if book is None or spine is None or page is None: raise ValueError(f"Cannot compile page – missing book/spine/page ({book_id}/{page_id})") if page.status == PageStatus.READY and not force: return page # Overview pages are built deterministically from the spine (intro, # concept graph, chapter index). Never run the generic LLM compiler over # their hand-built blocks — that would overwrite the deterministic # "How to read this book" intro and chapter index with hallucinated # prose. Reaching here means force=True or the page is not yet READY, so # rebuild deterministically (this also refreshes a changed spine). if page.content_type == ContentType.OVERVIEW: page.status = PageStatus.PENDING self.storage.save_page(page) await self._materialize_overview_page(spine, [page], book, stream=stream) page = self.storage.load_page(book_id, page_id) or page await self._maybe_finalize_book(book_id) return page if force: self._reset_page_for_force_compile(page) self.storage.save_page(page) chapter = spine.chapter_by_id(page.chapter_id) if chapter is None: raise ValueError(f"Page {page_id} references unknown chapter {page.chapter_id}") bus = stream or StreamBus() bstream = BookStream(bus) async with bstream.stage(STAGE_COMPILATION): page = await self.compiler.compile_page( book_id=book_id, chapter=chapter, page=page, stream=bstream, knowledge_bases=book.knowledge_bases, language=book.language, ) # Refresh KB fingerprints once we successfully ship a READY page. # We capture them lazily so a brand-new book gets its baseline as # soon as the very first page is compiled. if page.status == PageStatus.READY: try: from .kb_health import refresh_book_fingerprints refreshed = self.storage.load_book(book_id) if refreshed is not None and not refreshed.kb_fingerprints: refresh_book_fingerprints(book_id, storage=self.storage) except Exception as exc: # noqa: BLE001 logger.debug(f"fingerprint refresh skipped: {exc}") await self._maybe_finalize_book(book_id) return page # ── Background compilation queue ───────────────────────────────────── async def _enqueue_pending_pages( self, book_id: str, pages: list[Page], *, stream: StreamBus | None = None, ) -> None: runtime = await self._get_or_create_runtime(book_id, stream) async with runtime.lock: for page in pages: if page.status == PageStatus.READY: continue if page.id in runtime.queued: continue runtime.queued.add(page.id) await runtime.queue.put(page.id) self._ensure_worker(book_id) async def _get_or_create_runtime(self, book_id: str, stream: StreamBus | None) -> _BookRuntime: async with self._global_lock: runtime = self._runtimes.get(book_id) if runtime is None: runtime = _BookRuntime() self._runtimes[book_id] = runtime if stream is not None and runtime.stream is None: runtime.stream = BookStream(stream) elif runtime.stream is None: runtime.stream = BookStream(StreamBus()) return runtime def _ensure_worker(self, book_id: str) -> None: runtime = self._runtimes.get(book_id) if runtime is None: return if runtime.worker is not None and not runtime.worker.done(): return runtime.worker = asyncio.create_task(self._worker_loop(book_id)) async def _worker_loop(self, book_id: str) -> None: runtime = self._runtimes.get(book_id) if runtime is None: return bstream = runtime.stream or BookStream(StreamBus()) while True: try: page_id = await asyncio.wait_for(runtime.queue.get(), timeout=2.0) except asyncio.TimeoutError: async with runtime.lock: if runtime.queue.empty(): runtime.worker = None return continue try: book = self.storage.load_book(book_id) spine = self.storage.load_spine(book_id) page = self.storage.load_page(book_id, page_id) if book is None or spine is None or page is None: continue if page.status == PageStatus.READY: continue chapter = spine.chapter_by_id(page.chapter_id) if chapter is None: continue await self.compiler.compile_page( book_id=book_id, chapter=chapter, page=page, stream=bstream, knowledge_bases=book.knowledge_bases, language=book.language, ) except asyncio.CancelledError: raise except Exception as exc: logger.warning( f"Background compilation failed for {book_id}/{page_id}: {exc}", exc_info=True, ) self.storage.append_log( book_id, f"background compile failed for page {page_id}: {exc}", op="compile_error", ) finally: runtime.queued.discard(page_id) await self._maybe_finalize_book(book_id) async def _maybe_finalize_book(self, book_id: str) -> None: book = self.storage.load_book(book_id) if book is None: return pages = self.storage.list_pages(book_id) if not pages: return if all(p.status == PageStatus.READY for p in pages): if book.status != BookStatus.READY: book.status = BookStatus.READY self.storage.save_book(book) self.storage.append_log(book_id, "all pages ready → status=READY", op="finalize") # ── Block-level controls (Phase 1: regenerate single block) ───────── async def regenerate_block( self, *, book_id: str, page_id: str, block_id: str, params_override: dict[str, Any] | None = None, stream: StreamBus | None = None, ) -> Block | None: """Re-run a single block generator (e.g. user clicked 'regenerate').""" book = self.storage.load_book(book_id) spine = self.storage.load_spine(book_id) page = self.storage.load_page(book_id, page_id) if book is None or spine is None or page is None: return None chapter = spine.chapter_by_id(page.chapter_id) if chapter is None: return None block = page.block_by_id(block_id) if block is None: return None if params_override: block.params = {**block.params, **params_override} bus = stream or StreamBus() bstream = BookStream(bus) from .blocks.base import BlockContext, get_block_registry registry = get_block_registry() generator = registry.get(block.type) if generator is None: block.error = f"No generator for {block.type.value}" self.storage.save_page(page) return block ctx = BlockContext( book_id=book_id, chapter=chapter, page=page, block=block, language=book.language, knowledge_bases=book.knowledge_bases, ) # Resolve previous block by planning order so the bridge text can be # refreshed alongside the block itself. prev_block: Block | None = None for candidate in page.blocks: if candidate.id == block.id: break prev_block = candidate async with bstream.stage(STAGE_COMPILATION): await generator.generate(ctx) if block.status.value == "ready": await self.compiler.attach_bridge_text( block, chapter=chapter, previous_block=prev_block, language=book.language, ) self.storage.save_page(page) self.compiler._finalize_page_status(page) self.storage.save_page(page) await self._maybe_finalize_book(book_id) return block # ── Maintenance / health (Phase 4) ───────────────────────────────── def kb_drift_report(self, book_id: str) -> dict[str, Any]: """Compute and persist the current KB drift report for *book_id*.""" from .kb_health import mark_drift_on_book report = mark_drift_on_book(book_id, storage=self.storage) if report is None: return {"book_id": book_id, "has_drift": False, "missing": True} return report.to_dict() def refresh_kb_fingerprints(self, book_id: str) -> dict[str, Any] | None: from .kb_health import refresh_book_fingerprints book = refresh_book_fingerprints(book_id, storage=self.storage) if book is None: return None return { "book_id": book.id, "kb_fingerprints": book.kb_fingerprints, "stale_page_ids": book.stale_page_ids, } def log_health(self, book_id: str) -> dict[str, Any]: from .kb_health import scan_log_health return scan_log_health(book_id, storage=self.storage).to_dict() # ── Block CRUD operations (Phase 3) ──────────────────────────────── async def insert_block( self, *, book_id: str, page_id: str, block_type: BlockType, params: dict[str, Any] | None = None, position: int | None = None, stream: StreamBus | None = None, compile_now: bool = True, ) -> Block | None: """Insert a fresh PENDING block at *position* (default: end).""" spine = self.storage.load_spine(book_id) page = self.storage.load_page(book_id, page_id) if spine is None or page is None: return None chapter = spine.chapter_by_id(page.chapter_id) if chapter is None: return None merged_params: dict[str, Any] = { "chapter_title": chapter.title, "chapter_summary": chapter.summary, "objectives": chapter.learning_objectives, **(params or {}), } block = Block(type=block_type, status=BlockStatus.PENDING, params=merged_params) if position is None or position >= len(page.blocks) or position < 0: page.blocks.append(block) else: page.blocks.insert(position, block) self.storage.save_page(page) if compile_now and block_type != BlockType.USER_NOTE: from .blocks.base import BlockContext, get_block_registry generator = get_block_registry().get(block_type) if generator is not None: ctx = BlockContext( book_id=book_id, chapter=chapter, page=page, block=block, language=self.storage.load_book(book_id).language if self.storage.load_book(book_id) else "en", knowledge_bases=self.storage.load_book(book_id).knowledge_bases if self.storage.load_book(book_id) else [], ) bus = stream or StreamBus() bstream = BookStream(bus) async with bstream.stage(STAGE_COMPILATION): await generator.generate(ctx) self.compiler._finalize_page_status(page) self.storage.save_page(page) elif block_type == BlockType.USER_NOTE: block.status = BlockStatus.READY block.payload = { "format": "markdown", "body": str(merged_params.get("body") or ""), "author": "user", } self.storage.save_page(page) self.storage.append_log( book_id, f"inserted {block_type.value} block on page {page_id} (pos={position})", op="insert_block", ) return block async def delete_block(self, *, book_id: str, page_id: str, block_id: str) -> bool: page = self.storage.load_page(book_id, page_id) if page is None: return False before = len(page.blocks) page.blocks = [b for b in page.blocks if b.id != block_id] if len(page.blocks) == before: return False self.compiler._finalize_page_status(page) self.storage.save_page(page) self.storage.append_log( book_id, f"deleted block {block_id} from page {page_id}", op="delete_block" ) return True async def move_block( self, *, book_id: str, page_id: str, block_id: str, new_position: int ) -> bool: page = self.storage.load_page(book_id, page_id) if page is None: return False idx = next((i for i, b in enumerate(page.blocks) if b.id == block_id), -1) if idx < 0: return False new_position = max(0, min(len(page.blocks) - 1, new_position)) block = page.blocks.pop(idx) page.blocks.insert(new_position, block) self.storage.save_page(page) self.storage.append_log( book_id, f"moved block {block_id} on page {page_id} → pos {new_position}", op="move_block", ) return True async def change_block_type( self, *, book_id: str, page_id: str, block_id: str, new_type: BlockType, params_override: dict[str, Any] | None = None, stream: StreamBus | None = None, ) -> Block | None: spine = self.storage.load_spine(book_id) page = self.storage.load_page(book_id, page_id) if spine is None or page is None: return None chapter = spine.chapter_by_id(page.chapter_id) if chapter is None: return None block = page.block_by_id(block_id) if block is None: return None block.type = new_type block.status = BlockStatus.PENDING block.payload = {} block.error = "" if params_override: block.params = {**block.params, **params_override} self.storage.save_page(page) # Re-run generator immediately return await self.regenerate_block( book_id=book_id, page_id=page_id, block_id=block_id, stream=stream ) # ── Deep-dive sub-page (Phase 3) ────────────────────────────────── async def create_deep_dive_subpage( self, *, book_id: str, parent_page_id: str, topic: str, block_id: str | None = None, content_type: ContentType = ContentType.CONCEPT, stream: StreamBus | None = None, ) -> Page | None: """Spawn a child Page that deepens *topic* and link it from the parent.""" book = self.storage.load_book(book_id) spine = self.storage.load_spine(book_id) parent = self.storage.load_page(book_id, parent_page_id) if book is None or spine is None or parent is None: return None # Add a synthetic chapter so the planner has a target chapter = Chapter( title=f"{topic} (deep dive)", learning_objectives=[f"Go deeper into {topic}"], content_type=content_type, summary=f"Sub-chapter spawned from {parent.title}.", order=len(spine.chapters), ) spine.chapters.append(chapter) self.storage.save_spine(spine) sub = Page( book_id=book_id, chapter_id=chapter.id, title=topic, learning_objectives=list(chapter.learning_objectives), content_type=content_type, status=PageStatus.PENDING, order=len(self.storage.list_pages(book_id)), parent_page_id=parent.id, ) self.storage.save_page(sub) chapter.page_ids = [sub.id] self.storage.save_spine(spine) # Add link from parent → sub parent.links.append(PageLink(target_page_id=sub.id, relation="deepens", label=topic)) if block_id: block = parent.block_by_id(block_id) if block is not None: block.metadata = {**block.metadata, "deep_dive_page_id": sub.id} self.storage.save_page(parent) # Compile the new page now (blocking, so caller gets ready content) await self.compile_page(book_id=book_id, page_id=sub.id, stream=stream) self.storage.append_log( book_id, f"deep-dive page {sub.id} spawned from {parent_page_id}: {topic}", op="deep_dive", ) return self.storage.load_page(book_id, sub.id) # ── Quiz attempts (Phase 3) ──────────────────────────────────────── async def record_quiz_attempt( self, *, book_id: str, page_id: str, block_id: str, question_id: str, user_answer: str, is_correct: bool, ) -> Progress: progress = self.load_progress(book_id) progress.quiz_attempts.append( QuizAttempt( block_id=block_id, page_id=page_id, question_id=question_id, user_answer=user_answer, is_correct=is_correct, ) ) if not is_correct: page = self.storage.load_page(book_id, page_id) if page and page.chapter_id and page.chapter_id not in progress.weak_chapters: progress.weak_chapters.append(page.chapter_id) else: progress.score += 1 self.storage.save_progress(progress) return progress async def supplement_for_weakness( self, *, book_id: str, page_id: str, topic: str, stream: StreamBus | None = None, ) -> Block | None: """Append an extra TEXT + QUIZ block when the user struggles.""" await self.insert_block( book_id=book_id, page_id=page_id, block_type=BlockType.CALLOUT, params={"variant": "common_pitfall", "topic": topic}, stream=stream, ) await self.insert_block( book_id=book_id, page_id=page_id, block_type=BlockType.TEXT, params={"role": "remediation", "topic": topic}, stream=stream, ) return await self.insert_block( book_id=book_id, page_id=page_id, block_type=BlockType.QUIZ, params={"num_questions": 2, "difficulty": "easy", "topic": topic}, stream=stream, ) # ───────────────────────────────────────────────────────────────────────────── # Singleton accessor # ───────────────────────────────────────────────────────────────────────────── _engines: dict[str, BookEngine] = {} def get_book_engine() -> BookEngine: from deeptutor.services.path_service import get_path_service key = str(get_path_service().workspace_root.resolve()) if key not in _engines: _engines[key] = BookEngine() return _engines[key] __all__ = ["BookEngine", "get_book_engine"]