diff_chapter
Compare a chapter's current draft with its .prev.md snapshot to see changes since the last writer overwrite.
Instructions
Diff current draft vs .prev.md snapshot from last writer overwrite.
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| chapter_num | Yes | ||
| max_lines | No |
Output Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/storywright_mcp/workflow.py:611-633 (handler)Core implementation of diff_chapter. Reads the current draft and the .prev.md snapshot, computes a unified diff using difflib, truncates to max_lines, and returns a formatted diff block.
def diff_chapter(chapter_num: int, max_lines: int = 150) -> str: """Unified diff: previous draft snapshot (.prev.md) vs current draft.""" proj, _ = require_project() draft = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.md" prev = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.prev.md" if not draft.exists(): return "No current draft." if not prev.exists(): return ( "No `chapter-NN-draft.prev.md` — created when **run_writer_agent** overwrites " "an existing draft." ) a = prev.read_text(encoding="utf-8", errors="replace").splitlines() b = draft.read_text(encoding="utf-8", errors="replace").splitlines() diff = list(difflib.unified_diff(a, b, fromfile="draft.prev", tofile="draft", lineterm="")) if not diff: return "No textual difference between snapshot and current draft." cap = max_lines + 10 tail = diff[:cap] extra = "" if len(diff) > cap: extra = f"\n… {len(diff) - cap} more diff lines …" return "```diff\n" + "\n".join(tail) + extra + "\n```" - src/storywright_mcp/app.py:308-314 (registration)MCP tool registration of diff_chapter via @mcp.tool() decorator. Wraps workflow.diff_chapter as an async function.
@mcp.tool() async def diff_chapter(chapter_num: int, max_lines: int = 150) -> str: """Diff current draft vs `.prev.md` snapshot from last writer overwrite.""" try: return workflow.diff_chapter(chapter_num, max_lines) except ValueError as e: return str(e) - The difflib module is imported in workflow.py and used by diff_chapter to compute the unified diff.
import difflib import os import re from pathlib import Path import anthropic from . import __version__ from .agents.editor import build_editor_prompt from .agents.registry import get_agent, list_agents, render_agent_prompt from .agents.writer import build_writer_prompt from .config import get_settings from .editor_meta import extract_verdict_from_report, parse_meta_block, save_editor_meta from .llm import anthropic_auth_mode, api_key_problem_message, complete_user_prompt, get_client from .prior_chapters import gather_prior_approved_excerpt from .revision_queue import append_revision, format_revision_section from .models.continuity import ( CharacterState, CharacterStatus, ContinuityLog, EstablishedFact, InventoryItem, LocationVisit, ) from .models.project import ( ChapterStatus, Character, DeathEntry, ProjectConfig, RunningGag, ) from .session import bind_project, require_project, save_project_and_continuity from .templates.defaults import get_default_comedy_brief, get_default_editor_brief, get_default_writer_brief def create_book_project( project_name: str, book_title: str, genre: str = "", authors: list[str] | None = None, projects_root: str | None = None, third_agents: list[str] | None = None, ) -> str: root = Path(projects_root).resolve() if projects_root else get_settings().projects_root.resolve() base_path = root / "book_projects" / project_name.strip().replace(" ", "-").lower() base_path.mkdir(parents=True, exist_ok=True) (base_path / "briefs").mkdir(exist_ok=True) (base_path / "reports").mkdir(exist_ok=True) (base_path / "manuscript").mkdir(exist_ok=True) agents = ["comedy"] if third_agents is None else list(third_agents) proj = ProjectConfig( name=book_title, genre=genre, authors=authors or [], base_path=base_path, third_agents=agents, ) proj.save() cont = ContinuityLog() cont.save(base_path) (base_path / "briefs" / "writer_brief.md").write_text(get_default_writer_brief(), encoding="utf-8") (base_path / "briefs" / "editor_brief.md").write_text(get_default_editor_brief(), encoding="utf-8") (base_path / "briefs" / "comedy_brief.md").write_text(get_default_comedy_brief(), encoding="utf-8") bind_project(base_path) return ( f"Created **{book_title}** at `{base_path}`.\n" f"Third-pass agents: {', '.join(agents)}\n\n" f"Next: `load_book_project(\"{base_path}\")` from another cwd, or continue — project is loaded." ) def load_book_project(project_path: str) -> str: bind_project(Path(project_path)) proj, _ = require_project() return ( f"Loaded **{proj.name}** (`{proj.base_path}`) — " f"{len(proj.chapters)} chapters, {len(proj.characters)} characters." ) def check_environment() -> str: """Validate Anthropic API env and report Storywright version.""" msg = api_key_problem_message() lines = [ f"**Storywright** v{__version__}\n", ] if msg: lines.append(f"- API: **NOT READY** — {msg}\n") else: lines.append(f"- API credentials: {anthropic_auth_mode()}\n") bu = os.environ.get("ANTHROPIC_BASE_URL", "").strip() if bu: lines.append(f"- ANTHROPIC_BASE_URL: `{bu}`\n") s = get_settings() lines.append(f"- Model: `{s.anthropic_model}`\n") lines.append( f"- Prior chapter budget: max {s.prior_chapters_max_words} words / " f"{s.prior_chapters_max_count} chapters\n" ) return "".join(lines) def get_pipeline_status() -> str: """Human-readable next-step hints for every chapter + global checks.""" proj, _ = require_project() env = check_environment() lines = [env, "\n## Pipeline\n\n"] if not proj.chapters: lines.append("No chapters — use **add_chapter**.\n") return "".join(lines) for ch in sorted(proj.chapters, key=lambda c: c.num): st = ch.status if st == ChapterStatus.NOT_STARTED: nxt = f"`start_chapter({ch.num})`" elif st == ChapterStatus.DRAFT: nxt = f"`run_writer_agent({ch.num})`" elif st == ChapterStatus.AWAITING_EDITOR: nxt = f"`run_editor_review({ch.num})`" elif st == ChapterStatus.AWAITING_THIRD_PASS: missing = [a for a in proj.third_agents if a not in ch.third_pass_completed] if not proj.third_agents: nxt = f"`approve_chapter({ch.num})`" elif missing: nxt = f"`run_third_agent({ch.num}, agent_type=\"{missing[0]}\")` then others: {missing}" else: nxt = f"`approve_chapter({ch.num})`" else: nxt = "done" lines.append(f"- **Ch.{ch.num}** ({ch.status.value}) → {nxt}\n") return "".join(lines) def get_project_status() -> str: proj, _ = require_project() lines = [f"# {proj.name}\n", f"Genre: {proj.genre}\n\n"] if not proj.chapters: lines.append("No chapters — use add_chapter.\n") return "".join(lines) lines.append("| # | Title | Words | Status | Editor verdict | Third passes |\n") lines.append("|---:|---|---:|---|---|---|\n") for ch in sorted(proj.chapters, key=lambda c: c.num): tp = ", ".join(ch.third_pass_completed) or "—" lines.append( f"| {ch.num} | {ch.title} | {ch.target_words} | {ch.status.value} | " f"{ch.editor_verdict or '—'} | {tp} |\n" ) return "".join(lines) def get_continuity_log() -> str: _, cont = require_project() import json return json.dumps(cont.to_dict(), indent=2) def add_chapter(chapter_num: int, title: str, target_words: int = 5000) -> str: proj, _ = require_project() if proj.get_chapter(chapter_num): return f"Chapter {chapter_num} already exists." from .models.project import Chapter proj.chapters.append( Chapter( num=chapter_num, title=title, target_words=target_words, status=ChapterStatus.NOT_STARTED, ) ) save_project_and_continuity() return f"Added chapter {chapter_num}: {title}" def add_character( name: str, role: str, description: str = "", voice_notes: str = "", comedy_hook: str = "", ) -> str: proj, cont = require_project() proj.characters.append( Character(name=name, role=role, description=description, voice_notes=voice_notes, comedy_hook=comedy_hook) ) cont.living_characters.append(CharacterState(name=name, status=CharacterStatus.ALIVE)) save_project_and_continuity() return f"Added character {name}" def add_death(character: str, chapter: int, circumstances: str, death_style: str = "") -> str: proj, _ = require_project() proj.death_schedule.append( DeathEntry(character=character, chapter=chapter, circumstances=circumstances, death_style=death_style) ) save_project_and_continuity() return f"Death scheduled: {character} in chapter {chapter}" def add_running_gag( name: str, owner: str, setup_chapter: int, description: str = "", escalation_pattern: str = "", ) -> str: proj, _ = require_project() proj.running_gags.append( RunningGag( name=name, owner=owner, setup_chapter=setup_chapter, description=description, escalation_pattern=escalation_pattern, ) ) save_project_and_continuity() return f"Running gag added: {name}" def add_inventory_item( name: str, initial_holder: str, description: str = "", acquired_chapter: int | None = None, ) -> str: _, cont = require_project() cont.inventory.append( InventoryItem( name=name, description=description, acquired_chapter=acquired_chapter, current_holder=initial_holder, original_holder=initial_holder, ) ) save_project_and_continuity() return f"Item {name} → {initial_holder}" def start_chapter(chapter_num: int) -> str: proj, _ = require_project() chapter = proj.get_chapter(chapter_num) if not chapter: return f"Chapter {chapter_num} not found." if chapter.status != ChapterStatus.NOT_STARTED: return f"Chapter {chapter_num} already started ({chapter.status.value})." chapter.status = ChapterStatus.DRAFT save_project_and_continuity() draft = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.md" draft.write_text(f"# Chapter {chapter_num}: {chapter.title}\n\n", encoding="utf-8") return f"Started chapter {chapter_num}. Draft: `{draft}`" def _backup_draft_if_exists(draft_path: Path) -> None: """Save previous draft as sibling `chapter-NN-draft.prev.md` before overwriting.""" if not draft_path.exists(): return prev_path = draft_path.parent / f"{draft_path.stem}.prev.md" prev_path.write_bytes(draft_path.read_bytes()) def run_writer_agent(chapter_num: int, chapter_brief: str = "") -> str: bad = api_key_problem_message() if bad: return bad proj, cont = require_project() chapter = proj.get_chapter(chapter_num) if not chapter or chapter.status != ChapterStatus.DRAFT: return f"Chapter {chapter_num} must be in draft status (call start_chapter first)." settings = get_settings() excerpt, prior_warnings = gather_prior_approved_excerpt( proj, chapter_num, max_words=settings.prior_chapters_max_words, max_chapters=settings.prior_chapters_max_count, ) rev = format_revision_section(proj.base_path, chapter_num) prompt = build_writer_prompt( proj, cont, chapter, chapter_brief, revision_section=rev, prior_excerpt_section=excerpt, ) client = get_client() draft_path = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.md" _backup_draft_if_exists(draft_path) try: text, warnings = complete_user_prompt( client=client, model=settings.anthropic_model, prompt=prompt, max_tokens=16384, ) except anthropic.APIError as e: return ( f"Anthropic API error ({type(e).__name__}): {e}\n" f"If rate-limited or overloaded, retry shortly; check ANTHROPIC_API_KEY and model id." ) draft_path.write_text(text, encoding="utf-8") chapter.status = ChapterStatus.AWAITING_EDITOR chapter.third_pass_completed = [] save_project_and_continuity() warn_lines = list(warnings) + prior_warnings warn = "\n".join(warn_lines) if warn_lines else "" return ( f"Writer finished chapter {chapter_num}. Draft saved (previous draft backed up to " f"`{draft_path.stem}.prev.md` if it existed).\n{warn}\n\nNext: run_editor_review({chapter_num})" ) def run_editor_review(chapter_num: int) -> str: bad = api_key_problem_message() if bad: return bad proj, cont = require_project() chapter = proj.get_chapter(chapter_num) if not chapter: return "Chapter not found." if chapter.status != ChapterStatus.AWAITING_EDITOR: return f"Editor can only run after writer (status is {chapter.status.value})." draft_path = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.md" if not draft_path.exists(): return "No draft file." draft_content = draft_path.read_text(encoding="utf-8") prompt = build_editor_prompt(proj, cont, chapter, draft_content) settings = get_settings() client = get_client() try: text, warnings = complete_user_prompt( client=client, model=settings.anthropic_model, prompt=prompt, max_tokens=8192 ) except anthropic.APIError as e: return f"Anthropic API error ({type(e).__name__}): {e}" report_path = proj.base_path / "reports" / f"chapter-{chapter_num:02d}-editor-report.md" report_path.parent.mkdir(exist_ok=True) report_path.write_text(text, encoding="utf-8") parsed = parse_meta_block(text) verdict = parsed.get("verdict") or extract_verdict_from_report(text) save_editor_meta(proj.base_path, chapter_num, text, parsed) chapter.editor_verdict = verdict chapter.status = ChapterStatus.AWAITING_THIRD_PASS save_project_and_continuity() warn = "\n".join(warnings) if warnings else "" meta_path = proj.base_path / "reports" / f"chapter-{chapter_num:02d}-editor-meta.json" meta_hint = f"\nStructured meta: `{meta_path}`" if meta_path.exists() else "" if not proj.third_agents: return ( f"Editor report saved. No third-pass agents configured — approve when ready.\n{warn}\n" f"Report: `{report_path}`{meta_hint}\n\nNext: approve_chapter({chapter_num})" ) return ( f"Editor verdict: **{verdict}**\n{warn}\nReport: `{report_path}`{meta_hint}\n\n" f"Next: run_third_agent({chapter_num}, agent_type=\"{proj.third_agents[0]}\") " f"( repeat for each configured agent )." ) def run_third_agent(chapter_num: int, agent_type: str | None = None) -> str: bad = api_key_problem_message() if bad: return bad proj, cont = require_project() chapter = proj.get_chapter(chapter_num) if not chapter: return "Chapter not found." if chapter.status != ChapterStatus.AWAITING_THIRD_PASS: return ( f"Third-pass runs only in awaiting_third_pass status (after editor). Current: {chapter.status.value}" ) if not proj.third_agents: return "No third-pass agents configured — use approve_chapter." agent_name = (agent_type or proj.third_agents[0]).lower().strip() agent = get_agent(agent_name) if agent is None: return f"Unknown agent `{agent_name}`. Available: {', '.join(a.name for a in list_agents())}" draft_path = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.md" chapter_content = draft_path.read_text(encoding="utf-8") chars = [ { "name": c.name, "role": c.role, "voice_notes": c.voice_notes, "comedy_hook": c.comedy_hook or "", } for c in proj.characters ] deaths_data = [ {"character": d.character, "circumstances": d.circumstances, "death_style": d.death_style} for d in proj.death_schedule if d.chapter == chapter_num ] gags_data = [ {"name": g.name, "owner": g.owner, "description": g.description, "escalation_pattern": g.escalation_pattern} for g in proj.running_gags ] prompt = render_agent_prompt( agent=agent, book_name=proj.name, genre=proj.genre, chapter_num=chapter_num, chapter_title=chapter.title, chapter_content=chapter_content, characters=chars, deaths=deaths_data, running_gags=gags_data, ) settings = get_settings() client = get_client() try: text, warnings = complete_user_prompt( client=client, model=settings.anthropic_model, prompt=prompt, max_tokens=8192 ) except anthropic.APIError as e: return f"Anthropic API error ({type(e).__name__}): {e}" report_path = proj.base_path / "reports" / f"chapter-{chapter_num:02d}-{agent_name}-report.md" report_path.write_text(text, encoding="utf-8") if agent_name not in chapter.third_pass_completed: chapter.third_pass_completed.append(agent_name) save_project_and_continuity() warn = "\n".join(warnings) if warnings else "" done = proj.third_pass_requirements_met(chapter) hint = f"\n\nReady to approve: **{done}** (required: {proj.third_agents})" return ( f"{agent.name} report saved.\n{warn}\n`{report_path}`{hint}\n\n" f"When satisfied: approve_chapter({chapter_num}) or request_revision(...)." ) def approve_chapter(chapter_num: int, force: bool = False) -> str: proj, cont = require_project() chapter = proj.get_chapter(chapter_num) if not chapter: return "Chapter not found." if chapter.status == ChapterStatus.APPROVED: return "Already approved." if chapter.status != ChapterStatus.AWAITING_THIRD_PASS: return f"Approve only after editor/third pass (status={chapter.status.value})." if proj.third_agents and not proj.third_pass_requirements_met(chapter) and not force: need = [a for a in proj.third_agents if a not in chapter.third_pass_completed] return ( f"Missing third-pass reports for: {need}. Run run_third_agent for each, " f"or pass force=true if you intentionally skip." ) draft_path = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.md" approved_path = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-approved.md" if draft_path.exists(): draft_path.rename(approved_path) for death in [d for d in proj.death_schedule if d.chapter == chapter_num]: cont.mark_dead(death.character, chapter_num, death.circumstances) chapter.status = ChapterStatus.APPROVED save_project_and_continuity() return f"Chapter {chapter_num} approved → `{approved_path.name}`" def request_revision(chapter_num: int, notes: str) -> str: proj, cont = require_project() chapter = proj.get_chapter(chapter_num) if not chapter: return "Chapter not found." cont.add_revision_note(chapter_num, "editor", notes) append_revision(proj.base_path, chapter_num, notes, agent="editor") chapter.status = ChapterStatus.DRAFT chapter.third_pass_completed = [] chapter.editor_verdict = None save_project_and_continuity() return ( f"Revision requested — status DRAFT. Notes recorded in continuity and " f"`briefs/revision_queue.json`.\n\nNext: run_writer_agent({chapter_num})" ) def get_chapter_status(chapter_num: int) -> str: proj, _ = require_project() chapter = proj.get_chapter(chapter_num) if not chapter: return "Chapter not found." draft = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.md" appr = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-approved.md" return ( f"## Chapter {chapter_num}: {chapter.title}\n" f"- Status: {chapter.status.value}\n" f"- Verdict: {chapter.editor_verdict or '—'}\n" f"- Third passes done: {chapter.third_pass_completed}\n" f"- Draft exists: {draft.exists()}\n" f"- Approved exists: {appr.exists()}\n" ) def list_third_agents() -> str: lines = ["# Third-pass agents\n"] for a in list_agents(): lines.append(f"- **{a.name}**: {a.description}\n") return "".join(lines) def get_living_characters() -> str: _, cont = require_project() alive = [c for c in cont.living_characters if c.status == CharacterStatus.ALIVE] dead = [c for c in cont.living_characters if c.status == CharacterStatus.DEAD] out = [f"Living ({len(alive)}):\n"] for c in alive: out.append(f"- {c.name} items: {', '.join(c.items) if c.items else '—'}\n") if dead: out.append(f"\nDead ({len(dead)}):\n") for c in dead: out.append(f"- ~~{c.name}~~ ch.{c.death_chapter}\n") return "".join(out) def get_inventory() -> str: _, cont = require_project() if not cont.inventory: return "No tracked inventory." return "\n".join(f"- {i.name}: {i.current_holder}" for i in cont.inventory) def get_running_gags() -> str: _, cont = require_project() if not cont.running_gags: return "No gag tracking yet." lines = [] for g in cont.running_gags: lines.append(f"- {g.name}: fired {g.fire_count}x\n") return "".join(lines) def mark_character_dead(character_name: str, chapter: int, cause: str) -> str: _, cont = require_project() cont.mark_dead(character_name, chapter, cause) save_project_and_continuity() return f"{character_name} marked dead." def transfer_item(item_name: str, new_holder: str, chapter: int) -> str: _, cont = require_project() cont.transfer_item(item_name, new_holder, chapter) save_project_and_continuity() return f"{item_name} → {new_holder}" def fire_gag(gag_name: str, chapter: int) -> str: _, cont = require_project() cont.fire_gag(gag_name, chapter) save_project_and_continuity() return f"Recorded gag `{gag_name}` in chapter {chapter}." def add_location(chapter: int, location: str, description: str = "") -> str: _, cont = require_project() cont.locations_visited.append(LocationVisit(chapter=chapter, location=location, description=description)) save_project_and_continuity() return f"Location recorded: {location}" def add_established_fact(fact: str, chapter: int, source: str = "") -> str: _, cont = require_project() cont.established_facts.append(EstablishedFact(fact=fact, chapter=chapter, source=source)) save_project_and_continuity() return "Fact recorded." def export_manuscript(output_filename: str = "EXPORT-manuscript.md") -> str: """Concatenate all manuscript/chapter-*-approved.md files in chapter order.""" proj, _ = require_project() man = proj.base_path / "manuscript" if not man.exists(): return "No manuscript/ directory." def sort_key(p: Path) -> tuple[int, str]: m = re.search(r"chapter-(\d+)-approved\.md$", p.name, re.I) return (int(m.group(1)), p.name) if m else (9999, p.name) paths = sorted(man.glob("chapter-*-approved.md"), key=sort_key) if not paths: return "No chapter-*-approved.md files." parts = [f"<!-- Storywright export v{__version__} -->\n\n"] for p in paths: parts.append(p.read_text(encoding="utf-8", errors="replace").rstrip()) parts.append("\n\n---\n\n") out = proj.base_path / output_filename out.write_text("".join(parts).rstrip() + "\n", encoding="utf-8") return f"Wrote `{out}` — {len(paths)} approved chapter file(s)." def diff_chapter(chapter_num: int, max_lines: int = 150) -> str: """Unified diff: previous draft snapshot (.prev.md) vs current draft.""" proj, _ = require_project() draft = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.md" prev = proj.base_path / "manuscript" / f"chapter-{chapter_num:02d}-draft.prev.md" if not draft.exists(): return "No current draft." if not prev.exists(): return ( "No `chapter-NN-draft.prev.md` — created when **run_writer_agent** overwrites " "an existing draft." ) a = prev.read_text(encoding="utf-8", errors="replace").splitlines() b = draft.read_text(encoding="utf-8", errors="replace").splitlines() diff = list(difflib.unified_diff(a, b, fromfile="draft.prev", tofile="draft", lineterm=""))