"""COMPOSE_SCENE -- Star's VN scene director tool.
Star builds a JSON scene script specifying characters, backgrounds,
dialogue, stage directions, sound effects, and transitions. The script
is stored in Redis and published as a scene_event for the web client
to play back as full VN-style cutscenes.
@fire @skull THE DIRECTOR SPEAKS. THE STAGE OBEYS.
"""
from __future__ import annotations
import json
import logging
import time
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "compose_scene"
TOOL_DESCRIPTION = (
"Build and publish a VN scene script for the web client's scene engine. "
"The scene plays as a fullscreen cinematic overlay with character sprites, "
"dialogue boxes, background transitions, sound effects, and music.\n\n"
"Scene scripts use a step-based format. Step types:\n"
" CORE: bg, enter, exit, dialogue, move, wait, end\n"
" MEDIA: sfx, music, stop_music, ambience\n"
" EFFECTS: shake, flash, filter, fade\n"
" DANGANRONPA: bounce (drop+spring), slam (fast+impact), pulse (heartbeat), wobble (oscillate)\n"
" ACE ATTORNEY: overlay_text (OBJECTION!), emote (! ? ... above char), speed_lines\n"
" CAMERA: zoom, pan\n\n"
"PATH SHORTCUTS (auto-resolved):\n"
" sfx:filename.mp3 -> /assets/sound_effects/filename.mp3\n"
" bgm:filename.mp3 -> /assets/bgm/filename.mp3\n"
" ambience:filename.mp3 -> /assets/sound_effects/ambience_loops/filename.mp3\n"
" bg:filename.png -> /backgrounds/filename.png\n\n"
"Example:\n"
" compose_scene(title='Class Is In Session', characters=['stargazer','orion'], "
"script=[{'type':'bounce','char':'stargazer','position':85}, "
"{'type':'slam','char':'orion','position':15,'direction':'left'}, "
"{'type':'overlay_text','text':'OBJECTION!','style':'objection'}, "
"{'type':'dialogue','char':'stargazer','text':'Class is in session.','expression':'laugh'}])"
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Scene title (displayed on the title card before playback).",
},
"characters": {
"type": "array",
"items": {"type": "string"},
"description": (
"List of character IDs in this scene " "(e.g. ['stargazer', 'orion'])."
),
},
"background": {
"type": "string",
"description": (
"Background image filename (from /backgrounds/) "
"or full URL. Optional."
),
},
"music": {
"type": "string",
"description": "Background music track URL. Optional.",
},
"script": {
"type": "array",
"items": {"type": "object"},
"description": (
"Array of scene steps. Each step is an object with 'type' "
"and type-specific fields. See tool description for step types."
),
},
},
"required": ["characters", "script"],
}
# Redis keys
SCENE_KEY_PREFIX = "star:scenes:"
PENDING_SCENE_KEY = "star:pending_scene"
[docs]
async def run(
title: str = "Untitled Scene",
characters: list | None = None,
background: str | None = None,
music: str | None = None,
script: list | None = None,
ctx: "ToolContext | None" = None,
) -> str:
"""Build, validate, persist, and broadcast a VN scene script.
Entry point for the ``compose_scene`` tool, which lets Star direct fullscreen
visual-novel cutscenes on the web client. It validates that every script step
has a known ``type`` (against the in-function ``valid_types`` set), rewrites the
asset path shortcuts (``sfx:``/``bgm:``/``ambience:``/``bg:``) on the
``sound``/``track``/``loop``/``image`` fields to their served paths, ensures the
script terminates with an ``end`` step, and stamps the scene with owner
platform/channel/user ids for the web UI's access control. It then writes the
scene to Redis under ``star:scenes:{scene_id}``, marks it pending at
``star:pending_scene``, and publishes it on the ``star:scene:play`` channel for
immediate pickup. As a bonus, any ``dialogue`` step flagged ``as_ghost`` for a
non-Star character is delivered as a Matrix ghost message via
``egregore_bridge.get_bridge``.
Dispatched by ``tool_loader.py`` as the ``compose_scene`` handler (located via
``getattr(module, "run")``); not called directly elsewhere. Side effects are the
three Redis writes/publish on ``ctx.redis`` and the optional Matrix ghost-message
sends.
Args:
title (str): Scene title shown on the title card before playback.
characters (list | None): Character ids appearing in the scene (required).
background (str | None): Optional background filename or URL.
music (str | None): Optional background-music track URL.
script (list | None): Ordered list of step dicts, each with a ``type`` and
type-specific fields (required).
ctx (ToolContext | None): Tool execution context supplying ``redis`` plus the
owner ``platform``/``channel_id``/``user_id``.
Returns:
str: A JSON object with ``success``, the ``scene_id``, the step count, and
the ghost-message count on success, or an ``{"success": false, "error": ...}``
object on a missing context/Redis, missing required inputs, an invalid step,
or a publish failure.
"""
if ctx is None:
return json.dumps({"success": False, "error": "No tool context available."})
redis = getattr(ctx, "redis", None)
if redis is None:
return json.dumps({"success": False, "error": "Redis not available."})
if not characters or not script:
return json.dumps(
{
"success": False,
"error": "characters and script are required.",
}
)
# Validate script steps have 'type'
valid_types = {
# Core
"bg",
"enter",
"exit",
"dialogue",
"move",
"wait",
"end",
# Media
"sfx",
"music",
"stop_music",
"ambience",
# Screen effects
"shake",
"flash",
"filter",
"fade",
# Danganronpa specials
"bounce",
"slam",
"pulse",
"wobble",
# Ace Attorney specials
"overlay_text",
"emote",
"speed_lines",
# Camera
"zoom",
"pan",
}
for i, step in enumerate(script):
if not isinstance(step, dict) or "type" not in step:
return json.dumps(
{
"success": False,
"error": f"Step {i} missing 'type' field.",
}
)
if step["type"] not in valid_types:
return json.dumps(
{
"success": False,
"error": f"Step {i} has unknown type '{step['type']}'. "
f"Valid: {sorted(valid_types)}",
}
)
# Resolve asset path shortcuts in script steps
_PATH_PREFIXES = {
"sound": "sfx:",
"track": "bgm:",
"loop": "ambience:",
"image": "bg:",
}
_PATH_MAPS = {
"sfx:": "/assets/sound_effects/",
"bgm:": "/assets/bgm/",
"ambience:": "/assets/sound_effects/ambience_loops/",
"bg:": "/backgrounds/",
}
for step in script:
for field in ("sound", "track", "loop", "image"):
val = step.get(field)
if val and isinstance(val, str):
for prefix, resolved in _PATH_MAPS.items():
if val.startswith(prefix):
step[field] = resolved + val[len(prefix) :]
break
# Ensure script ends with 'end'
if not script or script[-1].get("type") != "end":
script.append({"type": "end"})
# Build scene object
scene_id = f"scene_{int(time.time())}_{hash(title) & 0xFFFF:04x}"
scene = {
"scene_id": scene_id,
"title": title,
"characters": characters,
"background": background,
"music": music,
"script": script,
"created_at": int(time.time()),
# Web UI access control (see web/scenes_api.py)
"owner_platform": getattr(ctx, "platform", "") or "",
"owner_channel_id": getattr(ctx, "channel_id", "") or "",
"owner_user_id": getattr(ctx, "user_id", "") or "",
}
try:
# Store scene persistently
await redis.set(
f"{SCENE_KEY_PREFIX}{scene_id}",
json.dumps(scene),
)
# Set as pending scene for next exhale
await redis.set(PENDING_SCENE_KEY, json.dumps(scene))
# Publish scene event for immediate pickup
await redis.publish("star:scene:play", json.dumps(scene))
# Send dialogue steps as Matrix ghost messages if flagged
ghost_messages_sent = 0
try:
from egregore_bridge import get_bridge
bridge = get_bridge()
room_id = getattr(ctx, "channel_id", None) if ctx else None
if room_id:
for step in script:
if (
step.get("type") == "dialogue"
and step.get("as_ghost")
and step.get("char")
and step["char"] != "stargazer"
):
char_name = step["char"]
text = step.get("text", "")
if text:
event_id = await bridge.send_message(
char_name,
room_id,
text,
)
if event_id:
ghost_messages_sent += 1
except Exception as e:
logger.warning("Ghost message sending failed: %s", e)
logger.info(
"Scene composed: %s (%d chars, %d steps, %d ghost msgs)",
scene_id,
len(characters),
len(script),
ghost_messages_sent,
)
return json.dumps(
{
"success": True,
"message": (
f"Scene '{title}' published. The web client will show "
f"a [Play Scene] button for the user to begin playback."
),
"scene_id": scene_id,
"characters": characters,
"steps": len(script),
"title": title,
"ghost_messages_sent": ghost_messages_sent,
},
indent=2,
)
except Exception as e:
logger.error("compose_scene error: %s", e, exc_info=True)
return json.dumps(
{
"success": False,
"error": f"Failed to publish scene: {e}",
}
)