kami
ff6a512630
Working tree (including .git) was lost to an rm. Rebuilt by replaying Write/Edit/ Read/attachment events from 25 Claude sessions and 22 successful codex apply_patch blocks into one timestamp-ordered timeline. Verified against ground truth recorded in the transcripts: wc -l on 10 files and ls -l on 5 files at 2026-07-18T13:13:44Z both match exactly; 18 files are byte-identical to their newest ~/.claude/file-history blob. See HANDOFF.md for sources, gaps, and how to rebuild .venv. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
301 lines
16 KiB
Python
301 lines
16 KiB
Python
# worker_script.py — stage 7 narration. FastAPI :8005. calls the warm gemma4 server (:8090).
|
|
import os, re, json
|
|
from fastapi import FastAPI
|
|
from pydantic import BaseModel
|
|
import requests
|
|
import transport
|
|
|
|
app = FastAPI()
|
|
transport.install_logging(app, "script")
|
|
GEMMA4_URL = os.environ.get("GEMMA4_URL", "http://127.0.0.1:8090")
|
|
|
|
|
|
class ScriptInput(BaseModel):
|
|
scene_graph: dict
|
|
chapter_context: str = ""
|
|
panel_id: str = ""
|
|
session_id: str = ""
|
|
names_by_id: dict = {} # character_id -> name, resolved from the final registry (A backfill)
|
|
genders_by_id: dict = {} # character_id -> m|f|unknown, from the registry (F2, for pronouns)
|
|
brief: str = "" # #3: static whole-chapter synopsis (background for consistency/tone)
|
|
recent: list = [] # #1: last few panels' narration text (local flow, no repeats)
|
|
introduced: list = [] # #2: names already narrated -> refer by name/pronoun, don't re-describe
|
|
panel_count: int = 1 # scene-level narration: #panels in this beat -> scales the length budget
|
|
|
|
|
|
class SummaryInput(BaseModel):
|
|
prior: str = "" # the running "story so far" before this batch
|
|
recent: list = [] # narration of the panels since the last refresh
|
|
session_id: str = ""
|
|
|
|
|
|
def _strip_thought(text: str) -> str:
|
|
parts = re.split(r"<\|?channel\|?>", text)
|
|
return parts[-1].strip() if len(parts) > 1 else text
|
|
|
|
|
|
# how each bubble type reads in the narration. narration/sfx have no speaker.
|
|
_VERB = {"speech": "says", "thought": "thinks", "shout": "shouts"}
|
|
|
|
|
|
def _render_line(d, name_by_id):
|
|
typ = d.get("type", "speech")
|
|
txt = d.get("text", "")
|
|
if typ == "narration":
|
|
return f'Caption (scene fact, weave in — do not announce it): "{txt}"'
|
|
if typ == "sfx":
|
|
return f"Sound effect: {txt}"
|
|
who = name_by_id.get(d.get("speaker"), "Someone") if d.get("speaker") else "Someone"
|
|
return f'{who} {_VERB.get(typ, "says")} "{txt}"'
|
|
|
|
|
|
def _name_map(characters, names_by_id=None):
|
|
"""id -> display name. Final registry name first (names_by_id, so a name learned in a LATER
|
|
panel backfills to this one); else the name baked at scene time; else the frozen descriptive
|
|
label ("the one with black hair and glasses"); else a stable positional label ("Person A").
|
|
The model never sees raw DB ids like character_11004ac9."""
|
|
names_by_id = names_by_id or {}
|
|
name_by_id, n = {}, 0
|
|
for c in characters:
|
|
if names_by_id.get(c["id"]) or c.get("name"):
|
|
name_by_id[c["id"]] = names_by_id.get(c["id"]) or c["name"]
|
|
elif c.get("label"):
|
|
name_by_id[c["id"]] = c["label"]
|
|
else:
|
|
name_by_id[c["id"]] = f"Person {chr(65 + n)}"
|
|
n += 1
|
|
return name_by_id
|
|
|
|
|
|
_PRONOUN = {"m": "he", "f": "she"}
|
|
|
|
|
|
def _chars_line(name_by_id, genders_by_id):
|
|
"""render 'Name (he), Other (she)' so the narrator gets pronouns right (F2)."""
|
|
genders_by_id = genders_by_id or {}
|
|
out = []
|
|
for cid, disp in name_by_id.items():
|
|
p = _PRONOUN.get((genders_by_id.get(cid) or "").lower())
|
|
out.append(f"{disp} ({p})" if p else disp)
|
|
return ", ".join(out) or "none"
|
|
|
|
|
|
def build_prompt(sg: dict, chapter_context: str, names_by_id=None,
|
|
brief: str = "", recent=None, introduced=None, genders_by_id=None,
|
|
panel_count: int = 1) -> str:
|
|
name_by_id = _name_map(sg.get("characters", []), names_by_id)
|
|
chars = _chars_line(name_by_id, genders_by_id)
|
|
dialogue = "\n".join(_render_line(d, name_by_id) for d in sg.get("dialogue", [])) or "none"
|
|
ents = ", ".join(e.get("name", "") for e in sg.get("entities", []) if e.get("name"))
|
|
# cast+entities roster (whole-manga memory), kept distinct from the running plot summary below —
|
|
# they used to both render as "Story so far", so the static roster masqueraded as the story.
|
|
ctx = f"Cast & places: {chapter_context}\n\n" if chapter_context else ""
|
|
# #3 running "story so far": only what has happened up to now, so it grounds tone/naming
|
|
# without any risk of narrating ahead (future panels aren't in it yet).
|
|
ov = (f"Story so far (background — what has happened up to now; stay consistent with it, do "
|
|
f"NOT restate it verbatim): {brief}\n\n") if brief else ""
|
|
# #1 local flow: keep continuity with what was just said, don't restate it — and don't reuse its
|
|
# sentence scaffolding (LLMs happily repeat openers/structure even when the facts differ).
|
|
rec = ("Just narrated (continue smoothly from this; do NOT repeat these facts, and do NOT reuse "
|
|
"their sentence structure or opening words — vary the rhythm):\n"
|
|
+ "\n".join(f"- {r}" for r in recent) + "\n\n") if recent else ""
|
|
# camera->pacing: let the director's shot drive sentence rhythm (the biggest "feels AI" tell is
|
|
# narration pacing that ignores the cut). fast/tight shot + hard cut -> punchy; held/drifting shot
|
|
# or soft dissolve -> can breathe.
|
|
cam = (sg.get("camera") or {}).get("effect") or ""
|
|
trans = sg.get("transition") or ""
|
|
pace = (f"Shot pacing — camera: {cam or 'static'}, transition into this panel: {trans or 'cut'}. "
|
|
"Match your line's rhythm to the shot: a hard cut or tight/fast camera wants a short, "
|
|
"punchy sentence; a held or slowly drifting shot or a soft dissolve can run a little "
|
|
"longer and calmer.\n") if (cam or trans) else ""
|
|
# #2 no re-introductions once a character has appeared.
|
|
intro = ("Already introduced — refer to them by name or pronoun, do NOT re-describe their "
|
|
"appearance: " + ", ".join(introduced) + "\n") if introduced else ""
|
|
# scene-level narration: one beat can span several panels, so scale the budget with the beat size
|
|
# (one flowing recap of the whole moment, not a line per panel). solo panel keeps a tight cap.
|
|
if panel_count > 1:
|
|
unit = f"this {panel_count}-panel manga beat (one continuous moment)"
|
|
budget = f"{min(1 + panel_count, 4)} short sentences, max ~{min(20 * panel_count, 60)} words"
|
|
thin = "beat"
|
|
else:
|
|
unit, budget, thin = "this manga panel", "1-2 short sentences, max ~25 words", "panel"
|
|
return (
|
|
ov + rec + ctx + intro +
|
|
f"Write TIGHT recap narration for {unit}: {budget}, "
|
|
"present tense. Tell it like you're recapping to a friend — natural, with momentum. Output "
|
|
"only the narration — no preamble, no notes, no alternatives, never ask for more info. If the "
|
|
f"{thin} has little dialogue or action, one brief scene-setting sentence from what is given.\n"
|
|
"Voice rules (avoid machinery):\n"
|
|
"- Lead with what happens or what's said. When a character speaks, use their ACTUAL words in "
|
|
"a short quote — don't paraphrase into 'he remarks that…' / 'she asks if…'.\n"
|
|
"- Do NOT open with a physical-action gerund ('Looking up,', 'Adjusting his glasses,', "
|
|
"'Leaning forward,'). Vary sentence openings.\n"
|
|
"- Do NOT invent intent or filler ('prepares to…', 'ready to assert…', 'seeking "
|
|
"confirmation'). Only what is actually shown.\n"
|
|
"- Refer to characters by the names below. For anyone WITHOUT a real name, use a SHORT handle "
|
|
"(e.g. 'the guy in glasses') at most ONCE, then 'he'/'she' — never restate their full "
|
|
"appearance from panel to panel.\n"
|
|
"- A thought reads as internal, a shout as raised. Caption/box text is a scene fact — weave "
|
|
"it in; never say 'a narration box' or 'the screen displays'.\n"
|
|
+ (f"Spell these proper nouns exactly: {ents}\n" if ents else "") + "\n"
|
|
+ pace +
|
|
f"Characters present: {chars}\n"
|
|
f"Dialogue:\n{dialogue}\n"
|
|
f"Action: {sg.get('action','')}\n\n"
|
|
"Narration:"
|
|
)
|
|
|
|
|
|
_SYSTEM = ("You are a manga recap narrator. You output only the finished narration paragraph — "
|
|
"never your reasoning, drafts, options, or requests for more information.")
|
|
|
|
|
|
def call_gemma4(prompt: str, system: str = _SYSTEM, max_tokens: int = 160,
|
|
temperature: float = 0.6) -> str:
|
|
payload = {"messages": [{"role": "system", "content": system},
|
|
{"role": "user", "content": prompt}],
|
|
"temperature": temperature, "max_tokens": max_tokens}
|
|
r = requests.post(f"{GEMMA4_URL}/v1/chat/completions", json=payload, timeout=300)
|
|
r.raise_for_status()
|
|
return _strip_thought(r.json()["choices"][0]["message"]["content"])
|
|
|
|
|
|
def build_summary_prompt(prior: str, recent) -> str:
|
|
prior_s = prior.strip() or "(nothing yet — this is the start of the chapter)"
|
|
new = "\n".join(f"- {r}" for r in recent if r) or "- (none)"
|
|
return (
|
|
"You maintain a running 'story so far' summary for a manga recap. Given the previous "
|
|
"summary and the narration of the panels since, produce an UPDATED summary in 3-6 "
|
|
"sentences: compact and factual, consistent names/places, fold in the new events and drop "
|
|
"stale detail. Only what has happened so far — never speculate ahead. Output only the "
|
|
"summary.\n\n"
|
|
f"Previous summary:\n{prior_s}\n\nNewly narrated:\n{new}\n\nUpdated summary:"
|
|
)
|
|
|
|
|
|
@app.post("/script/summary")
|
|
async def summary(data: SummaryInput):
|
|
return {"brief": call_gemma4(build_summary_prompt(data.prior, data.recent))}
|
|
|
|
|
|
@app.post("/script")
|
|
async def script(data: ScriptInput):
|
|
text = call_gemma4(build_prompt(data.scene_graph, data.chapter_context, data.names_by_id,
|
|
data.brief, data.recent, data.introduced, data.genders_by_id,
|
|
data.panel_count))
|
|
return {"panel_id": data.panel_id, "text": text}
|
|
|
|
|
|
# --- roster normalization (spec E2-lite): dedup the accumulated cast+entities in one pass ---
|
|
_NORM_SYSTEM = ("You clean up a manga's cast and entity lists. You output ONLY one JSON object, no "
|
|
"prose, no reasoning.")
|
|
|
|
|
|
class NormalizeInput(BaseModel):
|
|
names: list = [] # canonical character names already in the registry (distinct people)
|
|
entities: list = [] # [{"name","kind"}] accumulated over the chapter, full of near-dups
|
|
session_id: str = ""
|
|
|
|
|
|
def _extract_json(raw: str) -> dict:
|
|
text = _strip_thought(raw)
|
|
m = re.search(r"\{.*\}", text, re.DOTALL)
|
|
if not m:
|
|
raise ValueError(f"no json: {text[:200]}")
|
|
return json.loads(m.group(0))
|
|
|
|
|
|
def build_normalize_prompt(names, entities) -> str:
|
|
ents = "\n".join(f"- {e.get('name','')} ({e.get('kind','')})" for e in entities) or "- (none)"
|
|
nms = ", ".join(n for n in names if n) or "(none)"
|
|
return (
|
|
"Clean up these places/organizations/brands extracted from a manga chapter (may be noisy, "
|
|
"duplicated). Do THREE things:\n"
|
|
"1. MERGE entries that are the same thing into ONE canonical spelling — casing/spacing "
|
|
"variants and truncations (e.g. 'Product Planning' and 'Product Planning Team' -> one).\n"
|
|
"2. KEEP genuinely different things separate — do NOT merge distinct brands (e.g. EGENPICK "
|
|
"and TETOPICK stay two).\n"
|
|
"3. DROP non-entities: comparison phrases ('Teto Pick vs Ege'), fragments, and any entry that "
|
|
"is actually one of the CHARACTERS listed below (a person's name is not a place/org).\n"
|
|
"Also propose casing fixes for the character names (Title Case proper nouns; NEVER merge two "
|
|
"different people — only fix the spelling of the SAME name).\n\n"
|
|
f"Characters (people — for reference; drop entities that are actually these): {nms}\n"
|
|
f"Entities:\n{ents}\n\n"
|
|
'Respond with ONLY this JSON:\n'
|
|
'{"entities":[{"name":"Everyday","kind":"shop"}],"name_fixes":{"CHOI HAESEON":"Choi Haeseon"}}'
|
|
)
|
|
|
|
|
|
@app.post("/normalize")
|
|
async def normalize(data: NormalizeInput):
|
|
prompt = build_normalize_prompt(data.names, data.entities)
|
|
try:
|
|
res = _extract_json(call_gemma4(prompt, system=_NORM_SYSTEM, max_tokens=768, temperature=0.1))
|
|
except (ValueError, json.JSONDecodeError):
|
|
return {"entities": data.entities, "name_fixes": {}} # fail safe: leave the roster unchanged
|
|
res.setdefault("entities", data.entities)
|
|
res.setdefault("name_fixes", {})
|
|
return res
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# self-check: prompt includes character names + quoted dialogue.
|
|
sg = {"characters": [{"id": "c1", "name": "Gojo Satoru"},
|
|
{"id": "character_deadbeef", "label": "the one with white hair"},
|
|
{"id": "character_nolabel"}],
|
|
"dialogue": [{"speaker": "c1", "type": "shout", "text": "stand proud"},
|
|
{"speaker": "character_deadbeef", "type": "speech", "text": "no"},
|
|
{"speaker": "character_nolabel", "type": "speech", "text": "hm"},
|
|
{"speaker": "", "type": "narration", "text": "Years passed."}],
|
|
"entities": [{"name": "Jujutsu High", "kind": "org"}], "action": "approaches"}
|
|
p = build_prompt(sg, "prior scene")
|
|
assert "Gojo Satoru shouts \"stand proud\"" in p
|
|
assert 'the one with white hair says "no"' in p # unnamed + label -> descriptive phrase
|
|
assert 'Person A says "hm"' in p # unnamed, no label -> positional fallback
|
|
assert "character_deadbeef" not in p # raw db id must never reach the model
|
|
# A backfill: a name learned later (registry) overrides the scene-time label for THIS panel
|
|
pb = build_prompt(sg, "prior scene", {"character_deadbeef": "Nanami"})
|
|
assert 'Nanami says "no"' in pb and "the one with white hair" not in pb
|
|
assert 'Caption (scene fact' in p and '"Years passed."' in p
|
|
assert "Jujutsu High" in p and "prior scene" in p
|
|
# scene-level narration: a multi-panel beat scales the budget and talks about a "beat", not a panel.
|
|
assert "this manga panel" in p and "25 words" in p # solo default
|
|
pbeat = build_prompt(sg, "prior scene", panel_count=3)
|
|
assert "3-panel manga beat" in pbeat and "60 words" in pbeat and "beat has little" in pbeat
|
|
# camera->pacing renders only when the graph carries a shot; absent by default.
|
|
assert "Shot pacing" not in p
|
|
pcam = build_prompt({**sg, "camera": {"effect": "dolly_to_subject"}, "transition": "dissolve"}, "s")
|
|
assert "dolly_to_subject" in pcam and "dissolve" in pcam and "Shot pacing" in pcam
|
|
# anti-repetition names structure, not just facts
|
|
pr = build_prompt(sg, "s", recent=["He drew his blade."])
|
|
assert "sentence structure" in pr
|
|
# F2 gender -> pronoun annotation on the characters-present line
|
|
pg = build_prompt(sg, "prior scene", {"c1": "Gojo Satoru"}, genders_by_id={"c1": "m"})
|
|
assert "Gojo Satoru (he)" in pg
|
|
# #1/#2/#3: brief (background), sliding window, introduced-set all render into the prompt
|
|
pc = build_prompt(sg, "prior scene", brief="A duel unfolds.",
|
|
recent=["He drew his blade.", "The crowd fell silent."],
|
|
introduced=["Gojo Satoru"])
|
|
assert "A duel unfolds" in pc and "Story so far" in pc
|
|
assert "- He drew his blade." in pc and "The crowd fell silent." in pc
|
|
assert "Already introduced" in pc and "Gojo Satoru" in pc
|
|
# #3 incremental: prior summary + new narration -> updated-summary prompt
|
|
sp = build_summary_prompt("Hero left the village.", ["He reached the city gate."])
|
|
assert "Hero left the village." in sp and "He reached the city gate." in sp
|
|
assert "Updated summary:" in sp
|
|
sp0 = build_summary_prompt("", []) # cold start, no narration yet
|
|
assert "nothing yet" in sp0
|
|
# normalization: prompt lists the entities + names; parser reads back a canonical roster
|
|
np = build_normalize_prompt(["Lim Seonho"], [{"name": "Product Planning", "kind": "team"},
|
|
{"name": "Product Planning Team", "kind": "org"}])
|
|
assert "Product Planning Team" in np and "Lim Seonho" in np and "MERGE" in np
|
|
nr = _extract_json('{"entities":[{"name":"Everyday","kind":"shop"}],'
|
|
'"name_fixes":{"CHOI HAESEON":"Choi Haeseon"}}')
|
|
assert nr["entities"][0]["name"] == "Everyday" and nr["name_fixes"]["CHOI HAESEON"] == "Choi Haeseon"
|
|
print("worker_script self-check ok")
|