diff --git a/src/talemate/agents/base.py b/src/talemate/agents/base.py index 6bc4dd97..9d80604c 100644 --- a/src/talemate/agents/base.py +++ b/src/talemate/agents/base.py @@ -612,7 +612,9 @@ class Agent(ABC): """Return True if the named action is a dynamic-children registry.""" actions = getattr(self, "actions", None) or {} action = actions.get(action_key) - return bool(action and action.config and DYNAMIC_CHILDREN_FIELD in action.config) + return bool( + action and action.config and DYNAMIC_CHILDREN_FIELD in action.config + ) def dynamic_registry_keys(self) -> list[str]: actions = getattr(self, "actions", None) or {} @@ -693,9 +695,7 @@ class Agent(ABC): # persisting external state to disk), override # ``persist_dynamic_external_state`` instead — it's awaited by the # websocket plugin after the sync hook fires. - def on_dynamic_child_added( - self, registry_key: str, slug: str, label: str - ) -> None: + def on_dynamic_child_added(self, registry_key: str, slug: str, label: str) -> None: return None def on_dynamic_child_removed(self, registry_key: str, slug: str) -> None: @@ -730,9 +730,7 @@ class Agent(ABC): return set() # Mutation helpers. - def register_dynamic_child( - self, registry_key: str, slug: str, label: str - ) -> None: + def register_dynamic_child(self, registry_key: str, slug: str, label: str) -> None: if not self.is_dynamic_registry(registry_key): raise ValueError(f"'{registry_key}' is not a dynamic registry") if not slug: @@ -776,9 +774,7 @@ class Agent(ABC): self.on_dynamic_child_renamed(registry_key, slug, label or slug) # Resolver helpers — used by agents that dispatch via name-prefixed methods. - def dynamic_attr( - self, registry_key: str, slug: str, name: str, default=None - ): + def dynamic_attr(self, registry_key: str, slug: str, name: str, default=None): """Resolve a property-like helper for a dynamic child. Looks up ``self.__`` and invokes it with the @@ -789,9 +785,7 @@ class Agent(ABC): return default return fn(slug) - def dynamic_method( - self, registry_key: str, slug: str, name: str, default=None - ): + def dynamic_method(self, registry_key: str, slug: str, name: str, default=None): """Resolve a callable helper for a dynamic child. Returns a partial that pre-binds the slug, so callers can invoke it diff --git a/src/talemate/agents/tts/__init__.py b/src/talemate/agents/tts/__init__.py index edd1bdf1..db220e2b 100644 --- a/src/talemate/agents/tts/__init__.py +++ b/src/talemate/agents/tts/__init__.py @@ -51,7 +51,10 @@ from .chatterbox import ChatterboxMixin from .websocket_handler import TTSWebsocketHandler from .f5tts import F5TTSMixin from .pocket_tts import PocketTTSMixin -from .openai_compatible import OpenAICompatibleMixin, REGISTRY_KEY as OPENAI_COMPAT_REGISTRY_KEY # noqa: F811 +from .openai_compatible import ( + OpenAICompatibleMixin, + REGISTRY_KEY as OPENAI_COMPAT_REGISTRY_KEY, +) # noqa: F811 from .audio_tags import AudioTagsMixin from .util import parse_chunks, rejoin_chunks @@ -313,8 +316,7 @@ class TTSAgent( choices = [ c for c in choices - if c.get("value") not in backend_entries - and not c.get("_dynamic_backend") + if c.get("value") not in backend_entries and not c.get("_dynamic_backend") ] for slug, label in backend_entries.items(): choices.append( @@ -327,9 +329,7 @@ class TTSAgent( ) self.actions["_config"].config["apis"].choices = choices - def on_dynamic_child_added( - self, registry_key: str, slug: str, label: str - ) -> None: + def on_dynamic_child_added(self, registry_key: str, slug: str, label: str) -> None: if registry_key != OPENAI_COMPAT_REGISTRY_KEY: return self._refresh_apis_choices() diff --git a/src/talemate/agents/tts/openai_compatible.py b/src/talemate/agents/tts/openai_compatible.py index 3d11c11d..8d109cd0 100644 --- a/src/talemate/agents/tts/openai_compatible.py +++ b/src/talemate/agents/tts/openai_compatible.py @@ -244,8 +244,7 @@ class OpenAICompatibleMixin: icon="mdi-link", value="API URL not set", description=( - f"Set the base URL of the '{backend}' OpenAI-compatible " - "TTS server." + f"Set the base URL of the '{backend}' OpenAI-compatible TTS server." ), color="error", ).model_dump() @@ -300,9 +299,7 @@ class OpenAICompatibleMixin: headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} custom = self._openai_compatible_voices_endpoint(backend) - candidates = ( - [custom] if custom else list(DEFAULT_VOICE_LISTING_PATHS) - ) + candidates = [custom] if custom else list(DEFAULT_VOICE_LISTING_PATHS) attempted: list[str] = [] async with httpx.AsyncClient(timeout=10.0) as http: diff --git a/src/talemate/client/koboldcpp.py b/src/talemate/client/koboldcpp.py index 7ccb2257..16779a15 100644 --- a/src/talemate/client/koboldcpp.py +++ b/src/talemate/client/koboldcpp.py @@ -590,9 +590,7 @@ class KoboldCppClient(OpenAIVisionMixin, ClientBase): log.error("TTS setup task failed", exc=exc) return False - async def _tts_openai_compatible_setup_impl( - self, tts_agent: "TTSAgent" - ) -> bool: + async def _tts_openai_compatible_setup_impl(self, tts_agent: "TTSAgent") -> bool: """Internal: keep the TTS agent's enabled-apis state in sync with what this KoboldCpp instance currently has loaded. @@ -636,9 +634,7 @@ class KoboldCppClient(OpenAIVisionMixin, ClientBase): if voices_loaded is None: return False - apis_value = list( - tts_agent.actions["_config"].config["apis"].value or [] - ) + apis_value = list(tts_agent.actions["_config"].config["apis"].value or []) if voices_loaded: if existing_slug is None: @@ -647,11 +643,7 @@ class KoboldCppClient(OpenAIVisionMixin, ClientBase): reserved = tts_agent.reserved_slugs_for_registry( OPENAI_COMPAT_REGISTRY_KEY ) - taken = set( - tts_agent.dynamic_child_slugs( - OPENAI_COMPAT_REGISTRY_KEY - ) - ) + taken = set(tts_agent.dynamic_child_slugs(OPENAI_COMPAT_REGISTRY_KEY)) slug, n = base_slug, 2 while slug in taken or slug in reserved: slug = f"{base_slug}-{n}" @@ -669,9 +661,7 @@ class KoboldCppClient(OpenAIVisionMixin, ClientBase): tts_agent.actions[slug].config["api_url"].value = target_url if slug not in apis_value: apis_value.append(slug) - tts_agent.actions["_config"].config[ - "apis" - ].value = apis_value + tts_agent.actions["_config"].config["apis"].value = apis_value # Best-effort voice refresh; failure doesn't undo setup. # Bounded so a hung server can't pin the periodic status # loop that calls into setup_check. diff --git a/tests/test_dynamic_action_registry.py b/tests/test_dynamic_action_registry.py index 3708839e..4aad33b8 100644 --- a/tests/test_dynamic_action_registry.py +++ b/tests/test_dynamic_action_registry.py @@ -106,9 +106,7 @@ class _DynamicTestAgent(Agent): return f"{slug}-{n * 2}" # Lifecycle spy hooks - def on_dynamic_child_added( - self, registry_key: str, slug: str, label: str - ) -> None: + def on_dynamic_child_added(self, registry_key: str, slug: str, label: str) -> None: self.added.append((slug, label)) def on_dynamic_child_removed(self, registry_key: str, slug: str) -> None: @@ -189,9 +187,7 @@ class TestDynamicChildrenEntries: assert agent.dynamic_child_slugs(REGISTRY_KEY) == [] def test_garbage_json_returns_empty_with_warning(self, agent, caplog): - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = "{not json" + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = "{not json" # Should not raise — just return [] and log a warning. assert agent.dynamic_children_entries(REGISTRY_KEY) == [] assert agent.dynamic_child_slugs(REGISTRY_KEY) == [] @@ -201,16 +197,14 @@ class TestDynamicChildrenEntries: {"slug": "alpha", "label": "Alpha"}, {"slug": "beta", "label": "Beta"}, ] - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = json.dumps(entries) + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = json.dumps( + entries + ) assert agent.dynamic_children_entries(REGISTRY_KEY) == entries assert agent.dynamic_child_slugs(REGISTRY_KEY) == ["alpha", "beta"] def test_entries_without_slug_filtered_out(self, agent): - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = json.dumps( + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = json.dumps( [ {"slug": "alpha", "label": "Alpha"}, {"label": "Missing slug"}, @@ -234,9 +228,7 @@ class TestDynamicChildrenEntries: class TestInstallDynamicChildren: def test_synthesizes_one_action_per_entry(self, agent): - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = json.dumps( + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = json.dumps( [ {"slug": "alpha", "label": "Alpha"}, {"slug": "beta", "label": "Beta"}, @@ -249,16 +241,16 @@ class TestInstallDynamicChildren: assert agent.actions["beta"].label == "Beta" def test_installed_children_carry_parent_key(self, agent): - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = json.dumps([{"slug": "alpha", "label": "Alpha"}]) + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = json.dumps( + [{"slug": "alpha", "label": "Alpha"}] + ) agent.install_dynamic_children(REGISTRY_KEY) assert agent.actions["alpha"].parent_key == REGISTRY_KEY def test_idempotent_second_call_is_a_no_op(self, agent): - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = json.dumps([{"slug": "alpha", "label": "Alpha"}]) + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = json.dumps( + [{"slug": "alpha", "label": "Alpha"}] + ) agent.install_dynamic_children(REGISTRY_KEY) first = agent.actions["alpha"] # Mutate a child config value and re-install — value must survive, @@ -269,9 +261,7 @@ class TestInstallDynamicChildren: assert agent.actions["alpha"].config["value"].value == "user-edited" def test_stale_entries_pruned_when_blob_shrinks(self, agent): - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = json.dumps( + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = json.dumps( [ {"slug": "alpha", "label": "Alpha"}, {"slug": "beta", "label": "Beta"}, @@ -281,18 +271,18 @@ class TestInstallDynamicChildren: assert "beta" in agent.actions # Remove "beta" from the blob and re-install — it should be dropped. - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = json.dumps([{"slug": "alpha", "label": "Alpha"}]) + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = json.dumps( + [{"slug": "alpha", "label": "Alpha"}] + ) agent.install_dynamic_children(REGISTRY_KEY) assert "alpha" in agent.actions assert "beta" not in agent.actions def test_does_not_remove_unrelated_actions(self, agent): # Static action exists; install should not touch it. - agent.actions[REGISTRY_KEY].config[ - DYNAMIC_CHILDREN_FIELD - ].value = json.dumps([{"slug": "alpha", "label": "Alpha"}]) + agent.actions[REGISTRY_KEY].config[DYNAMIC_CHILDREN_FIELD].value = json.dumps( + [{"slug": "alpha", "label": "Alpha"}] + ) agent.install_dynamic_children(REGISTRY_KEY) assert "static_action" in agent.actions @@ -344,9 +334,7 @@ class TestRegisterDynamicChild: def test_rejects_reserved_slug(self, agent, monkeypatch): # Patch reserved_slugs_for_registry on this instance to claim "alpha". - monkeypatch.setattr( - agent, "reserved_slugs_for_registry", lambda key: {"alpha"} - ) + monkeypatch.setattr(agent, "reserved_slugs_for_registry", lambda key: {"alpha"}) with pytest.raises(ValueError, match="reserved"): agent.register_dynamic_child(REGISTRY_KEY, "alpha", "Alpha") @@ -509,9 +497,7 @@ class TestApplyConfigPrePass: "enabled": True, "config": { DYNAMIC_CHILDREN_FIELD: { - "value": json.dumps( - [{"slug": "alpha", "label": "Alpha"}] - ), + "value": json.dumps([{"slug": "alpha", "label": "Alpha"}]), }, }, }, @@ -524,10 +510,7 @@ class TestApplyConfigPrePass: } } asyncio.run(agent.apply_config(**saved_kwargs)) - assert ( - agent.actions["alpha"].config["value"].value - == "restored-from-disk" - ) + assert agent.actions["alpha"].config["value"].value == "restored-from-disk" def test_blob_value_restored_before_synthesis(self, agent): saved_kwargs = { diff --git a/tests/test_tts_agent.py b/tests/test_tts_agent.py index f0b3f7ed..4f71cd89 100644 --- a/tests/test_tts_agent.py +++ b/tests/test_tts_agent.py @@ -15,7 +15,7 @@ import json import os import types from pathlib import Path -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock import pydantic import pytest @@ -23,24 +23,19 @@ import pytest_asyncio import talemate.agents.tts.voice_library as voice_library import talemate.instance as instance -import talemate.util.dialogue as dialogue_utils from talemate.agents.tts import TTSAgent from talemate.agents.tts.openai_compatible import ( - OpenAICompatibleMixin, _parse_voices_payload, _resolve_voices_url, ) from talemate.agents.tts.schema import ( - APIStatus, Chunk, - GenerationContext, MAX_TAG_LENGTH, MAX_TAGS_PER_VOICE, Voice, VoiceLibrary, ) from talemate.agents.tts.voice_library import ( - VoiceLibrary as VoiceLibrarySchema, _apply_default_voice_migration, load_scene_voice_library, save_scene_voice_library, @@ -49,10 +44,9 @@ from talemate.agents.tts.voice_library import ( ) from talemate.scene_message import ( CharacterMessage, - ContextInvestigationMessage, NarratorMessage, ) -from talemate.ux.schema import Action, Note +from talemate.ux.schema import Action # --------------------------------------------------------------------------- @@ -66,9 +60,7 @@ def _make_voice( provider_id: str = "vid", **kwargs, ) -> Voice: - return Voice( - label=label, provider=provider, provider_id=provider_id, **kwargs - ) + return Voice(label=label, provider=provider, provider_id=provider_id, **kwargs) @pytest.fixture @@ -125,9 +117,7 @@ class FakeScene: class TestVoiceSchema: def test_id_is_provider_colon_provider_id(self): - voice = Voice( - label="Adam", provider="kokoro", provider_id="am_adam" - ) + voice = Voice(label="Adam", provider="kokoro", provider_id="am_adam") assert voice.id == "kokoro:am_adam" def test_id_changes_when_provider_changes(self): @@ -198,9 +188,7 @@ class TestChunk: assert "and laughed, loudly" in cleaned def test_cleaned_text_collapses_whitespace(self): - chunk = Chunk( - text=["Hello world\t\twith\nmany spaces"], type="exposition" - ) + chunk = Chunk(text=["Hello world\t\twith\nmany spaces"], type="exposition") cleaned = chunk.cleaned_text # only single spaces between words assert " " not in cleaned @@ -273,9 +261,7 @@ class TestVoiceLibraryMigration: "openai:alloy": Voice( label="Alloy", provider="openai", provider_id="alloy" ), - "openai:echo": Voice( - label="Echo", provider="openai", provider_id="echo" - ), + "openai:echo": Voice(label="Echo", provider="openai", provider_id="echo"), } monkeypatch.setattr(voice_library, "DEFAULT_VOICES", defaults) @@ -291,15 +277,11 @@ class TestVoiceLibraryMigration: "openai:alloy": Voice( label="Alloy", provider="openai", provider_id="alloy" ), - "openai:echo": Voice( - label="Echo", provider="openai", provider_id="echo" - ), + "openai:echo": Voice(label="Echo", provider="openai", provider_id="echo"), } monkeypatch.setattr(voice_library, "DEFAULT_VOICES", defaults) - existing = Voice( - label="Custom", provider="openai", provider_id="custom" - ) + existing = Voice(label="Custom", provider="openai", provider_id="custom") library = VoiceLibrary(voices={existing.id: existing}) changed = _apply_default_voice_migration(library) @@ -320,9 +302,7 @@ class TestVoiceLibraryMigration: monkeypatch.setattr(voice_library, "DEFAULT_VOICES", defaults) # User has an openai voice but no kokoro voices - existing = Voice( - label="Custom", provider="openai", provider_id="custom" - ) + existing = Voice(label="Custom", provider="openai", provider_id="custom") library = VoiceLibrary(voices={existing.id: existing}) changed = _apply_default_voice_migration(library) @@ -448,9 +428,7 @@ class TestSceneVoiceLibrary: # info_dir does not exist yet scene = FakeScene(info_dir=str(tmp_path / "info")) v = Voice(label="A", provider="kokoro", provider_id="a") - await save_scene_voice_library( - scene, VoiceLibrary(voices={v.id: v}) - ) + await save_scene_voice_library(scene, VoiceLibrary(voices={v.id: v})) assert (Path(scene.info_dir) / "voice-library.json").exists() @@ -471,9 +449,7 @@ class TestTTSAgentApiHelpers: assert tts_agent.api_configured("kokoro") is True # Dynamic backend: configured iff its api_url is set on the child action. - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") tts_agent.actions["test-be"].config["api_url"].value = "" assert tts_agent.api_configured("test-be") is False tts_agent.actions["test-be"].config[ @@ -486,9 +462,7 @@ class TestTTSAgentApiHelpers: assert tts_agent.api_configured("does_not_exist") is True def test_api_ready_requires_enabled_and_configured(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") # Registering injected the backend slug into apis.choices as a # dynamic-marked dict — flip the apis-enabled list to enable it. choices = tts_agent.actions["_config"].config["apis"].choices @@ -518,9 +492,7 @@ class TestTTSAgentApiHelpers: def test_ready_apis_only_includes_ready(self, tts_agent): # enable kokoro (always configured) and a dynamic backend (only if # api_url set) - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") tts_agent.actions["_config"].config["apis"].value = [ "kokoro", "test-be", @@ -563,17 +535,13 @@ class TestTTSAgentApiUsed: def test_returns_false_when_no_match(self, tts_agent, fresh_voice_library): # narrator not configured to a real voice in library - tts_agent.actions["_config"].config[ - "narrator_voice_id" - ].value = "nope:nope" + tts_agent.actions["_config"].config["narrator_voice_id"].value = "nope:nope" tts_agent.scene = FakeScene(characters=[]) assert tts_agent.api_used("kokoro") is False def test_returns_false_when_no_scene(self, tts_agent, fresh_voice_library): - tts_agent.actions["_config"].config[ - "narrator_voice_id" - ].value = "nope:nope" + tts_agent.actions["_config"].config["narrator_voice_id"].value = "nope:nope" # ensure no scene attribute if hasattr(tts_agent, "scene"): delattr(tts_agent, "scene") @@ -607,16 +575,12 @@ class TestUseAIAssistedSpeakerSeparation: def test_simple_with_quotes_still_false(self, tts_agent): self._set(tts_agent, "simple") msg = CharacterMessage(message='Alice: "hi"', source="ai") - assert ( - tts_agent.use_ai_assisted_speaker_separation('Alice: "hi"', msg) is False - ) + assert tts_agent.use_ai_assisted_speaker_separation('Alice: "hi"', msg) is False def test_player_source_always_false(self, tts_agent): self._set(tts_agent, "ai_assisted") msg = CharacterMessage(message='Alice: "hi"', source="player") - assert ( - tts_agent.use_ai_assisted_speaker_separation('Alice: "hi"', msg) is False - ) + assert tts_agent.use_ai_assisted_speaker_separation('Alice: "hi"', msg) is False def test_no_message_no_quotes_false(self, tts_agent): self._set(tts_agent, "ai_assisted") @@ -646,7 +610,9 @@ class TestUseAIAssistedSpeakerSeparation: bad = object() # accessing .source raises AttributeError # Bypass the no-message branch by ensuring text contains quotes AND # message is truthy; this means we hit the message.source line and raise. - assert tts_agent.use_ai_assisted_speaker_separation('he "said" hi', bad) is False + assert ( + tts_agent.use_ai_assisted_speaker_separation('he "said" hi', bad) is False + ) # --------------------------------------------------------------------------- @@ -655,9 +621,7 @@ class TestUseAIAssistedSpeakerSeparation: class TestNarratorVoiceIdChoices: - def test_only_includes_voices_for_ready_apis( - self, tts_agent, fresh_voice_library - ): + def test_only_includes_voices_for_ready_apis(self, tts_agent, fresh_voice_library): # Only kokoro ready by default v_kokoro = Voice(label="Adam", provider="kokoro", provider_id="am_adam") v_openai = Voice(label="Alloy", provider="openai", provider_id="alloy") @@ -741,9 +705,7 @@ class TestOpenAICompatibleMixin: tts_agent.actions[self.SLUG].config[ "api_url" ].value = "http://localhost:8000/v1" - assert ( - tts_agent._openai_compatible_not_configured_reason(self.SLUG) is None - ) + assert tts_agent._openai_compatible_not_configured_reason(self.SLUG) is None def test_not_configured_action_targets_settings(self, tts_agent): self._register(tts_agent) @@ -759,9 +721,7 @@ class TestOpenAICompatibleMixin: tts_agent.actions[self.SLUG].config[ "api_url" ].value = "http://localhost:8000/v1" - assert ( - tts_agent._openai_compatible_not_configured_action(self.SLUG) is None - ) + assert tts_agent._openai_compatible_not_configured_action(self.SLUG) is None def test_agent_details_error_when_not_configured(self, tts_agent): self._register(tts_agent) @@ -816,9 +776,7 @@ class TestApiStatus: def test_enabled_ready_configured_flags_reflect_state(self, tts_agent): # Register a dynamic OpenAI-compatible backend; it should appear in # api_status alongside the static APIs. - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") tts_agent.actions["_config"].config["apis"].value = ["kokoro"] # Ensure the dynamic backend is not configured tts_agent.actions["test-be"].config["api_url"].value = "" @@ -839,9 +797,7 @@ class TestApiStatus: def test_surfaces_not_configured_reason_as_error_note(self, tts_agent): # A registered dynamic backend with no api_url should surface the # "API base URL not set" reason as an error note in api_status. - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") tts_agent.actions["_config"].config["apis"].value = ["test-be"] tts_agent.actions["test-be"].config["api_url"].value = "" @@ -927,9 +883,7 @@ async def generate_agent(tts_agent, fresh_voice_library, monkeypatch): tts_agent.scene = FakeScene(characters=[]) # Suppress emit / set_processing emission noise - monkeypatch.setattr( - "talemate.agents.tts.emit", MagicMock(), raising=False - ) + monkeypatch.setattr("talemate.agents.tts.emit", MagicMock(), raising=False) # Replace _generate_chunk with a recording AsyncMock so we capture what # would be sent for synthesis without actually generating audio. @@ -984,9 +938,7 @@ class TestTTSAgentGenerate: char_voice = Voice(label="Char", provider="kokoro", provider_id="cv") fresh_voice_library.voices[char_voice.id] = char_voice - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "none" + generate_agent.actions["_config"].config["speaker_separation"].value = "none" char = FakeCharacter("Alice", voice=char_voice) generate_agent.scene = FakeScene(characters=[char]) @@ -1010,9 +962,7 @@ class TestTTSAgentGenerate: fresh_voice_library.voices[char_voice.id] = char_voice char = FakeCharacter("Alice", voice=char_voice) generate_agent.scene = FakeScene(characters=[char]) - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "none" + generate_agent.actions["_config"].config["speaker_separation"].value = "none" await generate_agent.generate("hi", character=char) await _drain_queue(generate_agent) @@ -1027,9 +977,7 @@ class TestTTSAgentGenerate: async def test_simple_separation_splits_dialogue_and_exposition( self, generate_agent ): - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "simple" + generate_agent.actions["_config"].config["speaker_separation"].value = "simple" text = 'She walked away. "Goodbye!" he called after her.' await generate_agent.generate(text) @@ -1044,9 +992,7 @@ class TestTTSAgentGenerate: @pytest.mark.asyncio async def test_oversized_chunk_split_by_size(self, generate_agent): - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "none" + generate_agent.actions["_config"].config["speaker_separation"].value = "none" # Force a small max_generation_length via chunk_size <= max generate_agent.actions["kokoro"].config["chunk_size"].value = 64 @@ -1077,9 +1023,7 @@ class TestTTSAgentGenerate: called.append(list(chunks)) generate_agent._inject_audio_tags = spy - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "none" + generate_agent.actions["_config"].config["speaker_separation"].value = "none" await generate_agent.generate("Hello world") await _drain_queue(generate_agent) assert len(called) == 1 @@ -1088,9 +1032,7 @@ class TestTTSAgentGenerate: async def test_chunk_generate_and_prepare_fn_resolved_via_getattr( self, generate_agent ): - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "none" + generate_agent.actions["_config"].config["speaker_separation"].value = "none" await generate_agent.generate("hi") await _drain_queue(generate_agent) @@ -1110,9 +1052,7 @@ class TestTTSAgentGenerate: class TestQueueLifecycle: @pytest.mark.asyncio async def test_stop_and_clear_resets_state(self, generate_agent): - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "none" + generate_agent.actions["_config"].config["speaker_separation"].value = "none" # Enqueue multiple items by making _generate_chunk slow block = asyncio.Event() @@ -1138,9 +1078,7 @@ class TestQueueLifecycle: @pytest.mark.asyncio async def test_back_to_back_generate_appends_to_same_queue(self, generate_agent): - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "none" + generate_agent.actions["_config"].config["speaker_separation"].value = "none" # Block processing so the queue stays alive between calls block = asyncio.Event() @@ -1171,9 +1109,7 @@ class TestQueueLifecycle: @pytest.mark.asyncio async def test_queue_drains_naturally_resets_state(self, generate_agent): - generate_agent.actions["_config"].config[ - "speaker_separation" - ].value = "none" + generate_agent.actions["_config"].config["speaker_separation"].value = "none" # default _generate_chunk is an AsyncMock that returns immediately await generate_agent.generate("just one") await _drain_queue(generate_agent) @@ -1249,22 +1185,16 @@ class TestTTSAgentDynamicBackends: """ def test_register_appends_slug_to_dynamic_child_slugs(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") assert "test-be" in tts_agent.dynamic_child_slugs("openai_compatible") def test_synthesized_child_has_parent_key(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") assert "test-be" in tts_agent.actions assert tts_agent.actions["test-be"].parent_key == "openai_compatible" def test_synthesized_child_carries_full_per_backend_config(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") config = tts_agent.actions["test-be"].config # Per-backend config that used to live on the management tab now # lives here. @@ -1272,33 +1202,23 @@ class TestTTSAgentDynamicBackends: assert key in config def test_apis_choices_gains_dynamic_marker_entry(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") choices = tts_agent.actions["_config"].config["apis"].choices - match = next( - (c for c in choices if c.get("value") == "test-be"), None - ) + match = next((c for c in choices if c.get("value") == "test-be"), None) assert match is not None assert match.get("label") == "Test Backend" assert match.get("_dynamic_backend") is True def test_api_attr_configured_reflects_api_url(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") tts_agent.actions["test-be"].config["api_url"].value = "" assert tts_agent.api_attr("test-be", "configured") is False - tts_agent.actions["test-be"].config[ - "api_url" - ].value = "http://example/v1" + tts_agent.actions["test-be"].config["api_url"].value = "http://example/v1" assert tts_agent.api_attr("test-be", "configured") is True def test_api_method_returns_callable_for_generate(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") method = tts_agent.api_method("test-be", "generate") assert callable(method) @@ -1308,9 +1228,7 @@ class TestTTSAgentDynamicBackends: import pytest as _pytest with _pytest.raises(ValueError, match="reserved"): - tts_agent.register_dynamic_child( - "openai_compatible", "kokoro", "My Kokoro" - ) + tts_agent.register_dynamic_child("openai_compatible", "kokoro", "My Kokoro") def test_rejects_slug_collision_with_registry_key(self, tts_agent): import pytest as _pytest @@ -1330,17 +1248,13 @@ class TestTTSAgentDynamicBackendLifecycle: """ def test_unregister_removes_synthesized_action(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") assert "test-be" in tts_agent.actions tts_agent.unregister_dynamic_child("openai_compatible", "test-be") assert "test-be" not in tts_agent.actions def test_unregister_strips_slug_from_apis_value(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") tts_agent.actions["_config"].config["apis"].value = [ "kokoro", "test-be", @@ -1361,9 +1275,7 @@ class TestTTSAgentDynamicBackendLifecycle: monkeypatch.setattr(voice_library, "save_voice_library", _fake_save) - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") # Seed library: a voice from the dynamic backend, a voice from kokoro # (must survive), and a voice from another backend slug. @@ -1386,9 +1298,7 @@ class TestTTSAgentDynamicBackendLifecycle: assert v_other.id in fresh_voice_library.voices def test_unregister_apis_value_only_drops_slug_if_present(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") # apis.value does NOT contain test-be — unregister still works. tts_agent.actions["_config"].config["apis"].value = ["kokoro"] tts_agent.unregister_dynamic_child("openai_compatible", "test-be") @@ -1407,9 +1317,7 @@ class TestTTSAgentApiAttrFallback: assert tts_agent.kokoro_configured is True def test_dynamic_api_resolves_through_registry(self, tts_agent): - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") # Set the per-backend api_url, then ask api_attr — it must dispatch # to _openai_compatible_configured("test-be") rather than to a # nonexistent "test-be_configured" attribute. @@ -1432,10 +1340,7 @@ class TestTTSAgentApiAttrFallback: sentinel = object() # "ghost" is not a registered slug and not a static api — falls # through to getattr with the default. - assert ( - tts_agent.api_attr("ghost", "configured", default=sentinel) - is sentinel - ) + assert tts_agent.api_attr("ghost", "configured", default=sentinel) is sentinel class TestAutoSetupClients: @@ -1446,9 +1351,7 @@ class TestAutoSetupClients: """ @pytest.mark.asyncio - async def test_skips_when_automatic_setup_off( - self, tts_agent, monkeypatch - ): + async def test_skips_when_automatic_setup_off(self, tts_agent, monkeypatch): tts_agent.actions["_config"].config["automatic_setup"].value = False called = {"count": 0} @@ -1556,18 +1459,14 @@ class TestAutoSetupClients: assert emit_called["count"] == 1 @pytest.mark.asyncio - async def test_no_persist_when_nothing_changed( - self, tts_agent, monkeypatch - ): + async def test_no_persist_when_nothing_changed(self, tts_agent, monkeypatch): save_called = {"count": 0} async def _save(): save_called["count"] += 1 monkeypatch.setattr(tts_agent, "save_config", _save) - monkeypatch.setattr( - tts_agent, "emit_status", AsyncMock(return_value=None) - ) + monkeypatch.setattr(tts_agent, "emit_status", AsyncMock(return_value=None)) class IdempotentClient: name = "kobold" @@ -1576,16 +1475,12 @@ class TestAutoSetupClients: async def tts_openai_compatible_setup(self_, agent): return False - monkeypatch.setattr( - instance, "CLIENTS", {"kobold": IdempotentClient()} - ) + monkeypatch.setattr(instance, "CLIENTS", {"kobold": IdempotentClient()}) await tts_agent.setup_check() assert save_called["count"] == 0 @pytest.mark.asyncio - async def test_one_client_raising_does_not_break_loop( - self, tts_agent, monkeypatch - ): + async def test_one_client_raising_does_not_break_loop(self, tts_agent, monkeypatch): invoked = [] class BoomClient: @@ -1729,12 +1624,8 @@ class TestKoboldCppTTSSetup: from talemate.client.koboldcpp import KoboldCppClient # Pre-existing enabled backend; should remain enabled. - tts_agent.register_dynamic_child( - "openai_compatible", "kobold", "kobold" - ) - tts_agent.actions["kobold"].config[ - "api_url" - ].value = "http://localhost:5001/v1" + tts_agent.register_dynamic_child("openai_compatible", "kobold", "kobold") + tts_agent.actions["kobold"].config["api_url"].value = "http://localhost:5001/v1" tts_agent.actions["_config"].config["apis"].value = ["kobold"] result = await KoboldCppClient._tts_openai_compatible_setup_impl( @@ -1775,9 +1666,7 @@ class TestKoboldCppTTSSetup: assert backend.config["api_url"].value == "http://localhost:5001/v1" # Auto-enabled in apis flags so the narrator-voice dropdown sees it. - assert "my-kobold" in ( - tts_agent.actions["_config"].config["apis"].value or [] - ) + assert "my-kobold" in (tts_agent.actions["_config"].config["apis"].value or []) # Voice refresh fired once. tts_agent.refresh_backend_voices.assert_awaited_once_with("my-kobold") @@ -1788,12 +1677,8 @@ class TestKoboldCppTTSSetup: ): # Backend already points at this kobold and is in apis.value — no # change. - tts_agent.register_dynamic_child( - "openai_compatible", "manual", "Manual" - ) - tts_agent.actions["manual"].config[ - "api_url" - ].value = "http://localhost:5001/v1" + tts_agent.register_dynamic_child("openai_compatible", "manual", "Manual") + tts_agent.actions["manual"].config["api_url"].value = "http://localhost:5001/v1" tts_agent.actions["_config"].config["apis"].value = ["manual"] self._patch_httpx( @@ -1818,12 +1703,8 @@ class TestKoboldCppTTSSetup: # Backend exists for this URL but is currently disabled (kobold was # restarted with TTS unloaded earlier). Now voices are back: should # auto-re-enable without creating a duplicate. - tts_agent.register_dynamic_child( - "openai_compatible", "kobold", "kobold" - ) - tts_agent.actions["kobold"].config[ - "api_url" - ].value = "http://localhost:5001/v1" + tts_agent.register_dynamic_child("openai_compatible", "kobold", "kobold") + tts_agent.actions["kobold"].config["api_url"].value = "http://localhost:5001/v1" # apis.value is empty → backend is currently disabled. tts_agent.actions["_config"].config["apis"].value = [] @@ -1850,12 +1731,8 @@ class TestKoboldCppTTSSetup: # Backend exists for this URL and is currently enabled. Kobold up # but reports empty voices list (TTS model unloaded after restart). # Should drop slug from apis.value but keep the backend's config. - tts_agent.register_dynamic_child( - "openai_compatible", "kobold", "kobold" - ) - tts_agent.actions["kobold"].config[ - "api_url" - ].value = "http://localhost:5001/v1" + tts_agent.register_dynamic_child("openai_compatible", "kobold", "kobold") + tts_agent.actions["kobold"].config["api_url"].value = "http://localhost:5001/v1" tts_agent.actions["kobold"].config["api_key"].value = "stay-please" tts_agent.actions["_config"].config["apis"].value = ["kobold"] @@ -1870,21 +1747,14 @@ class TestKoboldCppTTSSetup: assert tts_agent.actions["_config"].config["apis"].value == [] # Backend kept (config preserved). assert "kobold" in tts_agent.dynamic_child_slugs("openai_compatible") - assert ( - tts_agent.actions["kobold"].config["api_key"].value - == "stay-please" - ) + assert tts_agent.actions["kobold"].config["api_key"].value == "stay-please" @pytest.mark.asyncio async def test_no_change_when_already_disabled_and_no_voices( self, tts_agent, fresh_voice_library, fake_client, monkeypatch ): - tts_agent.register_dynamic_child( - "openai_compatible", "kobold", "kobold" - ) - tts_agent.actions["kobold"].config[ - "api_url" - ].value = "http://localhost:5001/v1" + tts_agent.register_dynamic_child("openai_compatible", "kobold", "kobold") + tts_agent.actions["kobold"].config["api_url"].value = "http://localhost:5001/v1" tts_agent.actions["_config"].config["apis"].value = [] self._patch_httpx(monkeypatch, 200, {"result": "KoboldCpp", "tts": False}) @@ -1903,12 +1773,8 @@ class TestKoboldCppTTSSetup: # Backend currently enabled. The probe raises (uncertain). State # must not change — otherwise transient network errors would flap # the user's setup. - tts_agent.register_dynamic_child( - "openai_compatible", "kobold", "kobold" - ) - tts_agent.actions["kobold"].config[ - "api_url" - ].value = "http://localhost:5001/v1" + tts_agent.register_dynamic_child("openai_compatible", "kobold", "kobold") + tts_agent.actions["kobold"].config["api_url"].value = "http://localhost:5001/v1" tts_agent.actions["_config"].config["apis"].value = ["kobold"] from talemate.client import koboldcpp as kobold_module @@ -1946,9 +1812,7 @@ class TestKoboldCppTTSSetup: "openai_compatible", "my-kobold", "Different Kobold" ) # That existing one points elsewhere, so idempotency check passes. - tts_agent.actions["my-kobold"].config[ - "api_url" - ].value = "http://other:5001/v1" + tts_agent.actions["my-kobold"].config["api_url"].value = "http://other:5001/v1" self._patch_httpx( monkeypatch, @@ -1994,23 +1858,19 @@ class TestKoboldCppTTSSetup: ) # Setup still succeeded — the voice fetch is best-effort. assert result is True - assert "my-kobold" in tts_agent.dynamic_child_slugs( - "openai_compatible" - ) + assert "my-kobold" in tts_agent.dynamic_child_slugs("openai_compatible") class TestRefreshBackendVoices: """``refresh_backend_voices`` should: - - fetch via api_method("fetch_voices") - - union new voices into the global library - - preserve user-customized voices (tags / non-default labels) - - drop auto-fetched voices the server no longer reports + - fetch via api_method("fetch_voices") + - union new voices into the global library + - preserve user-customized voices (tags / non-default labels) + - drop auto-fetched voices the server no longer reports """ @pytest.mark.asyncio - async def test_returns_zero_for_unknown_slug( - self, tts_agent, fresh_voice_library - ): + async def test_returns_zero_for_unknown_slug(self, tts_agent, fresh_voice_library): # No registered backend with that slug. count = await tts_agent.refresh_backend_voices("nope") assert count == 0 @@ -2026,9 +1886,7 @@ class TestRefreshBackendVoices: monkeypatch.setattr(voice_library, "save_voice_library", _fake_save) - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") fetched = [ Voice(label="alpha", provider="test-be", provider_id="alpha"), @@ -2066,9 +1924,7 @@ class TestRefreshBackendVoices: monkeypatch.setattr(voice_library, "save_voice_library", _fake_save) - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") # User has manually customized "alpha" with a non-default label and # tags. The backend still reports "alpha" — the customization must @@ -2095,9 +1951,7 @@ class TestRefreshBackendVoices: async def _fake_fetch(*args, **kwargs): # Server reports a fresh "alpha" with default label; the user's # customized entry should remain unchanged. - return [ - Voice(label="alpha", provider="test-be", provider_id="alpha") - ] + return [Voice(label="alpha", provider="test-be", provider_id="alpha")] monkeypatch.setattr( tts_agent, @@ -2125,9 +1979,7 @@ class TestRefreshBackendVoices: monkeypatch.setattr(voice_library, "save_voice_library", _fake_save) - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") # An auto-fetched voice (no tags, label == provider_id) the server # no longer reports — must be dropped. @@ -2139,9 +1991,7 @@ class TestRefreshBackendVoices: fresh_voice_library.voices[keep.id] = keep async def _fake_fetch(*args, **kwargs): - return [ - Voice(label="alpha", provider="test-be", provider_id="alpha") - ] + return [Voice(label="alpha", provider="test-be", provider_id="alpha")] monkeypatch.setattr( tts_agent, @@ -2164,9 +2014,7 @@ class TestRefreshBackendVoices: # Hard failures (HTTP errors, JSON shape mismatches, etc.) must # propagate so the websocket plugin can surface the real error to # the UI rather than rendering "no listing endpoint" copy. - tts_agent.register_dynamic_child( - "openai_compatible", "test-be", "Test Backend" - ) + tts_agent.register_dynamic_child("openai_compatible", "test-be", "Test Backend") async def _boom(*args, **kwargs): raise RuntimeError("simulated network failure") @@ -2197,9 +2045,7 @@ class TestParseVoicesPayload: ] def test_list_of_id_dicts(self): - voices = _parse_voices_payload( - [{"id": "alpha"}, {"id": "beta"}], "test-be" - ) + voices = _parse_voices_payload([{"id": "alpha"}, {"id": "beta"}], "test-be") assert [v.provider_id for v in voices] == ["alpha", "beta"] # No display name → label falls back to id. assert all(v.label == v.provider_id for v in voices) @@ -2260,24 +2106,18 @@ class TestResolveVoicesUrl: """ def test_relative_path_appends_under_versioned_base(self): - url = _resolve_voices_url( - "https://api.openai.com/v1", "audio/speech/voices" - ) + url = _resolve_voices_url("https://api.openai.com/v1", "audio/speech/voices") assert url == "https://api.openai.com/v1/audio/speech/voices" def test_relative_path_works_with_trailing_slash_base(self): - url = _resolve_voices_url( - "https://api.openai.com/v1/", "voices" - ) + url = _resolve_voices_url("https://api.openai.com/v1/", "voices") assert url == "https://api.openai.com/v1/voices" def test_absolute_path_anchors_to_host_root(self): # Leading "/" on the candidate means: replace the base URL's path, # use the host root. This lets users opt into endpoints that don't # live underneath /v1. - url = _resolve_voices_url( - "https://api.openai.com/v1", "/custom/voices" - ) + url = _resolve_voices_url("https://api.openai.com/v1", "/custom/voices") assert url == "https://api.openai.com/custom/voices" def test_full_url_passes_through(self):