"""Reset the Lyria music generation context for a hard transition."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "reset_music_context"
TOOL_DESCRIPTION = (
"Reset the Lyria music generation context. Causes a hard transition "
"without stopping the stream. Useful after major config changes (e.g. "
"BPM or scale). Requires active music session."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"guild_id": {"type": "string", "description": "Discord server (guild) ID."},
},
"required": ["guild_id"],
}
[docs]
async def run(
guild_id: str,
ctx: "ToolContext | None" = None,
) -> str:
"""Request a context reset for the active Lyria music stream.
Entry point for the ``reset_music_context`` tool. It forces a hard
transition in the bot's running Lyria generation so musical changes (BPM,
scale, prompts) take effect cleanly, without tearing down the audio stream
itself. This is useful right after a major config change where you want the
model to discard its prior musical context rather than smoothly morph into
the new settings.
It resolves the Discord client from the context via
:func:`tools._discord_helpers.get_discord_client` (returning that helper's
error string verbatim if the client is unavailable), reaches the
``lyria_service`` hanging off ``ctx.adapter`` (the ProxyPlatformAdapter on
the inference worker), and delegates to its ``reset_context`` coroutine
keyed by the parsed integer guild ID. No Redis, KG, or HTTP access happens
here; the side effect is the in-flight reset of the per-guild Lyria session.
Dispatched by the tool registry (:meth:`tools.ToolRegistry.call`) after the
loader picks up this module's ``run`` via its ``TOOL_NAME`` /
``TOOL_DESCRIPTION`` metadata; no direct in-code callers were found.
Args:
guild_id: Discord server (guild) ID as a string; it is parsed to an
integer to look up the active session.
ctx: Tool execution context. Must carry a non-None ``adapter`` exposing
``lyria_service``.
Returns:
str: The status message returned by ``lyria.reset_context``, or an
error string when the adapter or Discord client is missing or
``guild_id`` is not a valid integer.
"""
if ctx is None or not hasattr(ctx, "adapter") or ctx.adapter is None:
return "Error: Discord adapter (ctx.adapter) is required."
from tools._discord_helpers import get_discord_client
client = get_discord_client(ctx)
if isinstance(client, str):
return client
lyria = ctx.adapter.lyria_service
try:
guild_id_int = int(guild_id)
except (ValueError, TypeError):
return "Error: Invalid guild_id."
return await lyria.reset_context(guild_id_int)