wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
1250 lines
48 KiB
Python
1250 lines
48 KiB
Python
"""
|
||
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"]
|