Spaces:
Paused
Paused
| """World authoring pipeline (Section 9). | |
| Generation order matters — author the truth first, derive clues from it: | |
| 1. seed -> concept (best-of-n + judge, originality.py) | |
| 2. concept -> solution.md + timeline.md (the truth + the spine) | |
| 3. timeline -> characters/<name>.md (truth, knowledge boundary, cover, mask) | |
| 4. truth -> environment.md + clue_graph.md (every conviction fact reachable) | |
| 5. accomplice sets -> alibis/<pair>.md (with seams) | |
| Each step asks the author tier for JSON shaped to ``models.py``, which is then | |
| serialized to markdown-with-frontmatter via ``worldio``. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| import uuid | |
| from pathlib import Path | |
| from typing import Any | |
| from ..llm.client import LLMClient | |
| from ..llm.prompts import PromptRegistry | |
| from ..models import utcnow_iso | |
| from ..worldio import write_frontmatter | |
| from .originality import Concept | |
| def slugify(text: str) -> str: | |
| s = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") | |
| return s or "x" | |
| def new_world_id(concept: Concept) -> str: | |
| base = slugify(concept.setting or concept.one_line)[:24] | |
| return f"{base}-{uuid.uuid4().hex[:6]}" | |
| class WorldAuthor: | |
| def __init__(self, client: LLMClient, prompts: PromptRegistry) -> None: | |
| self.client = client | |
| self.prompts = prompts | |
| def author_world( | |
| self, *, world_dir: Path, concept: Concept, seed: str | |
| ) -> dict[str, Any]: | |
| world_dir.mkdir(parents=True, exist_ok=True) | |
| (world_dir / "characters").mkdir(exist_ok=True) | |
| (world_dir / "alibis").mkdir(exist_ok=True) | |
| # 2) solution + timeline | |
| truth = self._gen_json( | |
| "author/world.md.j2", task="world_author", | |
| concept=concept.__dict__, seed=seed, | |
| ) | |
| self._write_solution(world_dir, truth) | |
| self._write_timeline(world_dir, truth) | |
| self._write_world_md(world_dir, concept, truth, seed) | |
| # 3) characters | |
| chars = self._gen_json( | |
| "author/characters.md.j2", task="world_author", | |
| concept=concept.__dict__, | |
| solution=truth.get("solution", {}), | |
| timeline=truth.get("timeline", []), | |
| ) | |
| char_list = chars.get("characters", []) if isinstance(chars, dict) else [] | |
| self._write_characters(world_dir, char_list) | |
| # 4) environment + clues | |
| env = self._gen_json( | |
| "author/environment.md.j2", task="world_author", | |
| concept=concept.__dict__, solution=truth.get("solution", {}), | |
| timeline=truth.get("timeline", []), characters=char_list, | |
| ) | |
| self._write_environment(world_dir, env) | |
| self._write_clue_graph(world_dir, env) | |
| # 5) alibis (only if accomplices exist) | |
| accomplices = [c for c in char_list if c.get("role") == "accomplice"] | |
| if len(accomplices) >= 2: | |
| alibis = self._gen_json( | |
| "author/alibis.md.j2", task="world_author", | |
| accomplices=accomplices, solution=truth.get("solution", {}), | |
| timeline=truth.get("timeline", []), | |
| ) | |
| self._write_alibis(world_dir, alibis) | |
| return { | |
| "solution": truth.get("solution", {}), | |
| "characters": char_list, | |
| "clues": env.get("clues", []), | |
| } | |
| # -- generation helper -------------------------------------------------- | |
| def _gen_json(self, template: str, *, task: str, **vars: Any) -> dict[str, Any]: | |
| prompt = self.prompts.render(template, **vars) | |
| data, _ = self.client.complete_json(tier="author", task=task, user=prompt) | |
| return data if isinstance(data, dict) else {} | |
| # -- writers ------------------------------------------------------------ | |
| def _write_solution(self, d: Path, truth: dict[str, Any]) -> None: | |
| sol = truth.get("solution", {}) | |
| fm = { | |
| "culprit": sol.get("culprit", ""), | |
| "means": sol.get("means", ""), | |
| "motive": sol.get("motive", ""), | |
| "opportunity": sol.get("opportunity", ""), | |
| "true_sequence": sol.get("true_sequence", ""), | |
| } | |
| body = sol.get("body", "Ground truth of the case. Engine-only.") | |
| (d / "solution.md").write_text(write_frontmatter(fm, body), "utf-8") | |
| def _write_timeline(self, d: Path, truth: dict[str, Any]) -> None: | |
| slices = truth.get("timeline", []) | |
| clean = [ | |
| { | |
| "time_slice": s.get("time_slice", ""), | |
| "character": s.get("character", ""), | |
| "location": s.get("location", ""), | |
| "action": s.get("action", ""), | |
| "observed_by": s.get("observed_by", []), | |
| } | |
| for s in slices if isinstance(s, dict) | |
| ] | |
| body = "Structured movement schedule — the spine of the case." | |
| (d / "timeline.md").write_text( | |
| write_frontmatter({"slices": clean}, body), "utf-8" | |
| ) | |
| def _write_world_md( | |
| self, d: Path, concept: Concept, truth: dict[str, Any], seed: str | |
| ) -> None: | |
| fm = { | |
| "title": concept.one_line[:80], | |
| "seed": seed, | |
| "one_line": concept.one_line, | |
| "setting": concept.setting, | |
| "twist_tag": concept.twist_tag, | |
| "created": utcnow_iso(), | |
| "play_count": 0, | |
| } | |
| body = truth.get("world", concept.premise) | |
| (d / "world.md").write_text(write_frontmatter(fm, body), "utf-8") | |
| def _write_characters(self, d: Path, chars: list[dict[str, Any]]) -> None: | |
| for c in chars: | |
| name = c.get("name", "") | |
| if not name: | |
| continue | |
| fm = { | |
| "name": name, | |
| "role": c.get("role", "suspect"), | |
| "guilty": bool(c.get("guilty", False)), | |
| "truth": c.get("truth", ""), | |
| "knows": c.get("knows", { | |
| "witnessed": [], "topics_known": [], "topics_unknowable": [] | |
| }), | |
| "cover": c.get("cover", ""), | |
| "never_admit": c.get("never_admit", []), | |
| "cracks_when": c.get("cracks_when", ""), | |
| "crack_behavior": c.get("crack_behavior", "deflect"), | |
| "tells": c.get("tells", []), | |
| "secret_kind": c.get("secret_kind", "guilty"), | |
| "exoneration": c.get("exoneration"), | |
| } | |
| body = c.get("voice", "Speaks plainly.") | |
| (d / "characters" / f"{slugify(name)}.md").write_text( | |
| write_frontmatter(fm, body), "utf-8" | |
| ) | |
| def _write_environment(self, d: Path, env: dict[str, Any]) -> None: | |
| objs = env.get("objects", []) | |
| clean = [ | |
| { | |
| "id": o.get("id", slugify(o.get("description_true", "obj"))[:24]), | |
| "location": o.get("location", "scene"), | |
| "description_true": o.get("description_true", ""), | |
| "evidential": bool(o.get("evidential", False)), | |
| "clue": o.get("clue"), | |
| "visible_by_default": bool(o.get("visible_by_default", True)), | |
| } | |
| for o in objs if isinstance(o, dict) | |
| ] | |
| body = "Rooms, objects, hidden details — each with true state." | |
| (d / "environment.md").write_text( | |
| write_frontmatter({"objects": clean}, body), "utf-8" | |
| ) | |
| def _write_clue_graph(self, d: Path, env: dict[str, Any]) -> None: | |
| nodes = env.get("clues", []) | |
| clean = [ | |
| { | |
| "id": n.get("id", ""), | |
| "reveals": n.get("reveals", ""), | |
| "sources": n.get("sources", []), | |
| "unlocks": n.get("unlocks", []), | |
| "exonerates": n.get("exonerates", []), | |
| "required_for_solution": bool(n.get("required_for_solution", False)), | |
| "requires": n.get("requires", []), | |
| } | |
| for n in nodes if isinstance(n, dict) and n.get("id") | |
| ] | |
| body = "Clue nodes + unlock dependencies + exonerations." | |
| (d / "clue_graph.md").write_text( | |
| write_frontmatter({"nodes": clean}, body), "utf-8" | |
| ) | |
| def _write_alibis(self, d: Path, alibis: dict[str, Any]) -> None: | |
| for a in alibis.get("alibis", []): | |
| if not isinstance(a, dict): | |
| continue | |
| aid = a.get("id", slugify("-".join(a.get("characters", ["pair"])))) | |
| fm = { | |
| "id": aid, | |
| "characters": a.get("characters", []), | |
| "agreed_facts": a.get("agreed_facts", ""), | |
| "agreed_timeline": a.get("agreed_timeline", ""), | |
| "cover_per_character": a.get("cover_per_character", {}), | |
| "seams": a.get("seams", []), | |
| } | |
| body = a.get("body", "Rehearsed shared cover story.") | |
| (d / "alibis" / f"{slugify(aid)}.md").write_text( | |
| write_frontmatter(fm, body), "utf-8" | |
| ) | |