Services API Reference

This section provides technical documentation for the various services used by the agent.

Tweet Generator

services.tweet_generator

Tweet Generator Service for Pokemon LLM Agent.

Orchestrates the tweet generation flow: 1. Generate image via ComfyUI (with achievement-specific prompts) 2. Generate tweet text via LLM 3. Send to Chronicle as a DRAFT for Admin approval

Note: Discord approval and direct Twitter posting have been replaced by Chronicle Drafts.

Classes

TweetGenerator

Orchestrates tweet generation with image and text, sending to Chronicle as a draft.

Source code in services/tweet_generator.py

class TweetGenerator:
    """
    Orchestrates tweet generation with image and text, sending to Chronicle as a draft.
    """

    def __init__(
        self,
        image_service: ComfyUIImageService = None,
        llm_client: Any = None,
    ):
        """
        Initialize the tweet generator.
        """
        self.image_service = image_service or create_image_service()
        self.llm_client = llm_client
        self.chronicle_client = get_chronicle_client()
        self.enabled = True

        log.info("Tweet Generator initialized with Chronicle Client")

    async def generate_and_post_achievement_tweet(
        self,
        achievement: Achievement,
        run_state: Any = None,
        game_state: dict = None,
        screenshot_path: str = None,
    ) -> TweetGeneratorOutcome:
        """
        Generate content and send to Chronicle as a draft.

        Args:
            achievement: The achievement that triggered this tweet
            run_state: RunState object with run context
            game_state: Current game state dict
            screenshot_path: Path to the current game screenshot (optional)

        Returns:
            TweetGeneratorOutcome with result and details
        """
        if not self.enabled:
            return TweetGeneratorOutcome(
                result=TweetGeneratorResult.DISABLED,
                achievement_type=achievement.achievement_type,
                error="Tweet generation is disabled",
            )

        # Get achievement-specific image prompt
        image_prompt = ACHIEVEMENT_IMAGE_PROMPTS.get(
            achievement.achievement_type,
            AchievementImagePrompt(),  # Empty prompt as fallback
        )

        # Build run context
        run_context = self._build_run_context(
            is_continuing_run=True,
            run_state=run_state,
            game_state=game_state,
            achievement=achievement,
        )

        current_image_path = None
        current_tweet_text = None

        try:
            # 1. Generate Image
            log.info(
                f"Generating tweet image for {achievement.achievement_type.value}..."
            )
            current_image_path = await self.image_service.generate_image(
                positive_prompt_addition=image_prompt.positive_prompt,
                negative_prompt_addition=image_prompt.negative_prompt,
            )

            if not current_image_path:
                return TweetGeneratorOutcome(
                    result=TweetGeneratorResult.ERROR,
                    achievement_type=achievement.achievement_type,
                    error="Failed to generate image",
                )

            # 2. Generate Text
            log.info("Generating tweet text...")
            current_tweet_text = await self._generate_achievement_tweet_text(
                achievement, run_state, game_state
            )

            if not current_tweet_text:
                return TweetGeneratorOutcome(
                    result=TweetGeneratorResult.ERROR,
                    achievement_type=achievement.achievement_type,
                    image_path=current_image_path,
                    error="Failed to generate tweet text",
                )

            # 3. Send to Chronicle as Draft
            log.info("Sending draft to Chronicle...")
            success = await self._send_to_chronicle(
                tweet_text=current_tweet_text,
                image_path=current_image_path,
                achievement=achievement,
                run_context=run_context,
                screenshot_path=screenshot_path,
            )

            if success:
                log.info(
                    f"✅ Tweet draft created in Chronicle for {achievement.achievement_type.value}"
                )
                return TweetGeneratorOutcome(
                    result=TweetGeneratorResult.POSTED,
                    achievement_type=achievement.achievement_type,
                    tweet_text=current_tweet_text,
                    image_path=current_image_path,
                    tweet_url="chronicle-draft",  # Placeholder
                )
            else:
                return TweetGeneratorOutcome(
                    result=TweetGeneratorResult.ERROR,
                    achievement_type=achievement.achievement_type,
                    tweet_text=current_tweet_text,
                    image_path=current_image_path,
                    error="Failed to send draft to Chronicle",
                )

        except Exception as e:
            log.error(f"Tweet generation error: {e}")
            return TweetGeneratorOutcome(
                result=TweetGeneratorResult.ERROR,
                achievement_type=achievement.achievement_type,
                error=str(e),
            )

    async def generate_and_post_tweet(
        self,
        is_continuing_run: bool,
        run_state: Any = None,
        game_state: dict = None,
    ) -> TweetGeneratorOutcome:
        """
        Generate and post a stream start tweet (no specific achievement).
        """
        # Create a stream start achievement for consistent handling
        stream_achievement = Achievement(
            achievement_type=AchievementType.STREAM_START,
            triggered_at=datetime.now().isoformat(),
            context={"is_continuing": is_continuing_run},
        )

        return await self.generate_and_post_achievement_tweet(
            achievement=stream_achievement,
            run_state=run_state,
            game_state=game_state,
        )

    async def _send_to_chronicle(
        self,
        tweet_text: str,
        image_path: str,
        achievement: Achievement,
        run_context: Dict[str, Any],
        screenshot_path: str = None,
    ) -> bool:
        """Send tweet draft to Chronicle Server."""

        # Prepare attributes from context
        attributes = [
            {"trait_type": "Type", "value": "Tweet Draft"},
            {
                "trait_type": "Achievement",
                "value": achievement.achievement_type.value,
            },
        ]

        # Add simple context fields
        for k, v in run_context.items():
            if isinstance(v, (str, int, float, bool)):
                attributes.append(
                    {"trait_type": k.replace("_", " ").title(), "value": v}
                )

        gallery_paths = []
        # We can also attach the screenshot as a gallery item if available
        if screenshot_path and os.path.exists(screenshot_path):
            gallery_paths.append(screenshot_path)

        success = await self.chronicle_client.create_draft(
            name=f"Tweet: {achievement.get_title()}",
            symbol="TWEET",
            description=tweet_text,
            attributes=attributes,
            image_path=image_path,
            gallery_paths=gallery_paths,
            status="draft",
        )

        return success

    def _build_run_context(
        self,
        is_continuing_run: bool,
        run_state: Any = None,
        game_state: dict = None,
        achievement: Achievement = None,
    ) -> Dict[str, Any]:
        """Build run context dict."""
        context = {
            "is_continuing": is_continuing_run,
        }

        # Add achievement info
        if achievement:
            context["achievement"] = achievement.achievement_type.value
            if achievement.context:
                # Merge achievement context
                context.update(achievement.context)

        if game_state:
            # Location
            map_name = game_state.get("map_name", "")
            if map_name:
                context["location"] = map_name.replace("_", " ").title()

            # Badges
            badges = game_state.get("badges", [])
            if isinstance(badges, list):
                context["badges"] = len(badges)
            else:
                context["badges"] = badges

        if run_state:
            # Playtime
            elapsed = getattr(run_state, "elapsed_seconds", 0)
            if elapsed:
                hours = elapsed / 3600
                context["playtime"] = f"{hours:.1f} hours"

        return context

    async def _generate_achievement_tweet_text(
        self,
        achievement: Achievement,
        run_state: Any = None,
        game_state: dict = None,
    ) -> Optional[str]:
        """Generate tweet text for a specific achievement."""
        if not self.llm_client:
            log.warning("No LLM client configured for tweet generation")
            return self._fallback_achievement_tweet(achievement)

        try:
            from core.prompts import build_achievement_tweet_prompt

            prompt = build_achievement_tweet_prompt(achievement, run_state, game_state)

            # Call LLM
            response = await self._call_llm(prompt)

            if response:
                # Clean up response (remove quotes, extra whitespace)
                tweet_text = response.strip().strip("\"'")

                # Ensure it fits Twitter limit
                if len(tweet_text) > 280:
                    tweet_text = tweet_text[:277] + "..."

                return tweet_text

        except ImportError:
            log.warning("Achievement tweet prompt not found, using fallback")
        except Exception as e:
            log.error(f"LLM tweet generation failed: {e}")

        return self._fallback_achievement_tweet(achievement)

    async def _call_llm(self, prompt: str) -> Optional[str]:
        """Call the LLM to generate text."""
        if not self.llm_client:
            return None

        try:
            # Handle different LLM client types
            if hasattr(self.llm_client, "chat"):
                # OpenAI-style client (synchronous)
                response = self.llm_client.chat.completions.create(
                    model=os.getenv("Z_AI_MODEL", "glm-4.6"),
                    messages=[{"role": "user", "content": prompt}],
                    max_tokens=150,
                    temperature=0.8,
                )
                return response.choices[0].message.content
            elif callable(self.llm_client):
                # Simple callable client
                result = self.llm_client(prompt)
                if asyncio.iscoroutine(result):
                    return await result
                return result
            else:
                log.warning(f"Unknown LLM client type: {type(self.llm_client)}")
                return None
        except Exception as e:
            log.error(f"LLM call failed: {e}")
            return None

    def _fallback_achievement_tweet(
        self,
        achievement: Achievement,
    ) -> str:
        """Generate fallback tweet text for an achievement."""
        # Simple fallback map
        return f"Just achieved {achievement.achievement_type.value}! #PokemonRed #LassPlaysPokemon"

Functions

__init__(image_service=None, llm_client=None)

Initialize the tweet generator.

Source code in services/tweet_generator.py

def __init__(
    self,
    image_service: ComfyUIImageService = None,
    llm_client: Any = None,
):
    """
    Initialize the tweet generator.
    """
    self.image_service = image_service or create_image_service()
    self.llm_client = llm_client
    self.chronicle_client = get_chronicle_client()
    self.enabled = True

    log.info("Tweet Generator initialized with Chronicle Client")

generate_and_post_achievement_tweet(achievement, run_state=None, game_state=None, screenshot_path=None) async

Generate content and send to Chronicle as a draft.

Parameters:

Name	Type	Description	Default
achievement	Achievement	The achievement that triggered this tweet	required
run_state	Any	RunState object with run context	None
game_state	dict	Current game state dict	None
screenshot_path	str	Path to the current game screenshot (optional)	None

Returns:

Type	Description
TweetGeneratorOutcome	TweetGeneratorOutcome with result and details

Source code in services/tweet_generator.py

async def generate_and_post_achievement_tweet(
    self,
    achievement: Achievement,
    run_state: Any = None,
    game_state: dict = None,
    screenshot_path: str = None,
) -> TweetGeneratorOutcome:
    """
    Generate content and send to Chronicle as a draft.

    Args:
        achievement: The achievement that triggered this tweet
        run_state: RunState object with run context
        game_state: Current game state dict
        screenshot_path: Path to the current game screenshot (optional)

    Returns:
        TweetGeneratorOutcome with result and details
    """
    if not self.enabled:
        return TweetGeneratorOutcome(
            result=TweetGeneratorResult.DISABLED,
            achievement_type=achievement.achievement_type,
            error="Tweet generation is disabled",
        )

    # Get achievement-specific image prompt
    image_prompt = ACHIEVEMENT_IMAGE_PROMPTS.get(
        achievement.achievement_type,
        AchievementImagePrompt(),  # Empty prompt as fallback
    )

    # Build run context
    run_context = self._build_run_context(
        is_continuing_run=True,
        run_state=run_state,
        game_state=game_state,
        achievement=achievement,
    )

    current_image_path = None
    current_tweet_text = None

    try:
        # 1. Generate Image
        log.info(
            f"Generating tweet image for {achievement.achievement_type.value}..."
        )
        current_image_path = await self.image_service.generate_image(
            positive_prompt_addition=image_prompt.positive_prompt,
            negative_prompt_addition=image_prompt.negative_prompt,
        )

        if not current_image_path:
            return TweetGeneratorOutcome(
                result=TweetGeneratorResult.ERROR,
                achievement_type=achievement.achievement_type,
                error="Failed to generate image",
            )

        # 2. Generate Text
        log.info("Generating tweet text...")
        current_tweet_text = await self._generate_achievement_tweet_text(
            achievement, run_state, game_state
        )

        if not current_tweet_text:
            return TweetGeneratorOutcome(
                result=TweetGeneratorResult.ERROR,
                achievement_type=achievement.achievement_type,
                image_path=current_image_path,
                error="Failed to generate tweet text",
            )

        # 3. Send to Chronicle as Draft
        log.info("Sending draft to Chronicle...")
        success = await self._send_to_chronicle(
            tweet_text=current_tweet_text,
            image_path=current_image_path,
            achievement=achievement,
            run_context=run_context,
            screenshot_path=screenshot_path,
        )

        if success:
            log.info(
                f"✅ Tweet draft created in Chronicle for {achievement.achievement_type.value}"
            )
            return TweetGeneratorOutcome(
                result=TweetGeneratorResult.POSTED,
                achievement_type=achievement.achievement_type,
                tweet_text=current_tweet_text,
                image_path=current_image_path,
                tweet_url="chronicle-draft",  # Placeholder
            )
        else:
            return TweetGeneratorOutcome(
                result=TweetGeneratorResult.ERROR,
                achievement_type=achievement.achievement_type,
                tweet_text=current_tweet_text,
                image_path=current_image_path,
                error="Failed to send draft to Chronicle",
            )

    except Exception as e:
        log.error(f"Tweet generation error: {e}")
        return TweetGeneratorOutcome(
            result=TweetGeneratorResult.ERROR,
            achievement_type=achievement.achievement_type,
            error=str(e),
        )

generate_and_post_tweet(is_continuing_run, run_state=None, game_state=None) async

Generate and post a stream start tweet (no specific achievement).

Source code in services/tweet_generator.py

async def generate_and_post_tweet(
    self,
    is_continuing_run: bool,
    run_state: Any = None,
    game_state: dict = None,
) -> TweetGeneratorOutcome:
    """
    Generate and post a stream start tweet (no specific achievement).
    """
    # Create a stream start achievement for consistent handling
    stream_achievement = Achievement(
        achievement_type=AchievementType.STREAM_START,
        triggered_at=datetime.now().isoformat(),
        context={"is_continuing": is_continuing_run},
    )

    return await self.generate_and_post_achievement_tweet(
        achievement=stream_achievement,
        run_state=run_state,
        game_state=game_state,
    )

TweetGeneratorOutcome dataclass

Outcome of the tweet generation process.

Source code in services/tweet_generator.py

@dataclass
class TweetGeneratorOutcome:
    """Outcome of the tweet generation process."""

    result: TweetGeneratorResult
    achievement_type: Optional[AchievementType] = None
    tweet_text: Optional[str] = None
    image_path: Optional[str] = None
    tweet_url: Optional[str] = None  # Will be None for drafts
    regeneration_count: int = 0
    error: Optional[str] = None
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

TweetGeneratorResult

Bases: Enum

Final result of the tweet generation process.

Source code in services/tweet_generator.py

class TweetGeneratorResult(Enum):
    """Final result of the tweet generation process."""

    POSTED = "posted"  # Successfully sent to Chronicle as Draft
    MAX_REGENERATIONS = "max_regenerations"  # Hit max regeneration limit
    DISABLED = "disabled"  # Service disabled
    ERROR = "error"  # Error during process

Functions

create_tweet_generator(llm_client=None)

Factory function to create a TweetGenerator instance.

Source code in services/tweet_generator.py

def create_tweet_generator(llm_client: Any = None) -> TweetGenerator:
    """Factory function to create a TweetGenerator instance."""
    return TweetGenerator(llm_client=llm_client)

Zora Poster Service

services.zora_poster_service

Zora Poster Service.

Orchestrates the creation of Zora posts (coins) for achievements. - Polls ZoraAchievementTracker for pending achievements - Generates images (ComfyUI) or retrieves screenshots - Creates collages for gallery posts - Calls Chronicle Server (Zora Sidecar) to create DRAFTS - Handles rate limiting and queues - Supports token-gated exclusive content

Classes

ZoraPosterService

Orchestrates Zora posting workflow.

Source code in services/zora_poster_service.py

class ZoraPosterService:
    """
    Orchestrates Zora posting workflow.
    """

    def __init__(self):
        self.tracker = get_zora_achievement_tracker()
        self.screenshot_manager = get_screenshot_manager()
        self.chronicle_client = get_chronicle_client()

        self._running = False
        self._last_post_time = 0
        self._worker_task: Optional[asyncio.Task] = None

        # State for ComfyUI generation (if needed)
        self.comfyui_service = None  # To be injected or imported

        if not ZORA_POSTING_ENABLED:
            log.info("Zora posting is disabled in config")

    async def start(self):
        """Start the posting worker loop."""
        if not ZORA_POSTING_ENABLED or not HTTPX_AVAILABLE:
            return

        if self._running:
            return

        self._running = True
        self._worker_task = asyncio.create_task(self._worker_loop())
        log.info("Zora poster service started")

    async def stop(self):
        """Stop the posting worker."""
        self._running = False
        if self._worker_task:
            self._worker_task.cancel()
            try:
                await self._worker_task
            except asyncio.CancelledError:
                pass

        await self.chronicle_client.close()

    async def _worker_loop(self):
        """Main loop checking for pending posts."""
        while self._running:
            try:
                # 1. Check rate limit
                time_since_last = time.time() - self._last_post_time
                if time_since_last < MIN_POST_INTERVAL:
                    await asyncio.sleep(10)
                    continue

                # 2. Check pending achievement
                achievement = self.tracker.get_pending_achievement()
                if not achievement:
                    await asyncio.sleep(5)
                    continue

                log.info(f"Processing Zora achievement: {achievement.get_title()}")

                # 3. Prepare content (Image & Metadata)
                post_data = await self._prepare_post(achievement)

                if not post_data:
                    log.error(
                        "Failed to prepare post data. Skipping/Clearing achievement."
                    )
                    self.tracker.clear_pending_achievement()
                    continue

                # 4. Send to Chronicle Server (Sidecar) as DRAFT
                success = await self._send_to_sidecar(post_data)

                if success:
                    # 5. Mark complete
                    self.tracker.mark_posted(
                        key=achievement.achievement_type.value,
                        coin_address=post_data.get("coinAddress", "draft-id"),
                    )

                    # Update tracker with coin address and status by iterating keys
                    key = None
                    for k, v in self.tracker.triggered.items():
                        if v == achievement:
                            key = k
                            break
                    if key:
                        self.tracker.mark_posted(
                            key, post_data.get("coinAddress", "draft-id")
                        )

                    self.tracker.clear_pending_achievement()
                    self._last_post_time = time.time()
                    log.info(
                        f"✅ Successfully sent draft to Chronicle: {achievement.get_title()}"
                    )
                else:
                    log.error("Failed to send draft to Chronicle. Will retry later.")
                    await asyncio.sleep(60)  # Backoff

            except Exception as e:
                log.error(f"Error in Zora poster loop: {e}")
                await asyncio.sleep(30)

    async def _prepare_post(
        self, achievement: ZoraAchievement
    ) -> Optional[Dict[str, Any]]:
        """Prepare image and metadata for the post."""

        # Determine image path
        image_path = None
        exclusive_path = None

        if achievement.tier == ZoraAchievementTier.MAJOR:
            # Major: Try to generate AI image or use screenshot
            image_path = self.screenshot_manager.get_latest_path()

            # If gating enabled, we could use the raw screenshot or save file as exclusive content
            # For now, let's just use the screenshot as exclusive too for testing the pipeline
            if ZORA_GATING_ENABLED:
                exclusive_path = image_path

        elif achievement.tier == ZoraAchievementTier.PROGRESS:
            # Progress: Create gallery collage
            image_path = self._create_collage()

        else:  # MINOR
            # Minor: Latest screenshot
            image_path = self.screenshot_manager.get_latest_path()

        if not image_path or not os.path.exists(image_path):
            log.warning("No image available for Zora post")
            return None

        # Prepare metadata
        title = achievement.get_title()
        description = self._generate_description(achievement)

        attributes = [
            {"trait_type": "Tier", "value": achievement.tier.value},
            {"trait_type": "Type", "value": achievement.achievement_type.value},
            {"trait_type": "Post Number", "value": achievement.post_number},
        ]

        # Add context attributes
        for k, v in achievement.context.items():
            if isinstance(v, (str, int, float, bool)):
                attributes.append(
                    {"trait_type": k.replace("_", " ").title(), "value": v}
                )

        return {
            "name": title,
            "symbol": f"LLP-{achievement.post_number:03d}",
            "description": description,
            "image_path": image_path,
            "exclusive_path": exclusive_path,
            "attributes": attributes,
            "status": "draft",  # FORCE DRAFT
        }

    def _generate_description(self, achievement: ZoraAchievement) -> str:
        """Generate description text."""
        lines = [
            f"🎮 LassPlaysPokemon - Post #{achievement.post_number}",
            f"Achievement: {achievement.get_title()}",
            f"Time: {achievement.triggered_at}",
            "",
        ]

        # Context details
        for k, v in achievement.context.items():
            lines.append(f"• {k.replace('_', ' ').title()}: {v}")

        lines.append("")

        if achievement.should_include_footer():
            lines.append("Watch live: twitch.tv/lassplayspokemon")
            lines.append("Token:  on pump.fun")
            if ZORA_GATING_ENABLED:
                lines.append("🔒 Exclusive content available for holders.")

        return "\n".join(lines)

    def _create_collage(self) -> Optional[str]:
        """Create a collage from recent screenshots."""
        if not PIL_AVAILABLE:
            return self.screenshot_manager.get_latest_path()

        paths = self.screenshot_manager.get_gallery_paths(count=6)
        if not paths:
            return None

        try:
            images = [Image.open(p) for p in paths]
            if not images:
                return None

            # Create a 2x3 or 3x2 grid depending on count
            width, height = images[0].size
            cols = 2
            rows = (len(images) + 1) // 2

            collage = Image.new("RGB", (width * cols, height * rows))

            for i, img in enumerate(images):
                x = (i % cols) * width
                y = (i // cols) * height
                collage.paste(img, (x, y))

            # Save collage
            output_path = os.path.join(
                self.screenshot_manager.gallery_dir, f"collage_{int(time.time())}.png"
            )
            collage.save(output_path)
            return output_path

        except Exception as e:
            log.error(f"Failed to create collage: {e}")
            return self.screenshot_manager.get_latest_path()

    async def _send_to_sidecar(self, post_data: Dict) -> bool:
        """Send post data to Chronicle Server."""

        attributes = []
        try:
            attributes = post_data["attributes"]
        except Exception:
            attributes = []

        success = await self.chronicle_client.create_draft(
            name=post_data["name"],
            symbol=post_data["symbol"],
            description=post_data["description"],
            attributes=attributes,
            image_path=post_data["image_path"],
            exclusive_path=post_data.get("exclusive_path"),
            status=post_data.get("status", "draft"),
        )

        if success:
            # Side effect: Mock the coin address since it's a draft
            post_data["coinAddress"] = "draft-pending"

        return success

Functions

start() async

Start the posting worker loop.

Source code in services/zora_poster_service.py

async def start(self):
    """Start the posting worker loop."""
    if not ZORA_POSTING_ENABLED or not HTTPX_AVAILABLE:
        return

    if self._running:
        return

    self._running = True
    self._worker_task = asyncio.create_task(self._worker_loop())
    log.info("Zora poster service started")

stop() async

Stop the posting worker.

Source code in services/zora_poster_service.py

async def stop(self):
    """Stop the posting worker."""
    self._running = False
    if self._worker_task:
        self._worker_task.cancel()
        try:
            await self._worker_task
        except asyncio.CancelledError:
            pass

    await self.chronicle_client.close()

Functions

ComfyUI Image Service

services.comfyui_image_service

ComfyUI Image Generation Service for Pokemon LLM Agent.

Triggers image generation workflows on a ComfyUI server to create images for tweet attachments.

Classes

ComfyUIImageService

Image generation service using ComfyUI as the backend. Generates 1024x1024 images for tweet attachments.

Source code in services/comfyui_image_service.py

class ComfyUIImageService:
    """
    Image generation service using ComfyUI as the backend.
    Generates 1024x1024 images for tweet attachments.
    """

    def __init__(
        self,
        base_url: str = None,
        workflow_path: str = None,
        output_dir: str = None,
        timeout: float = 120.0,
    ):
        """
        Initialize the ComfyUI Image Generation service.

        Args:
            base_url: ComfyUI server URL (e.g., "http://localhost:8188")
            workflow_path: Path to the image generation workflow JSON file
            output_dir: Local directory to save generated images
            timeout: Request timeout in seconds (image gen can be slow)
        """
        self.base_url = (
            base_url or os.getenv("COMFYUI_URL", "http://localhost:8188")
        ).rstrip("/")
        self.workflow_path = workflow_path or os.getenv(
            "COMFYUI_IMAGE_WORKFLOW", "workflows/lass-image-gen.json"
        )
        self.output_dir = output_dir or os.path.join(os.getcwd(), "outputs", "images")
        self.timeout = timeout

        # Basic auth credentials (shared with TTS/Vision)
        self._auth_username = os.getenv("COMFYUI_USERNAME", "")
        self._auth_password = os.getenv("COMFYUI_PASSWORD", "")
        self._auth: Optional[httpx.BasicAuth] = None
        if self._auth_username and self._auth_password:
            self._auth = httpx.BasicAuth(self._auth_username, self._auth_password)
            log.info(f"ComfyUI image auth configured for user: {self._auth_username}")

        self._workflow_template: Optional[dict] = None

        # Ensure output directory exists
        Path(self.output_dir).mkdir(parents=True, exist_ok=True)

        # Check if workflow exists
        if os.path.exists(self.workflow_path):
            log.info(
                f"ComfyUI Image Service initialized with workflow: {self.workflow_path}"
            )
        else:
            log.warning(f"Image workflow not found: {self.workflow_path}")

        # Cleanup stale images from previous runs
        self._cleanup_stale_images()

    def _cleanup_stale_images(self) -> None:
        """
        Remove any leftover image files from previous sessions.
        Called on startup to ensure clean state.
        """
        if not os.path.exists(self.output_dir):
            return

        cleaned = 0
        for f in os.listdir(self.output_dir):
            if f.endswith((".png", ".jpg", ".jpeg", ".webp")):
                try:
                    os.remove(os.path.join(self.output_dir, f))
                    cleaned += 1
                except Exception as e:
                    log.warning(f"Failed to cleanup image {f}: {e}")

        if cleaned > 0:
            log.info(f"Cleaned up {cleaned} stale image files from {self.output_dir}")

    def cleanup_image(self, image_path: str) -> None:
        """
        Remove an image file after it has been used (posted/denied).
        Called after tweet flow completes.
        """
        if image_path and os.path.exists(image_path):
            try:
                os.remove(image_path)
                log.debug(f"Cleaned up image: {os.path.basename(image_path)}")
            except Exception as e:
                log.warning(f"Failed to cleanup image: {e}")

    def _create_client(self) -> httpx.AsyncClient:
        """Create a new client instance."""
        return httpx.AsyncClient(timeout=self.timeout, auth=self._auth)

    async def check_connection(self) -> bool:
        """Check if ComfyUI server is reachable."""
        try:
            async with self._create_client() as client:
                response = await client.get(f"{self.base_url}/system_stats")
                return response.status_code == 200
        except Exception as e:
            log.warning(f"ComfyUI image connection check failed: {e}")
            return False

    async def load_workflow(self) -> Optional[dict]:
        """Load and convert the workflow template."""
        if self._workflow_template:
            return self._workflow_template

        if not self.workflow_path or not os.path.exists(self.workflow_path):
            log.warning(f"Image workflow not found: {self.workflow_path}")
            return None

        try:
            with open(self.workflow_path, "r") as f:
                raw = json.load(f)

            # Check if export format (has "nodes" list) or API format
            if "nodes" in raw:
                # Fetch object info to map widget values
                object_info = await self._fetch_object_info()
                self._workflow_template = self._convert_export_to_api_format(
                    raw, object_info
                )
            else:
                self._workflow_template = raw

            return self._workflow_template
        except Exception as e:
            log.error(f"Failed to load image workflow: {e}")
            return None

    async def _fetch_object_info(self) -> dict:
        """Fetch node definitions from ComfyUI to map widgets."""
        try:
            async with self._create_client() as client:
                response = await client.get(f"{self.base_url}/object_info")
                if response.status_code == 200:
                    return response.json()
        except Exception as e:
            log.warning(f"Failed to fetch object_info: {e}")
        return {}

    def _convert_export_to_api_format(
        self, export_workflow: dict, object_info: dict
    ) -> dict:
        """
        Convert ComfyUI export format (graph with nodes list) to API format (prompt dict).
        Uses object_info to map unnamed widget values to named inputs.
        """
        nodes = export_workflow.get("nodes", [])
        links = export_workflow.get("links", [])

        # Build link map: link_id -> (from_node_id, from_slot_index)
        link_map = {}
        for link in links:
            if len(link) >= 5:
                link_id, from_node, from_slot, to_node, to_slot, *_ = link
                link_map[link_id] = (from_node, from_slot)

        api_prompt = {}
        for node in nodes:
            node_id = str(node.get("id"))
            node_type = node.get("type")
            inputs = {}

            # Process linked inputs
            for input_def in node.get("inputs", []):
                input_name = input_def.get("name")
                link_id = input_def.get("link")
                if link_id and link_id in link_map:
                    from_node, from_slot = link_map[link_id]
                    inputs[input_name] = [str(from_node), from_slot]

            # Process widget values using object_info schema
            widgets_values = node.get("widgets_values", [])
            if widgets_values:
                # Get input names from schema
                node_info = object_info.get(node_type, {})
                input_schema = node_info.get("input", {})
                required = input_schema.get("required", {})
                optional = input_schema.get("optional", {})

                # Combine keys in order (ComfyUI standard seems to be required then optional,
                # but widgets usually correspond to specific non-link inputs)

                # Heuristic: Map widgets_values to remaining inputs that aren't links
                # This is complex because ComfyUI's internal mapping varies.
                # A safer bet for common nodes is often just positional mapping if names match known patterns
                # OR rely on specific known node types if generic fails.

                # Better approach: Iterate over required keys. If a key is NOT a link input (check `inputs`),
                # then it might be a widget.

                # Let's use a simpler heuristic for now that covers 90% of cases:
                # Map values to keys sequentially from the 'required' list that match widget types
                # (INT, FLOAT, STRING, COMBO)

                widget_keys = []
                # Scan required inputs
                for group in [required, optional]:
                    if isinstance(group, dict):
                        for name, config in group.items():
                            type_name = config[0]
                            # If it's a primitive type or list (COMBO), it's likely a widget
                            if type_name in [
                                "INT",
                                "FLOAT",
                                "STRING",
                                "BOOLEAN",
                            ] or isinstance(type_name, list):
                                widget_keys.append(name)

                # Assign values
                for i, value in enumerate(widgets_values):
                    if i < len(widget_keys):
                        key = widget_keys[i]
                        # Don't overwrite if it's already linked (rare for widgets but possible)
                        if key not in inputs:
                            inputs[key] = value

            # Fallback for explicit SaveImage prefix if schema failed
            if (
                node_type == "SaveImage"
                and "filename_prefix" not in inputs
                and widgets_values
            ):
                inputs["filename_prefix"] = widgets_values[0]

            api_prompt[node_id] = {"class_type": node_type, "inputs": inputs}

        return {"prompt": api_prompt}

    async def generate_image(
        self,
        seed: int = None,
        positive_prompt_addition: str = None,
        negative_prompt_addition: str = None,
    ) -> Optional[str]:
        """
        Generate an image using the ComfyUI workflow.

        Args:
            seed: Optional seed for reproducibility. If None, uses random.
            positive_prompt_addition: Text to append to the positive prompt.
            negative_prompt_addition: Text to append to the negative prompt.

        Returns:
            Path to the downloaded image file, or None on failure.
        """
        # Load workflow
        workflow = await self.load_workflow()
        if not workflow:
            log.error("Failed to load image generation workflow")
            return None

        import copy
        import random

        # Ensure we have a dict, whether it was raw API format or converted
        prompt_data = workflow.get("prompt", workflow)
        if not isinstance(prompt_data, dict):
            log.error(
                f"Invalid workflow format: expected dict, got {type(prompt_data)}"
            )
            return None

        prompt = copy.deepcopy(prompt_data)

        # Update seed if provided (find KSampler or similar nodes)
        if seed is None:
            seed = random.randint(0, 2**32 - 1)

        for node_id, node_data in prompt.items():
            class_type = node_data.get("class_type", "")

            # Update seed on sampler nodes
            if "Sampler" in class_type or "KSampler" in class_type:
                if "seed" in node_data.get("inputs", {}):
                    node_data["inputs"]["seed"] = seed
                    log.debug(f"Set seed={seed} on node {node_id}")

            # Update prompts on CLIPTextEncode nodes
            if class_type == "CLIPTextEncode" and (
                positive_prompt_addition or negative_prompt_addition
            ):
                inputs = node_data.get("inputs", {})
                # Check if this is a text widget (widgets_values style)
                # or direct input style
                if "text" in inputs and isinstance(inputs["text"], str):
                    current_text = inputs["text"]

                    # Determine if positive or negative based on content
                    is_negative = any(
                        neg in current_text.lower()
                        for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                    )

                    if is_negative and negative_prompt_addition:
                        inputs["text"] = current_text + ", " + negative_prompt_addition
                        log.debug(f"Added to negative prompt on node {node_id}")
                    elif not is_negative and positive_prompt_addition:
                        inputs["text"] = current_text + ", " + positive_prompt_addition
                        log.debug(f"Added to positive prompt on node {node_id}")

        # Queue the workflow
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue image workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued image generation workflow: {prompt_id}")

            # Wait for completion and get the image
            return await self._wait_for_completion(prompt_id)

        except Exception as e:
            log.error(f"Image generation error: {e}")
            return None

        import copy
        import random

        # Ensure we have a dict, whether it was raw API format or converted
        prompt_data = workflow.get("prompt", workflow)
        if not isinstance(prompt_data, dict):
            log.error(
                f"Invalid workflow format: expected dict, got {type(prompt_data)}"
            )
            return None

        prompt = copy.deepcopy(prompt_data)

        # Update seed if provided (find KSampler or similar nodes)
        if seed is None:
            seed = random.randint(0, 2**32 - 1)

        for node_id, node_data in prompt.items():
            class_type = node_data.get("class_type", "")

            # Update seed on sampler nodes
            if "Sampler" in class_type or "KSampler" in class_type:
                if "seed" in node_data.get("inputs", {}):
                    node_data["inputs"]["seed"] = seed
                    log.debug(f"Set seed={seed} on node {node_id}")

            # Update prompts on CLIPTextEncode nodes
            if class_type == "CLIPTextEncode" and (
                positive_prompt_addition or negative_prompt_addition
            ):
                inputs = node_data.get("inputs", {})
                # Check if this is a text widget (widgets_values style)
                # or direct input style
                if "text" in inputs and isinstance(inputs["text"], str):
                    current_text = inputs["text"]

                    # Determine if positive or negative based on content
                    is_negative = any(
                        neg in current_text.lower()
                        for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                    )

                    if is_negative and negative_prompt_addition:
                        inputs["text"] = current_text + ", " + negative_prompt_addition
                        log.debug(f"Added to negative prompt on node {node_id}")
                    elif not is_negative and positive_prompt_addition:
                        inputs["text"] = current_text + ", " + positive_prompt_addition
                        log.debug(f"Added to positive prompt on node {node_id}")

        # Queue the workflow
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue image workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued image generation workflow: {prompt_id}")

            # Wait for completion and get the image
            return await self._wait_for_completion(prompt_id)

        except Exception as e:
            log.error(f"Image generation error: {e}")
            return None

        import copy
        import random

        # Ensure we have a dict, whether it was raw API format or converted
        prompt_data = workflow.get("prompt", workflow)
        if not isinstance(prompt_data, dict):
            log.error(
                f"Invalid workflow format: expected dict, got {type(prompt_data)}"
            )
            return None

        prompt = copy.deepcopy(prompt_data)

        # Update seed if provided (find KSampler or similar nodes)
        if seed is None:
            seed = random.randint(0, 2**32 - 1)

        for node_id, node_data in prompt.items():
            class_type = node_data.get("class_type", "")

            # Update seed on sampler nodes
            if "Sampler" in class_type or "KSampler" in class_type:
                if "seed" in node_data.get("inputs", {}):
                    node_data["inputs"]["seed"] = seed
                    log.debug(f"Set seed={seed} on node {node_id}")

            # Update prompts on CLIPTextEncode nodes
            if class_type == "CLIPTextEncode" and (
                positive_prompt_addition or negative_prompt_addition
            ):
                inputs = node_data.get("inputs", {})
                # Check if this is a text widget (widgets_values style)
                # or direct input style
                if "text" in inputs and isinstance(inputs["text"], str):
                    current_text = inputs["text"]

                    # Determine if positive or negative based on content
                    is_negative = any(
                        neg in current_text.lower()
                        for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                    )

                    if is_negative and negative_prompt_addition:
                        inputs["text"] = current_text + ", " + negative_prompt_addition
                        log.debug(f"Added to negative prompt on node {node_id}")
                    elif not is_negative and positive_prompt_addition:
                        inputs["text"] = current_text + ", " + positive_prompt_addition
                        log.debug(f"Added to positive prompt on node {node_id}")

        # Queue the workflow
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue image workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued image generation workflow: {prompt_id}")

            # Wait for completion and get the image
            return await self._wait_for_completion(prompt_id)

        except Exception as e:
            log.error(f"Image generation error: {e}")
            return None

        import copy
        import random

        # Ensure we have a dict, whether it was raw API format or converted
        prompt_data = workflow.get("prompt", workflow)
        if not isinstance(prompt_data, dict):
            log.error(
                f"Invalid workflow format: expected dict, got {type(prompt_data)}"
            )
            return None

        prompt = copy.deepcopy(prompt_data)

        # Update seed if provided (find KSampler or similar nodes)
        if seed is None:
            seed = random.randint(0, 2**32 - 1)

        for node_id, node_data in prompt.items():
            class_type = node_data.get("class_type", "")

            # Update seed on sampler nodes
            if "Sampler" in class_type or "KSampler" in class_type:
                if "seed" in node_data.get("inputs", {}):
                    node_data["inputs"]["seed"] = seed
                    log.debug(f"Set seed={seed} on node {node_id}")

            # Update prompts on CLIPTextEncode nodes
            if class_type == "CLIPTextEncode" and (
                positive_prompt_addition or negative_prompt_addition
            ):
                inputs = node_data.get("inputs", {})
                # Check if this is a text widget (widgets_values style)
                # or direct input style
                if "text" in inputs and isinstance(inputs["text"], str):
                    current_text = inputs["text"]

                    # Determine if positive or negative based on content
                    is_negative = any(
                        neg in current_text.lower()
                        for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                    )

                    if is_negative and negative_prompt_addition:
                        inputs["text"] = current_text + ", " + negative_prompt_addition
                        log.debug(f"Added to negative prompt on node {node_id}")
                    elif not is_negative and positive_prompt_addition:
                        inputs["text"] = current_text + ", " + positive_prompt_addition
                        log.debug(f"Added to positive prompt on node {node_id}")

        # Queue the workflow
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue image workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued image generation workflow: {prompt_id}")

            # Wait for completion and get the image
            return await self._wait_for_completion(prompt_id)

        except Exception as e:
            log.error(f"Image generation error: {e}")
            return None

        import copy
        import random

        # Ensure we have a dict, whether it was raw API format or converted
        prompt_data = workflow.get("prompt", workflow)
        if not isinstance(prompt_data, dict):
            log.error(
                f"Invalid workflow format: expected dict, got {type(prompt_data)}"
            )
            return None

        prompt = copy.deepcopy(prompt_data)

        # Update seed if provided (find KSampler or similar nodes)
        if seed is None:
            seed = random.randint(0, 2**32 - 1)

        for node_id, node_data in prompt.items():
            class_type = node_data.get("class_type", "")

            # Update seed on sampler nodes
            if "Sampler" in class_type or "KSampler" in class_type:
                if "seed" in node_data.get("inputs", {}):
                    node_data["inputs"]["seed"] = seed
                    log.debug(f"Set seed={seed} on node {node_id}")

            # Update prompts on CLIPTextEncode nodes
            if class_type == "CLIPTextEncode" and (
                positive_prompt_addition or negative_prompt_addition
            ):
                inputs = node_data.get("inputs", {})
                # Check if this is a text widget (widgets_values style)
                # or direct input style
                if "text" in inputs and isinstance(inputs["text"], str):
                    current_text = inputs["text"]

                    # Determine if positive or negative based on content
                    is_negative = any(
                        neg in current_text.lower()
                        for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                    )

                    if is_negative and negative_prompt_addition:
                        inputs["text"] = current_text + ", " + negative_prompt_addition
                        log.debug(f"Added to negative prompt on node {node_id}")
                    elif not is_negative and positive_prompt_addition:
                        inputs["text"] = current_text + ", " + positive_prompt_addition
                        log.debug(f"Added to positive prompt on node {node_id}")

        # Queue the workflow
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue image workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued image generation workflow: {prompt_id}")

            # Wait for completion and get the image
            return await self._wait_for_completion(prompt_id)

        except Exception as e:
            log.error(f"Image generation error: {e}")
            return None

        import copy
        import random

        # Ensure we have a dict, whether it was raw API format or converted
        prompt_data = workflow.get("prompt", workflow)
        if not isinstance(prompt_data, dict):
            log.error(
                f"Invalid workflow format: expected dict, got {type(prompt_data)}"
            )
            return None

        prompt = copy.deepcopy(prompt_data)

        # Update seed if provided (find KSampler or similar nodes)
        if seed is None:
            seed = random.randint(0, 2**32 - 1)

        for node_id, node_data in prompt.items():
            class_type = node_data.get("class_type", "")

            # Update seed on sampler nodes
            if "Sampler" in class_type or "KSampler" in class_type:
                if "seed" in node_data.get("inputs", {}):
                    node_data["inputs"]["seed"] = seed
                    log.debug(f"Set seed={seed} on node {node_id}")

            # Update prompts on CLIPTextEncode nodes
            if class_type == "CLIPTextEncode" and (
                positive_prompt_addition or negative_prompt_addition
            ):
                inputs = node_data.get("inputs", {})
                # Check if this is a text widget (widgets_values style)
                # or direct input style
                if "text" in inputs and isinstance(inputs["text"], str):
                    current_text = inputs["text"]

                    # Determine if positive or negative based on content
                    is_negative = any(
                        neg in current_text.lower()
                        for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                    )

                    if is_negative and negative_prompt_addition:
                        inputs["text"] = current_text + ", " + negative_prompt_addition
                        log.debug(f"Added to negative prompt on node {node_id}")
                    elif not is_negative and positive_prompt_addition:
                        inputs["text"] = current_text + ", " + positive_prompt_addition
                        log.debug(f"Added to positive prompt on node {node_id}")

        # Queue the workflow
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue image workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued image generation workflow: {prompt_id}")

            # Wait for completion and get the image
            return await self._wait_for_completion(prompt_id)

        except Exception as e:
            log.error(f"Image generation error: {e}")
            return None

        import copy
        import random

        # Ensure we have a dict, whether it was raw API format or converted
        prompt_data = workflow.get("prompt", workflow)
        if not isinstance(prompt_data, dict):
            log.error(
                f"Invalid workflow format: expected dict, got {type(prompt_data)}"
            )
            return None

        prompt = copy.deepcopy(prompt_data)

        # Update seed if provided (find KSampler or similar nodes)
        if seed is None:
            seed = random.randint(0, 2**32 - 1)

        for node_id, node_data in prompt.items():
            class_type = node_data.get("class_type", "")

            # Update seed on sampler nodes
            if "Sampler" in class_type or "KSampler" in class_type:
                if "seed" in node_data.get("inputs", {}):
                    node_data["inputs"]["seed"] = seed
                    log.debug(f"Set seed={seed} on node {node_id}")

            # Update prompts on CLIPTextEncode nodes
            if class_type == "CLIPTextEncode" and (
                positive_prompt_addition or negative_prompt_addition
            ):
                inputs = node_data.get("inputs", {})
                # Check if this is a text widget (widgets_values style)
                # or direct input style
                if "text" in inputs and isinstance(inputs["text"], str):
                    current_text = inputs["text"]

                    # Determine if positive or negative based on content
                    is_negative = any(
                        neg in current_text.lower()
                        for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                    )

                    if is_negative and negative_prompt_addition:
                        inputs["text"] = current_text + ", " + negative_prompt_addition
                        log.debug(f"Added to negative prompt on node {node_id}")
                    elif not is_negative and positive_prompt_addition:
                        inputs["text"] = current_text + ", " + positive_prompt_addition
                        log.debug(f"Added to positive prompt on node {node_id}")

        # Queue the workflow
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue image workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued image generation workflow: {prompt_id}")

            # Wait for completion and get the image
            return await self._wait_for_completion(prompt_id)

        except Exception as e:
            log.error(f"Image generation error: {e}")
            return None

        import copy
        import random

        # Ensure we have a dict, whether it was raw API format or converted
        prompt_data = workflow.get("prompt", workflow)
        if not isinstance(prompt_data, dict):
            log.error(
                f"Invalid workflow format: expected dict, got {type(prompt_data)}"
            )
            return None

        prompt = copy.deepcopy(prompt_data)

        # Update seed if provided (find KSampler or similar nodes)
        if seed is None:
            seed = random.randint(0, 2**32 - 1)

        for node_id, node_data in prompt.items():
            class_type = node_data.get("class_type", "")

            # Update seed on sampler nodes
            if "Sampler" in class_type or "KSampler" in class_type:
                if "seed" in node_data.get("inputs", {}):
                    node_data["inputs"]["seed"] = seed
                    log.debug(f"Set seed={seed} on node {node_id}")

            # Update prompts on CLIPTextEncode nodes
            if class_type == "CLIPTextEncode" and (
                positive_prompt_addition or negative_prompt_addition
            ):
                inputs = node_data.get("inputs", {})
                # Check if this is a text widget (widgets_values style)
                # or direct input style
                if "text" in inputs and isinstance(inputs["text"], str):
                    current_text = inputs["text"]

                    # Determine if positive or negative based on content
                    is_negative = any(
                        neg in current_text.lower()
                        for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                    )

                    if is_negative and negative_prompt_addition:
                        inputs["text"] = current_text + ", " + negative_prompt_addition
                        log.debug(f"Added to negative prompt on node {node_id}")
                    elif not is_negative and positive_prompt_addition:
                        inputs["text"] = current_text + ", " + positive_prompt_addition
                        log.debug(f"Added to positive prompt on node {node_id}")

        # Queue the workflow
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue image workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued image generation workflow: {prompt_id}")

            # Wait for completion and get the image
            return await self._wait_for_completion(prompt_id)

        except Exception as e:
            log.error(f"Image generation error: {e}")
            return None

    async def _wait_for_completion(self, prompt_id: str) -> Optional[str]:
        """Wait for workflow completion and download the generated image."""
        async with self._create_client() as client:
            start = time.time()

            while time.time() - start < self.timeout:
                try:
                    # Check history for completion
                    res = await client.get(f"{self.base_url}/history/{prompt_id}")
                    if res.status_code == 200:
                        history = res.json()
                        if prompt_id in history:
                            outputs = history[prompt_id].get("outputs", {})

                            # Find SaveImage node output
                            for node_id, node_output in outputs.items():
                                if "images" in node_output:
                                    images = node_output["images"]
                                    if images:
                                        # Download the first image
                                        img_info = images[0]
                                        filename = img_info.get("filename")
                                        subfolder = img_info.get("subfolder", "")

                                        log.info(f"Image generated: {filename}")
                                        return await self._download_image(
                                            client, filename, subfolder
                                        )

                            # Check if workflow errored
                            status = history[prompt_id].get("status", {})
                            if status.get("status_str") == "error":
                                log.error(f"Workflow execution error: {status}")
                                return None

                    await asyncio.sleep(1.0)
                except Exception as e:
                    log.warning(f"Error checking image status: {e}")
                    await asyncio.sleep(2.0)

            log.error("Image generation timed out")
            return None

    async def _download_image(
        self, client: httpx.AsyncClient, filename: str, subfolder: str = ""
    ) -> Optional[str]:
        """Download generated image from ComfyUI server."""
        try:
            params = {"filename": filename, "type": "output"}
            if subfolder:
                params["subfolder"] = subfolder

            response = await client.get(f"{self.base_url}/view", params=params)

            if response.status_code == 200:
                # Save to local output directory
                local_filename = f"tweet_{int(time.time())}_{filename}"
                local_path = os.path.join(self.output_dir, local_filename)

                with open(local_path, "wb") as f:
                    f.write(response.content)

                log.info(f"Downloaded image to: {local_path}")
                return local_path
            else:
                log.error(f"Failed to download image: {response.status_code}")
                return None
        except Exception as e:
            log.error(f"Image download error: {e}")
            return None

Functions

__init__(base_url=None, workflow_path=None, output_dir=None, timeout=120.0)

Initialize the ComfyUI Image Generation service.

Parameters:

Name	Type	Description	Default
base_url	str	ComfyUI server URL (e.g., "http://localhost:8188")	None
workflow_path	str	Path to the image generation workflow JSON file	None
output_dir	str	Local directory to save generated images	None
timeout	float	Request timeout in seconds (image gen can be slow)	120.0

Source code in services/comfyui_image_service.py

def __init__(
    self,
    base_url: str = None,
    workflow_path: str = None,
    output_dir: str = None,
    timeout: float = 120.0,
):
    """
    Initialize the ComfyUI Image Generation service.

    Args:
        base_url: ComfyUI server URL (e.g., "http://localhost:8188")
        workflow_path: Path to the image generation workflow JSON file
        output_dir: Local directory to save generated images
        timeout: Request timeout in seconds (image gen can be slow)
    """
    self.base_url = (
        base_url or os.getenv("COMFYUI_URL", "http://localhost:8188")
    ).rstrip("/")
    self.workflow_path = workflow_path or os.getenv(
        "COMFYUI_IMAGE_WORKFLOW", "workflows/lass-image-gen.json"
    )
    self.output_dir = output_dir or os.path.join(os.getcwd(), "outputs", "images")
    self.timeout = timeout

    # Basic auth credentials (shared with TTS/Vision)
    self._auth_username = os.getenv("COMFYUI_USERNAME", "")
    self._auth_password = os.getenv("COMFYUI_PASSWORD", "")
    self._auth: Optional[httpx.BasicAuth] = None
    if self._auth_username and self._auth_password:
        self._auth = httpx.BasicAuth(self._auth_username, self._auth_password)
        log.info(f"ComfyUI image auth configured for user: {self._auth_username}")

    self._workflow_template: Optional[dict] = None

    # Ensure output directory exists
    Path(self.output_dir).mkdir(parents=True, exist_ok=True)

    # Check if workflow exists
    if os.path.exists(self.workflow_path):
        log.info(
            f"ComfyUI Image Service initialized with workflow: {self.workflow_path}"
        )
    else:
        log.warning(f"Image workflow not found: {self.workflow_path}")

    # Cleanup stale images from previous runs
    self._cleanup_stale_images()

check_connection() async

Check if ComfyUI server is reachable.

Source code in services/comfyui_image_service.py

async def check_connection(self) -> bool:
    """Check if ComfyUI server is reachable."""
    try:
        async with self._create_client() as client:
            response = await client.get(f"{self.base_url}/system_stats")
            return response.status_code == 200
    except Exception as e:
        log.warning(f"ComfyUI image connection check failed: {e}")
        return False

cleanup_image(image_path)

Remove an image file after it has been used (posted/denied). Called after tweet flow completes.

Source code in services/comfyui_image_service.py

def cleanup_image(self, image_path: str) -> None:
    """
    Remove an image file after it has been used (posted/denied).
    Called after tweet flow completes.
    """
    if image_path and os.path.exists(image_path):
        try:
            os.remove(image_path)
            log.debug(f"Cleaned up image: {os.path.basename(image_path)}")
        except Exception as e:
            log.warning(f"Failed to cleanup image: {e}")

generate_image(seed=None, positive_prompt_addition=None, negative_prompt_addition=None) async

Generate an image using the ComfyUI workflow.

Parameters:

Name	Type	Description	Default
seed	int	Optional seed for reproducibility. If None, uses random.	None
positive_prompt_addition	str	Text to append to the positive prompt.	None
negative_prompt_addition	str	Text to append to the negative prompt.	None

Returns:

Type	Description
Optional[str]	Path to the downloaded image file, or None on failure.

Source code in services/comfyui_image_service.py

async def generate_image(
    self,
    seed: int = None,
    positive_prompt_addition: str = None,
    negative_prompt_addition: str = None,
) -> Optional[str]:
    """
    Generate an image using the ComfyUI workflow.

    Args:
        seed: Optional seed for reproducibility. If None, uses random.
        positive_prompt_addition: Text to append to the positive prompt.
        negative_prompt_addition: Text to append to the negative prompt.

    Returns:
        Path to the downloaded image file, or None on failure.
    """
    # Load workflow
    workflow = await self.load_workflow()
    if not workflow:
        log.error("Failed to load image generation workflow")
        return None

    import copy
    import random

    # Ensure we have a dict, whether it was raw API format or converted
    prompt_data = workflow.get("prompt", workflow)
    if not isinstance(prompt_data, dict):
        log.error(
            f"Invalid workflow format: expected dict, got {type(prompt_data)}"
        )
        return None

    prompt = copy.deepcopy(prompt_data)

    # Update seed if provided (find KSampler or similar nodes)
    if seed is None:
        seed = random.randint(0, 2**32 - 1)

    for node_id, node_data in prompt.items():
        class_type = node_data.get("class_type", "")

        # Update seed on sampler nodes
        if "Sampler" in class_type or "KSampler" in class_type:
            if "seed" in node_data.get("inputs", {}):
                node_data["inputs"]["seed"] = seed
                log.debug(f"Set seed={seed} on node {node_id}")

        # Update prompts on CLIPTextEncode nodes
        if class_type == "CLIPTextEncode" and (
            positive_prompt_addition or negative_prompt_addition
        ):
            inputs = node_data.get("inputs", {})
            # Check if this is a text widget (widgets_values style)
            # or direct input style
            if "text" in inputs and isinstance(inputs["text"], str):
                current_text = inputs["text"]

                # Determine if positive or negative based on content
                is_negative = any(
                    neg in current_text.lower()
                    for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                )

                if is_negative and negative_prompt_addition:
                    inputs["text"] = current_text + ", " + negative_prompt_addition
                    log.debug(f"Added to negative prompt on node {node_id}")
                elif not is_negative and positive_prompt_addition:
                    inputs["text"] = current_text + ", " + positive_prompt_addition
                    log.debug(f"Added to positive prompt on node {node_id}")

    # Queue the workflow
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue image workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued image generation workflow: {prompt_id}")

        # Wait for completion and get the image
        return await self._wait_for_completion(prompt_id)

    except Exception as e:
        log.error(f"Image generation error: {e}")
        return None

    import copy
    import random

    # Ensure we have a dict, whether it was raw API format or converted
    prompt_data = workflow.get("prompt", workflow)
    if not isinstance(prompt_data, dict):
        log.error(
            f"Invalid workflow format: expected dict, got {type(prompt_data)}"
        )
        return None

    prompt = copy.deepcopy(prompt_data)

    # Update seed if provided (find KSampler or similar nodes)
    if seed is None:
        seed = random.randint(0, 2**32 - 1)

    for node_id, node_data in prompt.items():
        class_type = node_data.get("class_type", "")

        # Update seed on sampler nodes
        if "Sampler" in class_type or "KSampler" in class_type:
            if "seed" in node_data.get("inputs", {}):
                node_data["inputs"]["seed"] = seed
                log.debug(f"Set seed={seed} on node {node_id}")

        # Update prompts on CLIPTextEncode nodes
        if class_type == "CLIPTextEncode" and (
            positive_prompt_addition or negative_prompt_addition
        ):
            inputs = node_data.get("inputs", {})
            # Check if this is a text widget (widgets_values style)
            # or direct input style
            if "text" in inputs and isinstance(inputs["text"], str):
                current_text = inputs["text"]

                # Determine if positive or negative based on content
                is_negative = any(
                    neg in current_text.lower()
                    for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                )

                if is_negative and negative_prompt_addition:
                    inputs["text"] = current_text + ", " + negative_prompt_addition
                    log.debug(f"Added to negative prompt on node {node_id}")
                elif not is_negative and positive_prompt_addition:
                    inputs["text"] = current_text + ", " + positive_prompt_addition
                    log.debug(f"Added to positive prompt on node {node_id}")

    # Queue the workflow
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue image workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued image generation workflow: {prompt_id}")

        # Wait for completion and get the image
        return await self._wait_for_completion(prompt_id)

    except Exception as e:
        log.error(f"Image generation error: {e}")
        return None

    import copy
    import random

    # Ensure we have a dict, whether it was raw API format or converted
    prompt_data = workflow.get("prompt", workflow)
    if not isinstance(prompt_data, dict):
        log.error(
            f"Invalid workflow format: expected dict, got {type(prompt_data)}"
        )
        return None

    prompt = copy.deepcopy(prompt_data)

    # Update seed if provided (find KSampler or similar nodes)
    if seed is None:
        seed = random.randint(0, 2**32 - 1)

    for node_id, node_data in prompt.items():
        class_type = node_data.get("class_type", "")

        # Update seed on sampler nodes
        if "Sampler" in class_type or "KSampler" in class_type:
            if "seed" in node_data.get("inputs", {}):
                node_data["inputs"]["seed"] = seed
                log.debug(f"Set seed={seed} on node {node_id}")

        # Update prompts on CLIPTextEncode nodes
        if class_type == "CLIPTextEncode" and (
            positive_prompt_addition or negative_prompt_addition
        ):
            inputs = node_data.get("inputs", {})
            # Check if this is a text widget (widgets_values style)
            # or direct input style
            if "text" in inputs and isinstance(inputs["text"], str):
                current_text = inputs["text"]

                # Determine if positive or negative based on content
                is_negative = any(
                    neg in current_text.lower()
                    for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                )

                if is_negative and negative_prompt_addition:
                    inputs["text"] = current_text + ", " + negative_prompt_addition
                    log.debug(f"Added to negative prompt on node {node_id}")
                elif not is_negative and positive_prompt_addition:
                    inputs["text"] = current_text + ", " + positive_prompt_addition
                    log.debug(f"Added to positive prompt on node {node_id}")

    # Queue the workflow
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue image workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued image generation workflow: {prompt_id}")

        # Wait for completion and get the image
        return await self._wait_for_completion(prompt_id)

    except Exception as e:
        log.error(f"Image generation error: {e}")
        return None

    import copy
    import random

    # Ensure we have a dict, whether it was raw API format or converted
    prompt_data = workflow.get("prompt", workflow)
    if not isinstance(prompt_data, dict):
        log.error(
            f"Invalid workflow format: expected dict, got {type(prompt_data)}"
        )
        return None

    prompt = copy.deepcopy(prompt_data)

    # Update seed if provided (find KSampler or similar nodes)
    if seed is None:
        seed = random.randint(0, 2**32 - 1)

    for node_id, node_data in prompt.items():
        class_type = node_data.get("class_type", "")

        # Update seed on sampler nodes
        if "Sampler" in class_type or "KSampler" in class_type:
            if "seed" in node_data.get("inputs", {}):
                node_data["inputs"]["seed"] = seed
                log.debug(f"Set seed={seed} on node {node_id}")

        # Update prompts on CLIPTextEncode nodes
        if class_type == "CLIPTextEncode" and (
            positive_prompt_addition or negative_prompt_addition
        ):
            inputs = node_data.get("inputs", {})
            # Check if this is a text widget (widgets_values style)
            # or direct input style
            if "text" in inputs and isinstance(inputs["text"], str):
                current_text = inputs["text"]

                # Determine if positive or negative based on content
                is_negative = any(
                    neg in current_text.lower()
                    for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                )

                if is_negative and negative_prompt_addition:
                    inputs["text"] = current_text + ", " + negative_prompt_addition
                    log.debug(f"Added to negative prompt on node {node_id}")
                elif not is_negative and positive_prompt_addition:
                    inputs["text"] = current_text + ", " + positive_prompt_addition
                    log.debug(f"Added to positive prompt on node {node_id}")

    # Queue the workflow
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue image workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued image generation workflow: {prompt_id}")

        # Wait for completion and get the image
        return await self._wait_for_completion(prompt_id)

    except Exception as e:
        log.error(f"Image generation error: {e}")
        return None

    import copy
    import random

    # Ensure we have a dict, whether it was raw API format or converted
    prompt_data = workflow.get("prompt", workflow)
    if not isinstance(prompt_data, dict):
        log.error(
            f"Invalid workflow format: expected dict, got {type(prompt_data)}"
        )
        return None

    prompt = copy.deepcopy(prompt_data)

    # Update seed if provided (find KSampler or similar nodes)
    if seed is None:
        seed = random.randint(0, 2**32 - 1)

    for node_id, node_data in prompt.items():
        class_type = node_data.get("class_type", "")

        # Update seed on sampler nodes
        if "Sampler" in class_type or "KSampler" in class_type:
            if "seed" in node_data.get("inputs", {}):
                node_data["inputs"]["seed"] = seed
                log.debug(f"Set seed={seed} on node {node_id}")

        # Update prompts on CLIPTextEncode nodes
        if class_type == "CLIPTextEncode" and (
            positive_prompt_addition or negative_prompt_addition
        ):
            inputs = node_data.get("inputs", {})
            # Check if this is a text widget (widgets_values style)
            # or direct input style
            if "text" in inputs and isinstance(inputs["text"], str):
                current_text = inputs["text"]

                # Determine if positive or negative based on content
                is_negative = any(
                    neg in current_text.lower()
                    for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                )

                if is_negative and negative_prompt_addition:
                    inputs["text"] = current_text + ", " + negative_prompt_addition
                    log.debug(f"Added to negative prompt on node {node_id}")
                elif not is_negative and positive_prompt_addition:
                    inputs["text"] = current_text + ", " + positive_prompt_addition
                    log.debug(f"Added to positive prompt on node {node_id}")

    # Queue the workflow
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue image workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued image generation workflow: {prompt_id}")

        # Wait for completion and get the image
        return await self._wait_for_completion(prompt_id)

    except Exception as e:
        log.error(f"Image generation error: {e}")
        return None

    import copy
    import random

    # Ensure we have a dict, whether it was raw API format or converted
    prompt_data = workflow.get("prompt", workflow)
    if not isinstance(prompt_data, dict):
        log.error(
            f"Invalid workflow format: expected dict, got {type(prompt_data)}"
        )
        return None

    prompt = copy.deepcopy(prompt_data)

    # Update seed if provided (find KSampler or similar nodes)
    if seed is None:
        seed = random.randint(0, 2**32 - 1)

    for node_id, node_data in prompt.items():
        class_type = node_data.get("class_type", "")

        # Update seed on sampler nodes
        if "Sampler" in class_type or "KSampler" in class_type:
            if "seed" in node_data.get("inputs", {}):
                node_data["inputs"]["seed"] = seed
                log.debug(f"Set seed={seed} on node {node_id}")

        # Update prompts on CLIPTextEncode nodes
        if class_type == "CLIPTextEncode" and (
            positive_prompt_addition or negative_prompt_addition
        ):
            inputs = node_data.get("inputs", {})
            # Check if this is a text widget (widgets_values style)
            # or direct input style
            if "text" in inputs and isinstance(inputs["text"], str):
                current_text = inputs["text"]

                # Determine if positive or negative based on content
                is_negative = any(
                    neg in current_text.lower()
                    for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                )

                if is_negative and negative_prompt_addition:
                    inputs["text"] = current_text + ", " + negative_prompt_addition
                    log.debug(f"Added to negative prompt on node {node_id}")
                elif not is_negative and positive_prompt_addition:
                    inputs["text"] = current_text + ", " + positive_prompt_addition
                    log.debug(f"Added to positive prompt on node {node_id}")

    # Queue the workflow
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue image workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued image generation workflow: {prompt_id}")

        # Wait for completion and get the image
        return await self._wait_for_completion(prompt_id)

    except Exception as e:
        log.error(f"Image generation error: {e}")
        return None

    import copy
    import random

    # Ensure we have a dict, whether it was raw API format or converted
    prompt_data = workflow.get("prompt", workflow)
    if not isinstance(prompt_data, dict):
        log.error(
            f"Invalid workflow format: expected dict, got {type(prompt_data)}"
        )
        return None

    prompt = copy.deepcopy(prompt_data)

    # Update seed if provided (find KSampler or similar nodes)
    if seed is None:
        seed = random.randint(0, 2**32 - 1)

    for node_id, node_data in prompt.items():
        class_type = node_data.get("class_type", "")

        # Update seed on sampler nodes
        if "Sampler" in class_type or "KSampler" in class_type:
            if "seed" in node_data.get("inputs", {}):
                node_data["inputs"]["seed"] = seed
                log.debug(f"Set seed={seed} on node {node_id}")

        # Update prompts on CLIPTextEncode nodes
        if class_type == "CLIPTextEncode" and (
            positive_prompt_addition or negative_prompt_addition
        ):
            inputs = node_data.get("inputs", {})
            # Check if this is a text widget (widgets_values style)
            # or direct input style
            if "text" in inputs and isinstance(inputs["text"], str):
                current_text = inputs["text"]

                # Determine if positive or negative based on content
                is_negative = any(
                    neg in current_text.lower()
                    for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                )

                if is_negative and negative_prompt_addition:
                    inputs["text"] = current_text + ", " + negative_prompt_addition
                    log.debug(f"Added to negative prompt on node {node_id}")
                elif not is_negative and positive_prompt_addition:
                    inputs["text"] = current_text + ", " + positive_prompt_addition
                    log.debug(f"Added to positive prompt on node {node_id}")

    # Queue the workflow
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue image workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued image generation workflow: {prompt_id}")

        # Wait for completion and get the image
        return await self._wait_for_completion(prompt_id)

    except Exception as e:
        log.error(f"Image generation error: {e}")
        return None

    import copy
    import random

    # Ensure we have a dict, whether it was raw API format or converted
    prompt_data = workflow.get("prompt", workflow)
    if not isinstance(prompt_data, dict):
        log.error(
            f"Invalid workflow format: expected dict, got {type(prompt_data)}"
        )
        return None

    prompt = copy.deepcopy(prompt_data)

    # Update seed if provided (find KSampler or similar nodes)
    if seed is None:
        seed = random.randint(0, 2**32 - 1)

    for node_id, node_data in prompt.items():
        class_type = node_data.get("class_type", "")

        # Update seed on sampler nodes
        if "Sampler" in class_type or "KSampler" in class_type:
            if "seed" in node_data.get("inputs", {}):
                node_data["inputs"]["seed"] = seed
                log.debug(f"Set seed={seed} on node {node_id}")

        # Update prompts on CLIPTextEncode nodes
        if class_type == "CLIPTextEncode" and (
            positive_prompt_addition or negative_prompt_addition
        ):
            inputs = node_data.get("inputs", {})
            # Check if this is a text widget (widgets_values style)
            # or direct input style
            if "text" in inputs and isinstance(inputs["text"], str):
                current_text = inputs["text"]

                # Determine if positive or negative based on content
                is_negative = any(
                    neg in current_text.lower()
                    for neg in ["lowres", "worst quality", "bad anatomy", "ugly"]
                )

                if is_negative and negative_prompt_addition:
                    inputs["text"] = current_text + ", " + negative_prompt_addition
                    log.debug(f"Added to negative prompt on node {node_id}")
                elif not is_negative and positive_prompt_addition:
                    inputs["text"] = current_text + ", " + positive_prompt_addition
                    log.debug(f"Added to positive prompt on node {node_id}")

    # Queue the workflow
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue image workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued image generation workflow: {prompt_id}")

        # Wait for completion and get the image
        return await self._wait_for_completion(prompt_id)

    except Exception as e:
        log.error(f"Image generation error: {e}")
        return None

load_workflow() async

Load and convert the workflow template.

Source code in services/comfyui_image_service.py

async def load_workflow(self) -> Optional[dict]:
    """Load and convert the workflow template."""
    if self._workflow_template:
        return self._workflow_template

    if not self.workflow_path or not os.path.exists(self.workflow_path):
        log.warning(f"Image workflow not found: {self.workflow_path}")
        return None

    try:
        with open(self.workflow_path, "r") as f:
            raw = json.load(f)

        # Check if export format (has "nodes" list) or API format
        if "nodes" in raw:
            # Fetch object info to map widget values
            object_info = await self._fetch_object_info()
            self._workflow_template = self._convert_export_to_api_format(
                raw, object_info
            )
        else:
            self._workflow_template = raw

        return self._workflow_template
    except Exception as e:
        log.error(f"Failed to load image workflow: {e}")
        return None

Functions

create_image_service(base_url=None, workflow_path=None, output_dir=None)

Factory function to create a ComfyUI Image Service instance.

Source code in services/comfyui_image_service.py

def create_image_service(
    base_url: str = None,
    workflow_path: str = None,
    output_dir: str = None,
) -> ComfyUIImageService:
    """Factory function to create a ComfyUI Image Service instance."""
    return ComfyUIImageService(
        base_url=base_url,
        workflow_path=workflow_path,
        output_dir=output_dir,
    )

ComfyUI TTS Service

services.comfyui_tts_service

ComfyUI Text-to-Speech Service for Pokemon LLM Agent. Triggers TTS workflows on a ComfyUI server to generate speech from text.

Classes

ComfyUITTSService

Text-to-Speech service using ComfyUI as the backend.

Triggers TTS workflows on ComfyUI and retrieves generated audio. Supports both localhost and hosted ComfyUI instances.

Source code in services/comfyui_tts_service.py

class ComfyUITTSService:
    """
    Text-to-Speech service using ComfyUI as the backend.

    Triggers TTS workflows on ComfyUI and retrieves generated audio.
    Supports both localhost and hosted ComfyUI instances.
    """

    # Priority levels
    PRIORITY_COMMENTARY = 100  # Game commentary (highest)
    PRIORITY_CHAT_RESPONSE = 50  # Chat responses
    PRIORITY_LOW = 10  # Low priority items

    def __init__(
        self,
        base_url: str = None,
        workflow_path: str = None,
        output_dir: str = None,
        timeout: float = 10.0,  # Reduced from 60s for faster fallback
        on_playback_start: callable = None,  # (text, duration, metadata) -> Awaitable
        audio_speed: float = None,
        audio_pitch: float = None,
    ):
        """
        Initialize the ComfyUI TTS service.

        Args:
            base_url: ComfyUI server URL (e.g., "http://localhost:8188")
            workflow_path: Path to the TTS workflow JSON file
            output_dir: Directory to save generated audio files
            timeout: Request timeout in seconds
            on_playback_start: Callback(text, duration_ms) called when audio starts playing
            audio_speed: Playback speed multiplier (e.g., 1.1 = 10% faster)
            audio_pitch: Pitch shift in semitones (e.g., 2 = 2 semitones higher)
        """
        self.base_url = (
            base_url or os.getenv("COMFYUI_URL", "http://localhost:8188")
        ).rstrip("/")
        self.workflow_path = workflow_path or os.getenv("COMFYUI_TTS_WORKFLOW", "")
        self.output_dir = output_dir or os.path.join(os.getcwd(), "outputs", "audio")
        self.timeout = timeout
        self.on_playback_start = on_playback_start

        # Basic auth credentials (optional - for hosted ComfyUI instances)
        self._auth_username = os.getenv("COMFYUI_USERNAME", "")
        self._auth_password = os.getenv("COMFYUI_PASSWORD", "")
        self._auth: Optional[httpx.BasicAuth] = None
        if self._auth_username and self._auth_password:
            self._auth = httpx.BasicAuth(self._auth_username, self._auth_password)
            log.info(
                f"🔐 ComfyUI basic auth configured for user: {self._auth_username}"
            )

        # Audio post-processing settings (from env or args)
        self.audio_speed = (
            audio_speed
            if audio_speed is not None
            else float(os.getenv("TTS_AUDIO_SPEED", "1.0"))
        )
        self.audio_pitch = (
            audio_pitch
            if audio_pitch is not None
            else float(os.getenv("TTS_AUDIO_PITCH", "0"))
        )  # In semitones

        # Request queue (priority queue simulation with sorting)
        self._queue: List[TTSRequest] = []
        self._processing = False
        self._client: Optional[httpx.AsyncClient] = None

        # Request tracking
        self._current_request: Optional[TTSRequest] = None
        self._current_audio_path: Optional[str] = (
            None  # Track for post-playback cleanup
        )
        self._cancelled = False

        # Lock to ensure TTS requests are serialized (no overlap)
        self._tts_lock = asyncio.Lock()

        # Cached workflow template
        self._workflow_template: Optional[dict] = None

        # Initialize audio components
        self.audio_player = AudioPlayer()
        self.audio_processor = AudioProcessor()

        # Create output directory
        os.makedirs(self.output_dir, exist_ok=True)

        # Cleanup stale audio files from previous runs
        self._cleanup_stale_audio()

        # Check if configured and enabled
        # TTS_ENABLED defaults to true if not set, for backward compatibility
        tts_enabled = os.getenv("TTS_ENABLED", "true").lower() in ("true", "1", "yes")
        self._is_configured = bool(self.base_url) and tts_enabled

        if not tts_enabled:
            log.info("🔇 TTS disabled via TTS_ENABLED=false in .env")
        elif not self._is_configured:
            log.warning("ComfyUI TTS not configured. Set COMFYUI_URL in .env")

    def _cleanup_stale_audio(self) -> None:
        """
        Remove any leftover audio files from previous sessions.
        Called on startup to ensure clean state.
        """
        if not os.path.exists(self.output_dir):
            return

        cleaned = 0
        for f in os.listdir(self.output_dir):
            if f.endswith((".flac", ".mp3", ".wav")):
                try:
                    os.remove(os.path.join(self.output_dir, f))
                    cleaned += 1
                except Exception as e:
                    log.warning(f"Failed to cleanup {f}: {e}")

        if cleaned > 0:
            log.info(f"♻️ Cleaned up {cleaned} stale audio files from {self.output_dir}")

    def clear_queue(self) -> int:
        """
        Clear all pending TTS requests from the queue.
        Useful on startup to clear orphaned requests from previous sessions.
        Returns the number of cleared items.
        """
        count = len(self._queue)
        self._queue.clear()
        self._processing = False
        self._cancelled = False
        if count > 0:
            log.info(f"🧹 Cleared {count} orphaned TTS requests from queue")
        return count

    def _cleanup_audio_file(self, audio_path: str) -> None:
        """
        Remove an audio file after playback.
        Called after wait_for_playback completes.
        """
        if audio_path and os.path.exists(audio_path):
            try:
                os.remove(audio_path)
                log.debug(f"♻️ Cleaned up: {os.path.basename(audio_path)}")
            except Exception as e:
                log.warning(f"Failed to cleanup audio: {e}")

    def _get_audio_duration_ms(self, audio_path: str) -> Optional[int]:
        """Get audio duration in milliseconds."""
        return self.audio_processor.get_duration_ms(audio_path)

    def _process_audio_speed_pitch(self, audio_path: str) -> str:
        """Apply speed and pitch adjustments to audio."""
        return self.audio_processor.process_speed_pitch(
            audio_path, speed=self.audio_speed, pitch=self.audio_pitch
        )

    @property
    def is_available(self) -> bool:
        """Check if TTS service is configured."""
        return self._is_configured

    async def _get_client(self) -> httpx.AsyncClient:
        """Get or create HTTP client with optional basic auth."""
        if self._client is None or self._client.is_closed:
            self._client = httpx.AsyncClient(timeout=self.timeout, auth=self._auth)
        return self._client

    def cancel_current(self):
        """Cancel the currently playing audio and any pending ComfyUI workflow."""
        self._cancelled = True

        # Stop audio player
        self.audio_player.stop_playback()

        # Also cancel any pending ComfyUI workflow via /interrupt endpoint
        try:
            import httpx

            with httpx.Client(timeout=2.0, auth=self._auth) as client:
                response = client.post(f"{self.base_url}/interrupt")
                if response.status_code == 200:
                    log.info("🔇 Cancelled pending ComfyUI TTS workflow")
                else:
                    log.debug(f"ComfyUI interrupt returned {response.status_code}")
        except Exception as e:
            log.debug(f"Could not interrupt ComfyUI workflow: {e}")

    def cancel_all_and_interrupt(self) -> int:
        """
        Cancel EVERYTHING - current playback, queue, and ComfyUI workflows.
        Use this when new high-priority commentary arrives to ensure it plays immediately.

        Returns:
            Number of queue items cleared
        """
        # 1. Stop any current playback immediately
        self._cancelled = True
        self.audio_player.stop_playback()

        # 2. Cancel all pending synthesis tasks
        cancelled_tasks = 0
        for request in self._queue:
            if request.synthesis_task and not request.synthesis_task.done():
                request.synthesis_task.cancel()
                cancelled_tasks += 1

        # 3. Clear the entire queue
        queue_count = len(self._queue)
        self._queue.clear()
        self._processing = False

        # 4. Interrupt any pending ComfyUI workflow
        try:
            import httpx

            with httpx.Client(timeout=2.0, auth=self._auth) as client:
                response = client.post(f"{self.base_url}/interrupt")
                if response.status_code == 200:
                    log.info("🔇 Interrupted pending ComfyUI TTS workflow")
        except Exception as e:
            log.debug(f"Could not interrupt ComfyUI workflow: {e}")

        if queue_count > 0 or cancelled_tasks > 0:
            log.info(
                f"🧹 CANCEL ALL: cleared {queue_count} queue items, cancelled {cancelled_tasks} synthesis tasks"
            )

        return queue_count

    def play_audio_ephemeral(self, audio_path: str) -> bool:
        """Play audio file ephemerally (no file retention)."""
        if self.audio_speed != 1.0 or self.audio_pitch != 0:
            audio_path = self._process_audio_speed_pitch(audio_path)

        return self.audio_player.play_audio(audio_path)

    async def wait_for_playback(self, timeout: float = 30.0) -> bool:
        """Wait for current audio playback to complete."""
        return await self.audio_player.wait_for_playback(timeout)

    async def check_connection(self) -> bool:
        """
        Check if ComfyUI server is reachable.
        Returns True if connected, False otherwise.
        """
        try:
            client = await self._get_client()
            response = await client.get(f"{self.base_url}/system_stats")
            return response.status_code == 200
        except Exception as e:
            log.warning(f"ComfyUI connection check failed: {e}")
            return False

    def load_workflow(self) -> Optional[dict]:
        """
        Load the TTS workflow template from file.
        Converts ComfyUI export format (nodes array) to API format (prompt dict).
        """
        if self._workflow_template:
            return self._workflow_template

        if not self.workflow_path or not os.path.exists(self.workflow_path):
            log.warning(f"TTS workflow not found: {self.workflow_path}")
            return None

        try:
            with open(self.workflow_path, "r") as f:
                raw_workflow = json.load(f)
            log.info(f"Loaded TTS workflow from: {self.workflow_path}")

            # Check if this is export format (has 'nodes' array) vs API format (has 'prompt' dict)
            if "nodes" in raw_workflow:
                # Convert from export format to API format
                api_format = self._convert_export_to_api_format(raw_workflow)
                self._workflow_template = api_format
                log.info(
                    f"Converted workflow from export format to API format ({len(api_format)} nodes)"
                )
            else:
                self._workflow_template = raw_workflow

            return self._workflow_template
        except Exception as e:
            log.error(f"Failed to load TTS workflow: {e}")
            return None

    def _convert_export_to_api_format(self, export_workflow: dict) -> dict:
        """
        Convert ComfyUI export format (nodes array) to API prompt format (dict of node dicts).

        Export format has:
        - "nodes": [{"id": 1, "type": "NodeType", "widgets_values": [...], "inputs": [...]}]
        - "links": [[link_id, from_node, from_slot, to_node, to_slot, type]]

        API format expects:
        - {"1": {"class_type": "NodeType", "inputs": {...}}}
        """
        nodes = export_workflow.get("nodes", [])
        links = export_workflow.get("links", [])

        # Build link lookup: (to_node_id, to_slot_idx) -> (from_node_id, from_slot_idx)
        link_map = {}
        for link in links:
            if len(link) >= 5:
                link_id, from_node, from_slot, to_node, to_slot, *_ = link
                link_map[link_id] = (from_node, from_slot)

        api_prompt = {}

        for node in nodes:
            node_id = str(node.get("id"))
            node_type = node.get("type")

            # Skip Note nodes - they're not part of the execution graph
            if node_type == "Note":
                continue

            # Build inputs dict from connected links and widget values
            inputs = {}

            # Get connected inputs from links
            node_inputs = node.get("inputs", [])
            for input_def in node_inputs:
                input_name = input_def.get("name")
                link_id = input_def.get("link")
                if link_id and link_id in link_map:
                    from_node, from_slot = link_map[link_id]
                    inputs[input_name] = [str(from_node), from_slot]

            # Store widget values for later text injection
            # For ChatterboxTTS: widgets_values is [model, text, max_len, cfg_temp, temp, top_p, top_k, ...]
            widget_values = node.get("widgets_values", [])

            # Map widget values to input names based on node type
            if node_type == "ChatterboxTTS":
                # Server's current ChatterboxTTS required inputs (from /object_info):
                # model_pack_name, text, max_new_tokens, flow_cfg_scale, exaggeration,
                # temperature, cfg_weight, repetition_penalty, min_p, top_p, seed, use_watermark
                # Optional: audio_prompt

                # Use server's default values to ensure compatibility
                chatterbox_defaults = {
                    "model_pack_name": "resembleai_default_voice",
                    "text": "Hello, this is a test of Chatterbox TTS.",
                    "max_new_tokens": 1000,
                    "flow_cfg_scale": 0.7,
                    "exaggeration": 0.5,
                    "temperature": 0.8,
                    "cfg_weight": 0.5,
                    "repetition_penalty": 1.2,
                    "min_p": 0.05,
                    "top_p": 1.0,
                    "seed": 0,
                    "use_watermark": False,
                }

                # Start with defaults
                inputs.update(chatterbox_defaults)

                # Try to map old widget values if present
                # Old format: [model, text, max_len, cfg_temp, temp, top_p, top_k, exaggeration, speed, seed, ...]
                if len(widget_values) >= 2:
                    if isinstance(widget_values[0], str):
                        inputs["model_pack_name"] = widget_values[0]
                    if isinstance(widget_values[1], str):
                        inputs["text"] = widget_values[1]
            elif node_type == "LoadAudio":
                # LoadAudio: audio file path
                if widget_values:
                    inputs["audio"] = widget_values[0]
            elif node_type == "SaveAudio":
                # SaveAudio: filename_prefix
                if widget_values:
                    inputs["filename_prefix"] = widget_values[0]

            api_prompt[node_id] = {"class_type": node_type, "inputs": inputs}

        return api_prompt

    def _prepare_workflow(self, text: str, workflow: dict = None) -> dict:
        """
        Prepare the workflow with the given text input.

        For ChatterboxTTS: injects text into the 'text' field of ChatterboxTTS nodes.

        Args:
            text: Text to synthesize
            workflow: Workflow template (uses cached if not provided)

        Returns:
            Modified workflow dict ready for execution (wrapped in {"prompt": ...})
        """
        if workflow is None:
            workflow = self.load_workflow()

        if workflow is None:
            log.warning("No workflow template - TTS cannot proceed.")
            return None

        # Deep copy to avoid modifying the template
        import copy

        prepared = copy.deepcopy(workflow)

        # Text injection patterns for various TTS nodes
        text_input_keys = [
            "text",
            "input_text",
            "prompt",
            "tts_text",
            "speech_text",
            "content",
            "message",
        ]

        text_injected = False
        for node_id, node_data in prepared.items():
            if isinstance(node_data, dict):
                class_type = node_data.get("class_type", "")
                inputs = node_data.get("inputs", {})

                # Inject text into ChatterboxTTS or similar TTS nodes
                if "TTS" in class_type or "Chatterbox" in class_type:
                    if "text" in inputs:
                        inputs["text"] = text
                        text_injected = True
                        log.debug(f"Injected text into {class_type} node {node_id}")
                else:
                    # Try generic text input keys
                    for key in text_input_keys:
                        if key in inputs and isinstance(inputs[key], str):
                            inputs[key] = text
                            text_injected = True
                            log.debug(f"Injected text into node {node_id}, field {key}")
                            break

        if not text_injected:
            log.warning(
                "Could not find text input field in workflow. TTS may not work correctly."
            )

        # Return in API format
        return {"prompt": prepared}

    async def queue_tts(
        self,
        text: str,
        priority: int = None,
        cycle_id: int = 0,
        metadata: Dict[str, Any] = None,
    ) -> TTSRequest:
        """
        Queue a TTS request.

        Args:
            text: Text to synthesize
            priority: Request priority (higher = more important)
            cycle_id: The game cycle this request is from (for pruning)
            metadata: Optional context data (e.g. reply info)
        """
        if priority is None:
            priority = self.PRIORITY_CHAT_RESPONSE

        request = TTSRequest(
            text=text,
            request_id=str(uuid.uuid4())[:8],
            priority=priority,
            created_at=time.time(),
            cycle_id=cycle_id,
            metadata=metadata,
        )

        self._queue.append(request)
        # Sort by priority (highest first)
        self._queue.sort(key=lambda r: -r.priority)

        log.info(
            f"🔊 Queued TTS (priority={priority}, cycle={cycle_id}): {text[:50]}..."
        )

        return request

    # Maximum queue size for chat responses (persists across cycles)
    MAX_QUEUE_SIZE = 4  # Up to 3 message responses + 1 commentary

    async def queue_and_start_synthesis(
        self,
        text: str,
        priority: int = None,
        cycle_id: int = 0,
        metadata: Dict[str, Any] = None,
    ) -> Optional[TTSRequest]:
        """
        Queue a TTS request AND start background synthesis immediately.
        Does not wait for synthesis to complete - returns immediately.

        Use get_next_ready_audio() to retrieve completed audio.

        Args:
            text: Text to synthesize
            priority: Request priority
            cycle_id: The game cycle this request is from

        Returns:
            TTSRequest with synthesis_task started, or None if queue full
        """
        # Prune old items first
        self.prune_old_items(current_cycle=cycle_id)

        # Check queue size limit
        pending = [r for r in self._queue if not r.completed]
        if len(pending) >= self.MAX_QUEUE_SIZE:
            log.info(
                f"⏳ TTS queue full ({len(pending)}/{self.MAX_QUEUE_SIZE}), skipping"
            )
            return None

        # Queue the request
        request = await self.queue_tts(text, priority, cycle_id, metadata)

        # Start synthesis in background
        async def _synthesize():
            try:
                audio_path = await self.synthesize_speech(request.text)
                request.audio_path = audio_path
                request.completed = True
                log.info(
                    f"✅ Background TTS ready: {request.request_id} -> {audio_path}"
                )
            except Exception as e:
                request.error = str(e)
                request.completed = True
                log.error(f"❌ Background TTS failed: {request.request_id}: {e}")

        request.synthesis_task = asyncio.create_task(_synthesize())
        log.info(f"🚀 Started background synthesis: {request.request_id}")

        return request

    def get_next_ready_audio(self) -> Optional[TTSRequest]:
        """
        Get the next completed TTS request that has audio ready.
        Removes it from the queue.

        Returns:
            TTSRequest with audio_path populated, or None if none ready
        """
        for i, request in enumerate(self._queue):
            if request.completed and request.audio_path and not request.error:
                self._queue.pop(i)
                log.info(f"🎵 Returning ready audio: {request.request_id}")
                return request
        return None

    def prune_old_items(self, current_cycle: int) -> int:
        """
        Remove old TTS items from previous cycles (keep max 1 prev cycle).
        Also enforces MAX_QUEUE_SIZE.

        Args:
            current_cycle: The current game cycle number

        Returns:
            Number of items pruned
        """
        original_count = len(self._queue)

        # Keep items from current cycle and previous cycle only
        min_cycle = max(0, current_cycle - 1)
        self._queue = [r for r in self._queue if r.cycle_id >= min_cycle]

        # Also enforce max size (keep newest)
        if len(self._queue) > self.MAX_QUEUE_SIZE:
            # Cancel synthesis tasks for items we're removing
            for request in self._queue[self.MAX_QUEUE_SIZE :]:
                if request.synthesis_task and not request.synthesis_task.done():
                    request.synthesis_task.cancel()
            self._queue = self._queue[: self.MAX_QUEUE_SIZE]

        pruned = original_count - len(self._queue)
        if pruned > 0:
            log.info(f"♻️ Pruned {pruned} old TTS items (keeping cycle >= {min_cycle})")

        return pruned

    def cancel_pending_chat_responses(self) -> int:
        """
        Cancel all pending chat response synthesis (for game commentary priority).
        Keeps the queue but cancels in-progress synthesis tasks.

        Returns:
            Number of tasks cancelled
        """
        cancelled = 0
        for request in self._queue:
            if request.priority < self.PRIORITY_COMMENTARY:  # Only chat responses
                if request.synthesis_task and not request.synthesis_task.done():
                    request.synthesis_task.cancel()
                    cancelled += 1

        if cancelled > 0:
            log.info(f"🔇 Cancelled {cancelled} pending chat TTS synthesis tasks")

        return cancelled

    def get_queue_status(self) -> dict:
        """Get queue status for debugging."""
        return {
            "total": len(self._queue),
            "pending": len([r for r in self._queue if not r.completed]),
            "ready": len([r for r in self._queue if r.completed and r.audio_path]),
            "errors": len([r for r in self._queue if r.error]),
        }

    async def play_ready_audio(self, request: TTSRequest, wait: bool = True) -> bool:
        """
        Play audio from a completed TTSRequest.

        Args:
            request: TTSRequest with audio_path populated
            wait: If True, wait for playback to complete

        Returns:
            True if playback completed successfully
        """
        if not request or not request.audio_path:
            log.warning("🔊 No audio path in request")
            return False

        audio_path = request.audio_path

        # Apply speed/pitch processing
        audio_path = self._process_audio_speed_pitch(audio_path)

        # Get duration for UI sync
        duration_ms = self._get_audio_duration_ms(audio_path)

        # Notify UI that playback is about to start
        if self.on_playback_start and duration_ms:
            try:
                log.info(
                    f"🔊 Calling on_playback_start: text={request.text[:30]}..., duration={duration_ms}ms, metadata={bool(request.metadata)}"
                )
                await self.on_playback_start(
                    request.text, duration_ms, request.metadata
                )
            except Exception as cb_err:
                log.warning(f"🔊 on_playback_start callback error: {cb_err}")

        # Play the audio
        if not self.play_audio_ephemeral(audio_path):
            log.warning("🔊 play_audio_ephemeral returned False")
            return False

        if wait:
            completed = await self.wait_for_playback()
            # Cleanup audio file after playback
            self._cleanup_audio_file(audio_path)
            return completed

        return True

    async def process_queue(self) -> Optional[TTSRequest]:
        """
        Process the next item in the TTS queue.

        Returns:
            Completed TTSRequest or None if queue is empty.
        """
        if not self._queue:
            return None

        if self._processing:
            log.debug("TTS already processing, skipping")
            return None

        self._processing = True
        request = self._queue.pop(0)

        try:
            audio_path = await self.synthesize_speech(request.text)
            request.audio_path = audio_path
            request.completed = True
            log.info(f"✅ TTS complete: {request.request_id} -> {audio_path}")
        except Exception as e:
            request.error = str(e)
            log.error(f"❌ TTS failed for {request.request_id}: {e}")
        finally:
            self._processing = False

        return request

    async def synthesize_speech(self, text: str) -> Optional[str]:
        """
        Synthesize speech from text using ComfyUI.

        Args:
            text: Text to convert to speech

        Returns:
            Path to generated audio file, or None if failed.
        """
        if not self.is_available:
            log.warning("TTS service not available")
            return None

        # Check connection
        if not await self.check_connection():
            log.error("ComfyUI server not reachable")
            return None

        # Prepare workflow
        workflow = self._prepare_workflow(text)

        if workflow is None:
            log.error("Failed to prepare TTS workflow")
            return None

        try:
            client = await self._get_client()

            # Queue the workflow - _prepare_workflow already returns {"prompt": ...} format
            prompt_response = await client.post(
                f"{self.base_url}/prompt", json=workflow
            )

            if prompt_response.status_code != 200:
                log.error(f"Failed to queue workflow: {prompt_response.text}")
                return None

            prompt_result = prompt_response.json()
            prompt_id = prompt_result.get("prompt_id")

            if not prompt_id:
                log.error("No prompt_id in response")
                return None

            log.info(f"Queued ComfyUI workflow: {prompt_id}")

            # Poll for completion
            audio_path = await self._wait_for_completion(prompt_id)
            return audio_path

        except Exception as e:
            log.error(f"TTS synthesis error: {e}")
            return None

    async def _wait_for_completion(
        self, prompt_id: str, poll_interval: float = 0.5, max_wait: float = 60.0
    ) -> Optional[str]:
        """
        Wait for a ComfyUI workflow to complete and download the output audio.

        Args:
            prompt_id: The prompt ID to monitor
            poll_interval: Seconds between status checks
            max_wait: Maximum seconds to wait

        Returns:
            Path to downloaded audio file, or None if failed/timeout.
        """
        client = await self._get_client()
        start_time = time.time()

        while time.time() - start_time < max_wait:
            try:
                # Check history for completion
                history_response = await client.get(
                    f"{self.base_url}/history/{prompt_id}"
                )

                if history_response.status_code == 200:
                    history = history_response.json()

                    if prompt_id in history:
                        prompt_history = history[prompt_id]
                        outputs = prompt_history.get("outputs", {})

                        # Check if we have outputs (indicates completion)
                        if outputs:
                            log.debug(
                                f"Workflow completed, outputs: {list(outputs.keys())}"
                            )

                            # Look for audio output in any node
                            for node_id, node_output in outputs.items():
                                # Handle both 'audio' and 'audios' keys
                                audio_list = node_output.get(
                                    "audio"
                                ) or node_output.get("audios", [])
                                if not isinstance(audio_list, list):
                                    audio_list = [audio_list]

                                for audio_data in audio_list:
                                    if isinstance(audio_data, dict):
                                        filename = audio_data.get("filename")
                                        subfolder = audio_data.get("subfolder", "")
                                        audio_type = audio_data.get("type", "output")

                                        if filename:
                                            # Download the audio file from ComfyUI server
                                            local_path = await self._download_audio(
                                                filename, subfolder, audio_type
                                            )
                                            if local_path:
                                                return local_path

                            # No audio found in outputs
                            log.warning(
                                f"Workflow completed but no audio output found. Outputs: {outputs}"
                            )
                            return None

                        # Check for error status
                        status_info = prompt_history.get("status", {})
                        if status_info.get("status_str") == "error":
                            messages = status_info.get("messages", [])
                            log.error(f"Workflow error: {messages}")
                            return None

                await asyncio.sleep(poll_interval)

            except Exception as e:
                log.warning(f"Error checking workflow status: {e}")
                await asyncio.sleep(poll_interval)

        log.warning(f"Timeout waiting for workflow {prompt_id}")
        return None

    async def _download_audio(
        self, filename: str, subfolder: str = "", file_type: str = "output"
    ) -> Optional[str]:
        """
        Download audio file from ComfyUI server.

        Args:
            filename: Name of the audio file
            subfolder: Subfolder within output directory
            file_type: 'output' or 'input'

        Returns:
            Local path to downloaded file, or None if failed.
        """
        try:
            client = await self._get_client()

            # Build download URL
            params = {"filename": filename, "type": file_type}
            if subfolder:
                params["subfolder"] = subfolder

            download_url = f"{self.base_url}/view"
            log.info(f"Downloading audio from ComfyUI: {filename}")

            response = await client.get(download_url, params=params)

            if response.status_code == 200:
                # Save to local output directory
                local_filename = f"{uuid.uuid4().hex[:8]}_{filename}"
                local_path = os.path.join(self.output_dir, local_filename)

                with open(local_path, "wb") as f:
                    f.write(response.content)

                log.info(f"Downloaded audio to: {local_path}")

                # Apply speed/pitch adjustments if configured
                local_path = self._process_audio_speed_pitch(local_path)

                return local_path
            else:
                log.error(f"Failed to download audio: HTTP {response.status_code}")
                return None

        except Exception as e:
            log.error(f"Error downloading audio: {e}")
            return None

    async def clear_queue(self):
        """Clear all pending TTS requests."""
        cleared = len(self._queue)
        self._queue.clear()
        log.info(f"Cleared {cleared} pending TTS requests")

    async def synthesize_and_play(
        self,
        text: str,
        priority: int = None,
        wait: bool = True,
        metadata: Dict[str, Any] = None,
    ) -> bool:
        """
        Synthesize speech and play it ephemerally (no file retention needed by caller).

        This is the main method for TTS playback - handles synthesis, playback,
        and waiting for completion.

        Args:
            text: Text to synthesize
            priority: Priority level (PRIORITY_COMMENTARY or PRIORITY_CHAT_RESPONSE)
            wait: If True, wait for playback to complete before returning
            metadata: Optional context data (e.g. reply_to info for UI display)

        Returns:
            True if synthesis and playback started successfully
        """
        if not self.is_available:
            log.warning("TTS service not available")
            return False

        if priority is None:
            priority = self.PRIORITY_CHAT_RESPONSE

        # For high-priority commentary, cancel EVERYTHING first to ensure it plays immediately
        if priority >= self.PRIORITY_COMMENTARY:
            cleared = self.cancel_all_and_interrupt()
            if cleared > 0:
                log.info(f"🔊 Cleared {cleared} queue items for priority commentary")

        # Acquire lock to ensure TTS requests are serialized (no overlap)
        async with self._tts_lock:
            # Reset cancelled flag for new request
            self._cancelled = False
            total_start = time.time()

            # Log the request
            log.info(
                f"🔊 TTS START: synthesizing (priority={priority}): {text[:50]}..."
            )

            try:
                # Synthesize speech
                synthesis_start = time.time()
                audio_path = await self.synthesize_speech(text)
                synthesis_time = time.time() - synthesis_start

                if not audio_path:
                    log.warning(
                        f"🔊 TTS FAILED: synthesis returned no audio path after {synthesis_time:.2f}s"
                    )
                    return False

                log.info(
                    f"🔊 TTS Synthesis complete: {synthesis_time:.2f}s -> {audio_path}"
                )

                # Check if cancelled during synthesis
                if self._cancelled:
                    log.info("🔇 TTS cancelled during synthesis")
                    return False

                # Get audio duration for UI sync
                duration_ms = self._get_audio_duration_ms(audio_path)

                # Notify UI that playback is about to start (for typewriter sync)
                if self.on_playback_start and duration_ms:
                    try:
                        log.info(
                            f"🔊 Calling on_playback_start callback: text={text[:30]}..., duration={duration_ms}ms, metadata={bool(metadata)}"
                        )
                        await self.on_playback_start(text, duration_ms, metadata)
                    except Exception as cb_err:
                        log.warning(f"🔊 on_playback_start callback error: {cb_err}")

                # Play the audio
                playback_start = time.time()
                if not self.play_audio_ephemeral(audio_path):
                    log.warning("🔊 TTS FAILED: play_audio_ephemeral returned False")
                    return False

                # Wait for playback if requested
                if wait:
                    completed = await self.wait_for_playback()
                    playback_time = time.time() - playback_start
                    total_time = time.time() - total_start
                    log.info(
                        f"🔊 TTS COMPLETE: synthesis={synthesis_time:.2f}s, playback={playback_time:.2f}s, total={total_time:.2f}s"
                    )
                    # Cleanup audio file after playback
                    self._cleanup_audio_file(audio_path)
                    return completed

                total_time = time.time() - total_start
                log.info(
                    f"🔊 TTS STARTED PLAYBACK: synthesis={synthesis_time:.2f}s, total_so_far={total_time:.2f}s (not waiting)"
                )
                return True

            except Exception as e:
                total_time = time.time() - total_start
                log.error(f"🔊 TTS ERROR after {total_time:.2f}s: {e}")
                return False

    async def synthesize_only(self, text: str) -> Optional[str]:
        """
        Synthesize TTS audio without playing it.
        Used for preloading audio during countdown.

        Args:
            text: Text to synthesize

        Returns:
            Path to generated audio file, or None if failed.
        """
        if not self.is_available:
            log.warning("TTS service not available for preload")
            return None

        log.info(f"🔊 TTS PRELOAD: synthesizing (no playback): {text[:50]}...")

        try:
            audio_path = await self.synthesize_speech(text)
            if audio_path:
                log.info(f"🔊 TTS PRELOAD complete: {audio_path}")
            return audio_path
        except Exception as e:
            log.error(f"🔊 TTS PRELOAD failed: {e}")
            return None

    async def play_preloaded(
        self, audio_path: str, text: str, wait: bool = True
    ) -> bool:
        """
        Play a preloaded audio file with UI callbacks.

        Args:
            audio_path: Path to the preloaded audio file
            text: Original text for callbacks
            wait: If True, wait for playback to complete

        Returns:
            True if playback completed successfully
        """
        if not audio_path or not os.path.exists(audio_path):
            log.warning(f"🔊 Preloaded audio not found: {audio_path}")
            return False

        total_start = time.time()

        # Apply speed/pitch processing
        audio_path = self._process_audio_speed_pitch(audio_path)

        # Get duration for UI sync
        duration_ms = self._get_audio_duration_ms(audio_path)

        # Notify UI that playback is about to start
        if self.on_playback_start and duration_ms:
            try:
                log.info(
                    f"🔊 Calling on_playback_start: text={text[:30]}..., duration={duration_ms}ms"
                )
                await self.on_playback_start(text, duration_ms)
            except Exception as cb_err:
                log.warning(f"🔊 on_playback_start callback error: {cb_err}")

        # Play the audio
        log.info(f"🔊 Playing preloaded audio: {os.path.basename(audio_path)}")
        if not self.play_audio_ephemeral(audio_path):
            log.warning("🔊 play_audio_ephemeral returned False")
            return False

        if wait:
            playback_start = time.time()
            completed = await self.wait_for_playback()
            playback_time = time.time() - playback_start
            total_time = time.time() - total_start

            # Cleanup audio file after playback
            self._cleanup_audio_file(audio_path)

            log.info(
                f"🔊 TTS PRELOADED PLAY COMPLETE: playback={playback_time:.2f}s, total={total_time:.2f}s"
            )
            return completed

        return True

    async def close(self):
        """Close the HTTP client."""
        if self._client:
            await self._client.aclose()
            self._client = None

    def get_stats(self) -> dict:
        """Get service statistics."""
        return {
            "available": self.is_available,
            "queue_size": len(self._queue),
            "processing": self._processing,
            "base_url": self.base_url,
        }

Attributes

is_available property

Check if TTS service is configured.

Functions

__init__(base_url=None, workflow_path=None, output_dir=None, timeout=10.0, on_playback_start=None, audio_speed=None, audio_pitch=None)

Initialize the ComfyUI TTS service.

Parameters:

Name	Type	Description	Default
base_url	str	ComfyUI server URL (e.g., "http://localhost:8188")	None
workflow_path	str	Path to the TTS workflow JSON file	None
output_dir	str	Directory to save generated audio files	None
timeout	float	Request timeout in seconds	10.0
on_playback_start	callable	Callback(text, duration_ms) called when audio starts playing	None
audio_speed	float	Playback speed multiplier (e.g., 1.1 = 10% faster)	None
audio_pitch	float	Pitch shift in semitones (e.g., 2 = 2 semitones higher)	None

Source code in services/comfyui_tts_service.py

def __init__(
    self,
    base_url: str = None,
    workflow_path: str = None,
    output_dir: str = None,
    timeout: float = 10.0,  # Reduced from 60s for faster fallback
    on_playback_start: callable = None,  # (text, duration, metadata) -> Awaitable
    audio_speed: float = None,
    audio_pitch: float = None,
):
    """
    Initialize the ComfyUI TTS service.

    Args:
        base_url: ComfyUI server URL (e.g., "http://localhost:8188")
        workflow_path: Path to the TTS workflow JSON file
        output_dir: Directory to save generated audio files
        timeout: Request timeout in seconds
        on_playback_start: Callback(text, duration_ms) called when audio starts playing
        audio_speed: Playback speed multiplier (e.g., 1.1 = 10% faster)
        audio_pitch: Pitch shift in semitones (e.g., 2 = 2 semitones higher)
    """
    self.base_url = (
        base_url or os.getenv("COMFYUI_URL", "http://localhost:8188")
    ).rstrip("/")
    self.workflow_path = workflow_path or os.getenv("COMFYUI_TTS_WORKFLOW", "")
    self.output_dir = output_dir or os.path.join(os.getcwd(), "outputs", "audio")
    self.timeout = timeout
    self.on_playback_start = on_playback_start

    # Basic auth credentials (optional - for hosted ComfyUI instances)
    self._auth_username = os.getenv("COMFYUI_USERNAME", "")
    self._auth_password = os.getenv("COMFYUI_PASSWORD", "")
    self._auth: Optional[httpx.BasicAuth] = None
    if self._auth_username and self._auth_password:
        self._auth = httpx.BasicAuth(self._auth_username, self._auth_password)
        log.info(
            f"🔐 ComfyUI basic auth configured for user: {self._auth_username}"
        )

    # Audio post-processing settings (from env or args)
    self.audio_speed = (
        audio_speed
        if audio_speed is not None
        else float(os.getenv("TTS_AUDIO_SPEED", "1.0"))
    )
    self.audio_pitch = (
        audio_pitch
        if audio_pitch is not None
        else float(os.getenv("TTS_AUDIO_PITCH", "0"))
    )  # In semitones

    # Request queue (priority queue simulation with sorting)
    self._queue: List[TTSRequest] = []
    self._processing = False
    self._client: Optional[httpx.AsyncClient] = None

    # Request tracking
    self._current_request: Optional[TTSRequest] = None
    self._current_audio_path: Optional[str] = (
        None  # Track for post-playback cleanup
    )
    self._cancelled = False

    # Lock to ensure TTS requests are serialized (no overlap)
    self._tts_lock = asyncio.Lock()

    # Cached workflow template
    self._workflow_template: Optional[dict] = None

    # Initialize audio components
    self.audio_player = AudioPlayer()
    self.audio_processor = AudioProcessor()

    # Create output directory
    os.makedirs(self.output_dir, exist_ok=True)

    # Cleanup stale audio files from previous runs
    self._cleanup_stale_audio()

    # Check if configured and enabled
    # TTS_ENABLED defaults to true if not set, for backward compatibility
    tts_enabled = os.getenv("TTS_ENABLED", "true").lower() in ("true", "1", "yes")
    self._is_configured = bool(self.base_url) and tts_enabled

    if not tts_enabled:
        log.info("🔇 TTS disabled via TTS_ENABLED=false in .env")
    elif not self._is_configured:
        log.warning("ComfyUI TTS not configured. Set COMFYUI_URL in .env")

cancel_all_and_interrupt()

Cancel EVERYTHING - current playback, queue, and ComfyUI workflows. Use this when new high-priority commentary arrives to ensure it plays immediately.

Returns:

Type	Description
int	Number of queue items cleared

Source code in services/comfyui_tts_service.py

def cancel_all_and_interrupt(self) -> int:
    """
    Cancel EVERYTHING - current playback, queue, and ComfyUI workflows.
    Use this when new high-priority commentary arrives to ensure it plays immediately.

    Returns:
        Number of queue items cleared
    """
    # 1. Stop any current playback immediately
    self._cancelled = True
    self.audio_player.stop_playback()

    # 2. Cancel all pending synthesis tasks
    cancelled_tasks = 0
    for request in self._queue:
        if request.synthesis_task and not request.synthesis_task.done():
            request.synthesis_task.cancel()
            cancelled_tasks += 1

    # 3. Clear the entire queue
    queue_count = len(self._queue)
    self._queue.clear()
    self._processing = False

    # 4. Interrupt any pending ComfyUI workflow
    try:
        import httpx

        with httpx.Client(timeout=2.0, auth=self._auth) as client:
            response = client.post(f"{self.base_url}/interrupt")
            if response.status_code == 200:
                log.info("🔇 Interrupted pending ComfyUI TTS workflow")
    except Exception as e:
        log.debug(f"Could not interrupt ComfyUI workflow: {e}")

    if queue_count > 0 or cancelled_tasks > 0:
        log.info(
            f"🧹 CANCEL ALL: cleared {queue_count} queue items, cancelled {cancelled_tasks} synthesis tasks"
        )

    return queue_count

cancel_current()

Cancel the currently playing audio and any pending ComfyUI workflow.

Source code in services/comfyui_tts_service.py

def cancel_current(self):
    """Cancel the currently playing audio and any pending ComfyUI workflow."""
    self._cancelled = True

    # Stop audio player
    self.audio_player.stop_playback()

    # Also cancel any pending ComfyUI workflow via /interrupt endpoint
    try:
        import httpx

        with httpx.Client(timeout=2.0, auth=self._auth) as client:
            response = client.post(f"{self.base_url}/interrupt")
            if response.status_code == 200:
                log.info("🔇 Cancelled pending ComfyUI TTS workflow")
            else:
                log.debug(f"ComfyUI interrupt returned {response.status_code}")
    except Exception as e:
        log.debug(f"Could not interrupt ComfyUI workflow: {e}")

cancel_pending_chat_responses()

Cancel all pending chat response synthesis (for game commentary priority). Keeps the queue but cancels in-progress synthesis tasks.

Returns:

Type	Description
int	Number of tasks cancelled

Source code in services/comfyui_tts_service.py

def cancel_pending_chat_responses(self) -> int:
    """
    Cancel all pending chat response synthesis (for game commentary priority).
    Keeps the queue but cancels in-progress synthesis tasks.

    Returns:
        Number of tasks cancelled
    """
    cancelled = 0
    for request in self._queue:
        if request.priority < self.PRIORITY_COMMENTARY:  # Only chat responses
            if request.synthesis_task and not request.synthesis_task.done():
                request.synthesis_task.cancel()
                cancelled += 1

    if cancelled > 0:
        log.info(f"🔇 Cancelled {cancelled} pending chat TTS synthesis tasks")

    return cancelled

check_connection() async

Check if ComfyUI server is reachable. Returns True if connected, False otherwise.

Source code in services/comfyui_tts_service.py

async def check_connection(self) -> bool:
    """
    Check if ComfyUI server is reachable.
    Returns True if connected, False otherwise.
    """
    try:
        client = await self._get_client()
        response = await client.get(f"{self.base_url}/system_stats")
        return response.status_code == 200
    except Exception as e:
        log.warning(f"ComfyUI connection check failed: {e}")
        return False

clear_queue() async

Clear all pending TTS requests.

Source code in services/comfyui_tts_service.py

async def clear_queue(self):
    """Clear all pending TTS requests."""
    cleared = len(self._queue)
    self._queue.clear()
    log.info(f"Cleared {cleared} pending TTS requests")

close() async

Close the HTTP client.

Source code in services/comfyui_tts_service.py

async def close(self):
    """Close the HTTP client."""
    if self._client:
        await self._client.aclose()
        self._client = None

get_next_ready_audio()

Get the next completed TTS request that has audio ready. Removes it from the queue.

Returns:

Type	Description
Optional[TTSRequest]	TTSRequest with audio_path populated, or None if none ready

Source code in services/comfyui_tts_service.py

def get_next_ready_audio(self) -> Optional[TTSRequest]:
    """
    Get the next completed TTS request that has audio ready.
    Removes it from the queue.

    Returns:
        TTSRequest with audio_path populated, or None if none ready
    """
    for i, request in enumerate(self._queue):
        if request.completed and request.audio_path and not request.error:
            self._queue.pop(i)
            log.info(f"🎵 Returning ready audio: {request.request_id}")
            return request
    return None

get_queue_status()

Get queue status for debugging.

Source code in services/comfyui_tts_service.py

def get_queue_status(self) -> dict:
    """Get queue status for debugging."""
    return {
        "total": len(self._queue),
        "pending": len([r for r in self._queue if not r.completed]),
        "ready": len([r for r in self._queue if r.completed and r.audio_path]),
        "errors": len([r for r in self._queue if r.error]),
    }

get_stats()

Get service statistics.

Source code in services/comfyui_tts_service.py

def get_stats(self) -> dict:
    """Get service statistics."""
    return {
        "available": self.is_available,
        "queue_size": len(self._queue),
        "processing": self._processing,
        "base_url": self.base_url,
    }

load_workflow()

Load the TTS workflow template from file. Converts ComfyUI export format (nodes array) to API format (prompt dict).

Source code in services/comfyui_tts_service.py

def load_workflow(self) -> Optional[dict]:
    """
    Load the TTS workflow template from file.
    Converts ComfyUI export format (nodes array) to API format (prompt dict).
    """
    if self._workflow_template:
        return self._workflow_template

    if not self.workflow_path or not os.path.exists(self.workflow_path):
        log.warning(f"TTS workflow not found: {self.workflow_path}")
        return None

    try:
        with open(self.workflow_path, "r") as f:
            raw_workflow = json.load(f)
        log.info(f"Loaded TTS workflow from: {self.workflow_path}")

        # Check if this is export format (has 'nodes' array) vs API format (has 'prompt' dict)
        if "nodes" in raw_workflow:
            # Convert from export format to API format
            api_format = self._convert_export_to_api_format(raw_workflow)
            self._workflow_template = api_format
            log.info(
                f"Converted workflow from export format to API format ({len(api_format)} nodes)"
            )
        else:
            self._workflow_template = raw_workflow

        return self._workflow_template
    except Exception as e:
        log.error(f"Failed to load TTS workflow: {e}")
        return None

play_audio_ephemeral(audio_path)

Play audio file ephemerally (no file retention).

Source code in services/comfyui_tts_service.py

def play_audio_ephemeral(self, audio_path: str) -> bool:
    """Play audio file ephemerally (no file retention)."""
    if self.audio_speed != 1.0 or self.audio_pitch != 0:
        audio_path = self._process_audio_speed_pitch(audio_path)

    return self.audio_player.play_audio(audio_path)

play_preloaded(audio_path, text, wait=True) async

Play a preloaded audio file with UI callbacks.

Parameters:

Name	Type	Description	Default
audio_path	str	Path to the preloaded audio file	required
text	str	Original text for callbacks	required
wait	bool	If True, wait for playback to complete	True

Returns:

Type	Description
bool	True if playback completed successfully

Source code in services/comfyui_tts_service.py

async def play_preloaded(
    self, audio_path: str, text: str, wait: bool = True
) -> bool:
    """
    Play a preloaded audio file with UI callbacks.

    Args:
        audio_path: Path to the preloaded audio file
        text: Original text for callbacks
        wait: If True, wait for playback to complete

    Returns:
        True if playback completed successfully
    """
    if not audio_path or not os.path.exists(audio_path):
        log.warning(f"🔊 Preloaded audio not found: {audio_path}")
        return False

    total_start = time.time()

    # Apply speed/pitch processing
    audio_path = self._process_audio_speed_pitch(audio_path)

    # Get duration for UI sync
    duration_ms = self._get_audio_duration_ms(audio_path)

    # Notify UI that playback is about to start
    if self.on_playback_start and duration_ms:
        try:
            log.info(
                f"🔊 Calling on_playback_start: text={text[:30]}..., duration={duration_ms}ms"
            )
            await self.on_playback_start(text, duration_ms)
        except Exception as cb_err:
            log.warning(f"🔊 on_playback_start callback error: {cb_err}")

    # Play the audio
    log.info(f"🔊 Playing preloaded audio: {os.path.basename(audio_path)}")
    if not self.play_audio_ephemeral(audio_path):
        log.warning("🔊 play_audio_ephemeral returned False")
        return False

    if wait:
        playback_start = time.time()
        completed = await self.wait_for_playback()
        playback_time = time.time() - playback_start
        total_time = time.time() - total_start

        # Cleanup audio file after playback
        self._cleanup_audio_file(audio_path)

        log.info(
            f"🔊 TTS PRELOADED PLAY COMPLETE: playback={playback_time:.2f}s, total={total_time:.2f}s"
        )
        return completed

    return True

play_ready_audio(request, wait=True) async

Play audio from a completed TTSRequest.

Parameters:

Name	Type	Description	Default
request	TTSRequest	TTSRequest with audio_path populated	required
wait	bool	If True, wait for playback to complete	True

Returns:

Type	Description
bool	True if playback completed successfully

Source code in services/comfyui_tts_service.py

async def play_ready_audio(self, request: TTSRequest, wait: bool = True) -> bool:
    """
    Play audio from a completed TTSRequest.

    Args:
        request: TTSRequest with audio_path populated
        wait: If True, wait for playback to complete

    Returns:
        True if playback completed successfully
    """
    if not request or not request.audio_path:
        log.warning("🔊 No audio path in request")
        return False

    audio_path = request.audio_path

    # Apply speed/pitch processing
    audio_path = self._process_audio_speed_pitch(audio_path)

    # Get duration for UI sync
    duration_ms = self._get_audio_duration_ms(audio_path)

    # Notify UI that playback is about to start
    if self.on_playback_start and duration_ms:
        try:
            log.info(
                f"🔊 Calling on_playback_start: text={request.text[:30]}..., duration={duration_ms}ms, metadata={bool(request.metadata)}"
            )
            await self.on_playback_start(
                request.text, duration_ms, request.metadata
            )
        except Exception as cb_err:
            log.warning(f"🔊 on_playback_start callback error: {cb_err}")

    # Play the audio
    if not self.play_audio_ephemeral(audio_path):
        log.warning("🔊 play_audio_ephemeral returned False")
        return False

    if wait:
        completed = await self.wait_for_playback()
        # Cleanup audio file after playback
        self._cleanup_audio_file(audio_path)
        return completed

    return True

process_queue() async

Process the next item in the TTS queue.

Returns:

Type	Description
Optional[TTSRequest]	Completed TTSRequest or None if queue is empty.

Source code in services/comfyui_tts_service.py

async def process_queue(self) -> Optional[TTSRequest]:
    """
    Process the next item in the TTS queue.

    Returns:
        Completed TTSRequest or None if queue is empty.
    """
    if not self._queue:
        return None

    if self._processing:
        log.debug("TTS already processing, skipping")
        return None

    self._processing = True
    request = self._queue.pop(0)

    try:
        audio_path = await self.synthesize_speech(request.text)
        request.audio_path = audio_path
        request.completed = True
        log.info(f"✅ TTS complete: {request.request_id} -> {audio_path}")
    except Exception as e:
        request.error = str(e)
        log.error(f"❌ TTS failed for {request.request_id}: {e}")
    finally:
        self._processing = False

    return request

prune_old_items(current_cycle)

Remove old TTS items from previous cycles (keep max 1 prev cycle). Also enforces MAX_QUEUE_SIZE.

Parameters:

Name	Type	Description	Default
current_cycle	int	The current game cycle number	required

Returns:

Type	Description
int	Number of items pruned

Source code in services/comfyui_tts_service.py

def prune_old_items(self, current_cycle: int) -> int:
    """
    Remove old TTS items from previous cycles (keep max 1 prev cycle).
    Also enforces MAX_QUEUE_SIZE.

    Args:
        current_cycle: The current game cycle number

    Returns:
        Number of items pruned
    """
    original_count = len(self._queue)

    # Keep items from current cycle and previous cycle only
    min_cycle = max(0, current_cycle - 1)
    self._queue = [r for r in self._queue if r.cycle_id >= min_cycle]

    # Also enforce max size (keep newest)
    if len(self._queue) > self.MAX_QUEUE_SIZE:
        # Cancel synthesis tasks for items we're removing
        for request in self._queue[self.MAX_QUEUE_SIZE :]:
            if request.synthesis_task and not request.synthesis_task.done():
                request.synthesis_task.cancel()
        self._queue = self._queue[: self.MAX_QUEUE_SIZE]

    pruned = original_count - len(self._queue)
    if pruned > 0:
        log.info(f"♻️ Pruned {pruned} old TTS items (keeping cycle >= {min_cycle})")

    return pruned

queue_and_start_synthesis(text, priority=None, cycle_id=0, metadata=None) async

Queue a TTS request AND start background synthesis immediately. Does not wait for synthesis to complete - returns immediately.

Use get_next_ready_audio() to retrieve completed audio.

Parameters:

Name	Type	Description	Default
text	str	Text to synthesize	required
priority	int	Request priority	None
cycle_id	int	The game cycle this request is from	0

Returns:

Type	Description
Optional[TTSRequest]	TTSRequest with synthesis_task started, or None if queue full

Source code in services/comfyui_tts_service.py

async def queue_and_start_synthesis(
    self,
    text: str,
    priority: int = None,
    cycle_id: int = 0,
    metadata: Dict[str, Any] = None,
) -> Optional[TTSRequest]:
    """
    Queue a TTS request AND start background synthesis immediately.
    Does not wait for synthesis to complete - returns immediately.

    Use get_next_ready_audio() to retrieve completed audio.

    Args:
        text: Text to synthesize
        priority: Request priority
        cycle_id: The game cycle this request is from

    Returns:
        TTSRequest with synthesis_task started, or None if queue full
    """
    # Prune old items first
    self.prune_old_items(current_cycle=cycle_id)

    # Check queue size limit
    pending = [r for r in self._queue if not r.completed]
    if len(pending) >= self.MAX_QUEUE_SIZE:
        log.info(
            f"⏳ TTS queue full ({len(pending)}/{self.MAX_QUEUE_SIZE}), skipping"
        )
        return None

    # Queue the request
    request = await self.queue_tts(text, priority, cycle_id, metadata)

    # Start synthesis in background
    async def _synthesize():
        try:
            audio_path = await self.synthesize_speech(request.text)
            request.audio_path = audio_path
            request.completed = True
            log.info(
                f"✅ Background TTS ready: {request.request_id} -> {audio_path}"
            )
        except Exception as e:
            request.error = str(e)
            request.completed = True
            log.error(f"❌ Background TTS failed: {request.request_id}: {e}")

    request.synthesis_task = asyncio.create_task(_synthesize())
    log.info(f"🚀 Started background synthesis: {request.request_id}")

    return request

queue_tts(text, priority=None, cycle_id=0, metadata=None) async

Queue a TTS request.

Parameters:

Name	Type	Description	Default
text	str	Text to synthesize	required
priority	int	Request priority (higher = more important)	None
cycle_id	int	The game cycle this request is from (for pruning)	0
metadata	Dict[str, Any]	Optional context data (e.g. reply info)	None

Source code in services/comfyui_tts_service.py

async def queue_tts(
    self,
    text: str,
    priority: int = None,
    cycle_id: int = 0,
    metadata: Dict[str, Any] = None,
) -> TTSRequest:
    """
    Queue a TTS request.

    Args:
        text: Text to synthesize
        priority: Request priority (higher = more important)
        cycle_id: The game cycle this request is from (for pruning)
        metadata: Optional context data (e.g. reply info)
    """
    if priority is None:
        priority = self.PRIORITY_CHAT_RESPONSE

    request = TTSRequest(
        text=text,
        request_id=str(uuid.uuid4())[:8],
        priority=priority,
        created_at=time.time(),
        cycle_id=cycle_id,
        metadata=metadata,
    )

    self._queue.append(request)
    # Sort by priority (highest first)
    self._queue.sort(key=lambda r: -r.priority)

    log.info(
        f"🔊 Queued TTS (priority={priority}, cycle={cycle_id}): {text[:50]}..."
    )

    return request

synthesize_and_play(text, priority=None, wait=True, metadata=None) async

Synthesize speech and play it ephemerally (no file retention needed by caller).

This is the main method for TTS playback - handles synthesis, playback, and waiting for completion.

Parameters:

Name	Type	Description	Default
text	str	Text to synthesize	required
priority	int	Priority level (PRIORITY_COMMENTARY or PRIORITY_CHAT_RESPONSE)	None
wait	bool	If True, wait for playback to complete before returning	True
metadata	Dict[str, Any]	Optional context data (e.g. reply_to info for UI display)	None

Returns:

Type	Description
bool	True if synthesis and playback started successfully

Source code in services/comfyui_tts_service.py

async def synthesize_and_play(
    self,
    text: str,
    priority: int = None,
    wait: bool = True,
    metadata: Dict[str, Any] = None,
) -> bool:
    """
    Synthesize speech and play it ephemerally (no file retention needed by caller).

    This is the main method for TTS playback - handles synthesis, playback,
    and waiting for completion.

    Args:
        text: Text to synthesize
        priority: Priority level (PRIORITY_COMMENTARY or PRIORITY_CHAT_RESPONSE)
        wait: If True, wait for playback to complete before returning
        metadata: Optional context data (e.g. reply_to info for UI display)

    Returns:
        True if synthesis and playback started successfully
    """
    if not self.is_available:
        log.warning("TTS service not available")
        return False

    if priority is None:
        priority = self.PRIORITY_CHAT_RESPONSE

    # For high-priority commentary, cancel EVERYTHING first to ensure it plays immediately
    if priority >= self.PRIORITY_COMMENTARY:
        cleared = self.cancel_all_and_interrupt()
        if cleared > 0:
            log.info(f"🔊 Cleared {cleared} queue items for priority commentary")

    # Acquire lock to ensure TTS requests are serialized (no overlap)
    async with self._tts_lock:
        # Reset cancelled flag for new request
        self._cancelled = False
        total_start = time.time()

        # Log the request
        log.info(
            f"🔊 TTS START: synthesizing (priority={priority}): {text[:50]}..."
        )

        try:
            # Synthesize speech
            synthesis_start = time.time()
            audio_path = await self.synthesize_speech(text)
            synthesis_time = time.time() - synthesis_start

            if not audio_path:
                log.warning(
                    f"🔊 TTS FAILED: synthesis returned no audio path after {synthesis_time:.2f}s"
                )
                return False

            log.info(
                f"🔊 TTS Synthesis complete: {synthesis_time:.2f}s -> {audio_path}"
            )

            # Check if cancelled during synthesis
            if self._cancelled:
                log.info("🔇 TTS cancelled during synthesis")
                return False

            # Get audio duration for UI sync
            duration_ms = self._get_audio_duration_ms(audio_path)

            # Notify UI that playback is about to start (for typewriter sync)
            if self.on_playback_start and duration_ms:
                try:
                    log.info(
                        f"🔊 Calling on_playback_start callback: text={text[:30]}..., duration={duration_ms}ms, metadata={bool(metadata)}"
                    )
                    await self.on_playback_start(text, duration_ms, metadata)
                except Exception as cb_err:
                    log.warning(f"🔊 on_playback_start callback error: {cb_err}")

            # Play the audio
            playback_start = time.time()
            if not self.play_audio_ephemeral(audio_path):
                log.warning("🔊 TTS FAILED: play_audio_ephemeral returned False")
                return False

            # Wait for playback if requested
            if wait:
                completed = await self.wait_for_playback()
                playback_time = time.time() - playback_start
                total_time = time.time() - total_start
                log.info(
                    f"🔊 TTS COMPLETE: synthesis={synthesis_time:.2f}s, playback={playback_time:.2f}s, total={total_time:.2f}s"
                )
                # Cleanup audio file after playback
                self._cleanup_audio_file(audio_path)
                return completed

            total_time = time.time() - total_start
            log.info(
                f"🔊 TTS STARTED PLAYBACK: synthesis={synthesis_time:.2f}s, total_so_far={total_time:.2f}s (not waiting)"
            )
            return True

        except Exception as e:
            total_time = time.time() - total_start
            log.error(f"🔊 TTS ERROR after {total_time:.2f}s: {e}")
            return False

synthesize_only(text) async

Synthesize TTS audio without playing it. Used for preloading audio during countdown.

Parameters:

Name	Type	Description	Default
text	str	Text to synthesize	required

Returns:

Type	Description
Optional[str]	Path to generated audio file, or None if failed.

Source code in services/comfyui_tts_service.py

async def synthesize_only(self, text: str) -> Optional[str]:
    """
    Synthesize TTS audio without playing it.
    Used for preloading audio during countdown.

    Args:
        text: Text to synthesize

    Returns:
        Path to generated audio file, or None if failed.
    """
    if not self.is_available:
        log.warning("TTS service not available for preload")
        return None

    log.info(f"🔊 TTS PRELOAD: synthesizing (no playback): {text[:50]}...")

    try:
        audio_path = await self.synthesize_speech(text)
        if audio_path:
            log.info(f"🔊 TTS PRELOAD complete: {audio_path}")
        return audio_path
    except Exception as e:
        log.error(f"🔊 TTS PRELOAD failed: {e}")
        return None

synthesize_speech(text) async

Synthesize speech from text using ComfyUI.

Parameters:

Name	Type	Description	Default
text	str	Text to convert to speech	required

Returns:

Type	Description
Optional[str]	Path to generated audio file, or None if failed.

Source code in services/comfyui_tts_service.py

async def synthesize_speech(self, text: str) -> Optional[str]:
    """
    Synthesize speech from text using ComfyUI.

    Args:
        text: Text to convert to speech

    Returns:
        Path to generated audio file, or None if failed.
    """
    if not self.is_available:
        log.warning("TTS service not available")
        return None

    # Check connection
    if not await self.check_connection():
        log.error("ComfyUI server not reachable")
        return None

    # Prepare workflow
    workflow = self._prepare_workflow(text)

    if workflow is None:
        log.error("Failed to prepare TTS workflow")
        return None

    try:
        client = await self._get_client()

        # Queue the workflow - _prepare_workflow already returns {"prompt": ...} format
        prompt_response = await client.post(
            f"{self.base_url}/prompt", json=workflow
        )

        if prompt_response.status_code != 200:
            log.error(f"Failed to queue workflow: {prompt_response.text}")
            return None

        prompt_result = prompt_response.json()
        prompt_id = prompt_result.get("prompt_id")

        if not prompt_id:
            log.error("No prompt_id in response")
            return None

        log.info(f"Queued ComfyUI workflow: {prompt_id}")

        # Poll for completion
        audio_path = await self._wait_for_completion(prompt_id)
        return audio_path

    except Exception as e:
        log.error(f"TTS synthesis error: {e}")
        return None

wait_for_playback(timeout=30.0) async

Wait for current audio playback to complete.

Source code in services/comfyui_tts_service.py

async def wait_for_playback(self, timeout: float = 30.0) -> bool:
    """Wait for current audio playback to complete."""
    return await self.audio_player.wait_for_playback(timeout)

TTSRequest dataclass

Represents a pending TTS request with async synthesis support.

Source code in services/comfyui_tts_service.py

@dataclass
class TTSRequest:
    """Represents a pending TTS request with async synthesis support."""

    text: str
    request_id: str
    priority: int  # Higher = more important (game commentary > chat response)
    created_at: float
    cycle_id: int = 0  # Which game cycle this request came from
    audio_path: Optional[str] = None
    completed: bool = False
    error: Optional[str] = None
    synthesis_task: Optional[asyncio.Task] = None  # Background synthesis task
    metadata: Optional[Dict[str, Any]] = None  # Extra context (e.g. reply_to info)

Functions

create_tts_service(base_url=None, workflow_path=None, output_dir=None, on_playback_start=None)

Factory function to create a ComfyUITTSService instance.

Uses environment variables if parameters not provided.

Parameters:

Name	Type	Description	Default
on_playback_start	callable	Async callback(text, duration_ms) called when audio starts	None

Source code in services/comfyui_tts_service.py

def create_tts_service(
    base_url: str = None,
    workflow_path: str = None,
    output_dir: str = None,
    on_playback_start: callable = None,
) -> ComfyUITTSService:
    """
    Factory function to create a ComfyUITTSService instance.

    Uses environment variables if parameters not provided.

    Args:
        on_playback_start: Async callback(text, duration_ms) called when audio starts
    """
    return ComfyUITTSService(
        base_url=base_url,
        workflow_path=workflow_path,
        output_dir=output_dir,
        on_playback_start=on_playback_start,
    )

ComfyUI Vision Service

services.comfyui_vision_service

ComfyUI Vision Service for Pokemon LLM Agent. Triggers Vision workflows on a ComfyUI server to analyze images using models like Moondream.

Classes

ComfyUIVisionService

Vision service using ComfyUI as the backend.

Source code in services/comfyui_vision_service.py

class ComfyUIVisionService:
    """
    Vision service using ComfyUI as the backend.
    """

    def __init__(
        self,
        base_url: str = None,
        workflow_path: str = None,
        timeout: float = 30.0,
        node_id: str = "7",
    ):
        """
        Initialize the ComfyUI Vision service.

        Args:
            base_url: ComfyUI server URL (e.g., "http://localhost:8188")
            workflow_path: Path to the Vision workflow JSON file
            timeout: Request timeout in seconds
            node_id: ID of the node producing the text output
        """
        self.base_url = (
            base_url or os.getenv("COMFYUI_URL", "http://localhost:8188")
        ).rstrip("/")
        self.workflow_path = workflow_path or os.getenv(
            "COMFYUI_VISION_WORKFLOW", "workflows/lass-vision.json"
        )
        self.output_node_id = node_id or os.getenv("COMFYUI_VISION_NODE_ID", "7")
        self.timeout = timeout

        # Basic auth credentials
        self._auth_username = os.getenv("COMFYUI_USERNAME", "")
        self._auth_password = os.getenv("COMFYUI_PASSWORD", "")
        self._auth: Optional[httpx.BasicAuth] = None
        if self._auth_username and self._auth_password:
            self._auth = httpx.BasicAuth(self._auth_username, self._auth_password)
            log.info(
                f"🔐 ComfyUI vision auth configured for user: {self._auth_username}"
            )

        self._workflow_template: Optional[dict] = None

        # Check configuration
        self.enabled = os.getenv("VISION_PROVIDER", "ZAI") == "COMFYUI"
        if self.enabled:
            log.info(f"👁️ ComfyUI Vision Service ENABLED (Node {self.output_node_id})")
        else:
            log.info("👁️ ComfyUI Vision Service available but not active provider")

    def _create_client(self) -> httpx.AsyncClient:
        """Create a new client instance (caller must close or use context manager)."""
        return httpx.AsyncClient(timeout=self.timeout, auth=self._auth)

    async def check_connection(self) -> bool:
        try:
            async with self._create_client() as client:
                response = await client.get(f"{self.base_url}/system_stats")
                return response.status_code == 200
        except Exception as e:
            log.warning(f"ComfyUI vision connection check failed: {e}")
            return False

    def load_workflow(self) -> Optional[dict]:
        if self._workflow_template:
            return self._workflow_template

        if not self.workflow_path or not os.path.exists(self.workflow_path):
            log.warning(f"Vision workflow not found: {self.workflow_path}")
            return None

        try:
            with open(self.workflow_path, "r") as f:
                raw = json.load(f)

            # Simple API format check (if keys are node IDs)
            # The lass-vision.json is in API format (nodes list)?
            # Wait, the file I read started with {"id":..., "nodes": [...]}
            # That is the EXPORT format (Graph). API needs PROMPT format.
            # I need to convert it.

            if "nodes" in raw:
                # Need conversion
                self._workflow_template = self._convert_export_to_api_format(raw)
            else:
                # Assume already API format
                self._workflow_template = raw

            return self._workflow_template
        except Exception as e:
            log.error(f"Failed to load Vision workflow: {e}")
            return None

    def _convert_export_to_api_format(self, export_workflow: dict) -> dict:
        # Simplified conversion similar to TTS service
        nodes = export_workflow.get("nodes", [])
        links = export_workflow.get("links", [])

        link_map = {}
        for link in links:
            if len(link) >= 5:
                link_id, from_node, from_slot, to_node, to_slot, *_ = link
                link_map[link_id] = (from_node, from_slot)

        api_prompt = {}
        for node in nodes:
            node_id = str(node.get("id"))
            node_type = node.get("type")
            inputs = {}

            # Inputs from links
            for input_def in node.get("inputs", []):
                input_name = input_def.get("name")
                link_id = input_def.get("link")
                if link_id and link_id in link_map:
                    from_node, from_slot = link_map[link_id]
                    inputs[input_name] = [str(from_node), from_slot]

            # Widget values
            widgets_values = node.get("widgets_values", [])

            # Map known nodes
            if node_type == "LoadImage":
                if widgets_values:
                    inputs["image"] = widgets_values[0]
                    inputs["upload"] = "image"  # typical default
            elif node_type == "UnifiedVisionPromptGenerator":
                # widgets: [question, max_tokens, temp, seed, model, keep_loaded]
                # Map based on index if input names match?
                # Actually API format usually expects inputs dict with keys matching widget names if they are converted to inputs?
                # Or just pass them if they are simple values.
                # But UnifiedVisionPromptGenerator might expect inputs dict to contain values for widgets.
                # Let's verify mapping based on widget names in input definition if possible?
                # In export format, 'widgets_values' is a list. In API format, 'inputs' dict has keys.
                # I need to know the mapping.
                # Based on file read:
                # inputs: image, question, max_new_tokens...
                # question, max_new_tokens etc have "widget": {"name": ...}
                # So they correspond to widgets_values in order?
                pass

            # For simplicity, if we don't know the mapping, we might struggle.
            # But wait, TTS service mapping was manual.
            # Let's inspect the JSON again to see order.
            # UnifiedVisionPromptGenerator: ["Describe...", 1024, 0.7, 9014..., "randomize", "vikhyatk/moondream2", false]
            # Inputs: image (link), question (widget), max_new_tokens (widget), temperature (widget), seed (widget), model (widget), keep_model_loaded (widget)
            # So order matches.

            if node_type == "UnifiedVisionPromptGenerator":
                # Map widgets_values list to named inputs
                # Order from JSON: question, max_new_tokens, temperature, seed, model, keep_model_loaded
                # (Note: 'randomize' is usually control_after_generate, ignored in API execution usually?)
                # Wait, 'seed' widget value is int. 'control_after_generate' is string "randomize".
                # The JSON has `widgets_values`: ["Describe...", 1024, 0.7, 901..., "randomize", "vikhyatk/...", false]
                # indices: 0=question, 1=max_tokens, 2=temp, 3=seed, 4=control?, 5=model, 6=keep_loaded?
                # Actually ComfyUI API expects specific keys.

                inputs["question"] = widgets_values[0]
                inputs["max_new_tokens"] = widgets_values[1]
                inputs["temperature"] = widgets_values[2]
                inputs["seed"] = widgets_values[3]
                # skip 4 "randomize"
                inputs["model"] = widgets_values[5]
                inputs["keep_model_loaded"] = widgets_values[6]

            api_prompt[node_id] = {"class_type": node_type, "inputs": inputs}

        return {"prompt": api_prompt}

    async def upload_image(self, image_path: str) -> Optional[str]:
        """Upload image to ComfyUI and return filename."""
        if not os.path.exists(image_path):
            return None

        try:
            async with self._create_client() as client:
                with open(image_path, "rb") as f:
                    files = {"image": f}
                    response = await client.post(
                        f"{self.base_url}/upload/image", files=files
                    )

            if response.status_code == 200:
                data = response.json()
                # Returns: {"name": "filename.png", "subfolder": "", "type": "input"}
                return data.get("name")
            return None
        except Exception as e:
            log.error(f"Image upload failed: {e}")
            return None

    async def analyze_image(
        self, image_path: str, prompt_text: str = "Describe this image in detail."
    ) -> Optional[str]:
        """
        Analyze image using ComfyUI workflow.
        Returns description string.
        """
        if not self.enabled:
            return None

        # 1. Upload Image
        filename = await self.upload_image(image_path)
        if not filename:
            log.error("Failed to upload image to ComfyUI")
            return None

        # 2. Prepare Workflow
        workflow = self.load_workflow()
        if not workflow:
            return None

        import copy

        prompt = copy.deepcopy(workflow["prompt"])

        # Update inputs
        # Node 2: LoadImage
        # Node 1: UnifiedVisionPromptGenerator

        # Find LoadImage node (usually '2' based on our file)
        # But we should find by type to be safe? Or stick to ID since we converted it?
        # The IDs are preserved in conversion.

        if "2" in prompt:
            prompt["2"]["inputs"]["image"] = filename
        else:
            # Fallback search
            for nid, data in prompt.items():
                if data["class_type"] == "LoadImage":
                    data["inputs"]["image"] = filename
                    break

        # Update Prompt
        if self.output_node_id in prompt:
            prompt[self.output_node_id]["inputs"]["question"] = prompt_text

        # 3. Queue Prompt
        try:
            async with self._create_client() as client:
                response = await client.post(
                    f"{self.base_url}/prompt", json={"prompt": prompt}
                )

            if response.status_code != 200:
                log.error(f"Failed to queue vision workflow: {response.text}")
                return None

            prompt_id = response.json().get("prompt_id")
            if not prompt_id:
                return None

            log.info(f"Queued vision workflow: {prompt_id}")

            # 4. Wait for completion
            return await self._wait_for_result(prompt_id)

        except Exception as e:
            log.error(f"Vision analysis error: {e}")
            return None

    async def _wait_for_result(self, prompt_id: str) -> Optional[str]:
        async with self._create_client() as client:
            start = time.time()

            while time.time() - start < self.timeout:
                try:
                    # Check history
                    res = await client.get(f"{self.base_url}/history/{prompt_id}")
                    if res.status_code == 200:
                        history = res.json()
                        if prompt_id in history:
                            outputs = history[prompt_id].get("outputs", {})

                            # Look for our target node output
                            if self.output_node_id in outputs:
                                node_output = outputs[self.output_node_id]
                                # UnifiedVisionPromptGenerator outputs 'prompt' (STRING)
                                # API output structure: {"prompt": ["The description text"]}
                                # Usually values are lists
                                if "prompt" in node_output:
                                    val = node_output["prompt"]
                                    if isinstance(val, list) and len(val) > 0:
                                        return val[0]
                                    return str(val)

                                # If key isn't 'prompt', return first value
                                for k, v in node_output.items():
                                    if isinstance(v, list) and len(v) > 0:
                                        return v[0]
                                    return str(v)
                            else:
                                log.debug(
                                    f"Target node {self.output_node_id} not in outputs. Available nodes: {list(outputs.keys())}"
                                )

                                # Fallback: Check if ANY node has text output?
                                # Or check for 'text' or 'prompt' keys in any node
                                for nid, nout in outputs.items():
                                    for k, v in nout.items():
                                        if k in ["prompt", "text", "string", "output"]:
                                            val = (
                                                v[0]
                                                if isinstance(v, list) and v
                                                else str(v)
                                            )
                                            log.info(
                                                f"Found fallback output in node {nid} key {k}"
                                            )
                                            return val

                    await asyncio.sleep(0.5)
                except Exception as e:
                    log.warning(f"Error checking vision status: {e}")
                    await asyncio.sleep(1)

            log.error("Vision workflow timed out")
            return None

Functions

__init__(base_url=None, workflow_path=None, timeout=30.0, node_id='7')

Initialize the ComfyUI Vision service.

Parameters:

Name	Type	Description	Default
base_url	str	ComfyUI server URL (e.g., "http://localhost:8188")	None
workflow_path	str	Path to the Vision workflow JSON file	None
timeout	float	Request timeout in seconds	30.0
node_id	str	ID of the node producing the text output	'7'

Source code in services/comfyui_vision_service.py

def __init__(
    self,
    base_url: str = None,
    workflow_path: str = None,
    timeout: float = 30.0,
    node_id: str = "7",
):
    """
    Initialize the ComfyUI Vision service.

    Args:
        base_url: ComfyUI server URL (e.g., "http://localhost:8188")
        workflow_path: Path to the Vision workflow JSON file
        timeout: Request timeout in seconds
        node_id: ID of the node producing the text output
    """
    self.base_url = (
        base_url or os.getenv("COMFYUI_URL", "http://localhost:8188")
    ).rstrip("/")
    self.workflow_path = workflow_path or os.getenv(
        "COMFYUI_VISION_WORKFLOW", "workflows/lass-vision.json"
    )
    self.output_node_id = node_id or os.getenv("COMFYUI_VISION_NODE_ID", "7")
    self.timeout = timeout

    # Basic auth credentials
    self._auth_username = os.getenv("COMFYUI_USERNAME", "")
    self._auth_password = os.getenv("COMFYUI_PASSWORD", "")
    self._auth: Optional[httpx.BasicAuth] = None
    if self._auth_username and self._auth_password:
        self._auth = httpx.BasicAuth(self._auth_username, self._auth_password)
        log.info(
            f"🔐 ComfyUI vision auth configured for user: {self._auth_username}"
        )

    self._workflow_template: Optional[dict] = None

    # Check configuration
    self.enabled = os.getenv("VISION_PROVIDER", "ZAI") == "COMFYUI"
    if self.enabled:
        log.info(f"👁️ ComfyUI Vision Service ENABLED (Node {self.output_node_id})")
    else:
        log.info("👁️ ComfyUI Vision Service available but not active provider")

analyze_image(image_path, prompt_text='Describe this image in detail.') async

Analyze image using ComfyUI workflow. Returns description string.

Source code in services/comfyui_vision_service.py

async def analyze_image(
    self, image_path: str, prompt_text: str = "Describe this image in detail."
) -> Optional[str]:
    """
    Analyze image using ComfyUI workflow.
    Returns description string.
    """
    if not self.enabled:
        return None

    # 1. Upload Image
    filename = await self.upload_image(image_path)
    if not filename:
        log.error("Failed to upload image to ComfyUI")
        return None

    # 2. Prepare Workflow
    workflow = self.load_workflow()
    if not workflow:
        return None

    import copy

    prompt = copy.deepcopy(workflow["prompt"])

    # Update inputs
    # Node 2: LoadImage
    # Node 1: UnifiedVisionPromptGenerator

    # Find LoadImage node (usually '2' based on our file)
    # But we should find by type to be safe? Or stick to ID since we converted it?
    # The IDs are preserved in conversion.

    if "2" in prompt:
        prompt["2"]["inputs"]["image"] = filename
    else:
        # Fallback search
        for nid, data in prompt.items():
            if data["class_type"] == "LoadImage":
                data["inputs"]["image"] = filename
                break

    # Update Prompt
    if self.output_node_id in prompt:
        prompt[self.output_node_id]["inputs"]["question"] = prompt_text

    # 3. Queue Prompt
    try:
        async with self._create_client() as client:
            response = await client.post(
                f"{self.base_url}/prompt", json={"prompt": prompt}
            )

        if response.status_code != 200:
            log.error(f"Failed to queue vision workflow: {response.text}")
            return None

        prompt_id = response.json().get("prompt_id")
        if not prompt_id:
            return None

        log.info(f"Queued vision workflow: {prompt_id}")

        # 4. Wait for completion
        return await self._wait_for_result(prompt_id)

    except Exception as e:
        log.error(f"Vision analysis error: {e}")
        return None

upload_image(image_path) async

Upload image to ComfyUI and return filename.

Source code in services/comfyui_vision_service.py

async def upload_image(self, image_path: str) -> Optional[str]:
    """Upload image to ComfyUI and return filename."""
    if not os.path.exists(image_path):
        return None

    try:
        async with self._create_client() as client:
            with open(image_path, "rb") as f:
                files = {"image": f}
                response = await client.post(
                    f"{self.base_url}/upload/image", files=files
                )

        if response.status_code == 200:
            data = response.json()
            # Returns: {"name": "filename.png", "subfolder": "", "type": "input"}
            return data.get("name")
        return None
    except Exception as e:
        log.error(f"Image upload failed: {e}")
        return None

WebSocket Service

services.websocket_service

Functions

broadcast_message(message) async

Sends a JSON message to all connected clients.

Source code in services/websocket_service.py

async def broadcast_message(message):
    """Sends a JSON message to all connected clients."""
    if not connected_clients:
        return

    message_json = json.dumps(message)
    # Use create_task for better concurrency handling if many clients exist
    send_tasks = [asyncio.create_task(client.send(message_json)) for client in connected_clients]
    if not send_tasks:
        return

    results = await asyncio.gather(*send_tasks, return_exceptions=True)

    disconnected_clients = set()
    clients_list = list(connected_clients) # Create stable list for indexing
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            # Ensure index is valid before accessing clients_list
            if i < len(clients_list):
                client = clients_list[i]
                log.warning(f"WS: Failed to send to client {client.remote_address}: {result}. Removing.")
                disconnected_clients.add(client)
            else:
                 # This case should theoretically not happen with gather but added defensively
                 log.error(f"WS: Index {i} out of bounds for clients list of size {len(clients_list)} during error handling.")

    connected_clients.difference_update(disconnected_clients)

run_server_forever(app_state_dict) async

Starts the WebSocket server and keeps it running indefinitely.

Source code in services/websocket_service.py

async def run_server_forever(app_state_dict):
    """Starts the WebSocket server and keeps it running indefinitely."""

    # Define the handler that websockets.serve will call.
    # It captures app_state_dict from the outer scope (closure).
    async def handler_entrypoint(websocket):
        await _actual_handler_code(websocket, app_state_dict)

    async with websockets.serve(handler_entrypoint, "localhost", WEBSOCKET_PORT):
        log.info(f"WebSocket server running on ws://localhost:{WEBSOCKET_PORT}")
        await asyncio.Future() # Keep server running until cancelled

Screenshot Manager

services.screenshot_manager

Screenshot Manager for Zora Gallery Posts.

Maintains a rolling buffer of recent screenshots for creating gallery posts on the LLMLetsPlay Zora account. Unlike the ScreenshotHistoryTracker (which keeps only 2 screenshots for UI diff checking), this manager keeps 5 screenshots for creating rich gallery posts.

Screenshots are NOT deleted - they're just rotated out of the buffer. The actual cleanup is handled elsewhere.

Classes

ScreenshotEntry dataclass

A single screenshot with context metadata.

Source code in services/screenshot_manager.py

@dataclass
class ScreenshotEntry:
    """A single screenshot with context metadata."""

    path: str
    timestamp: float
    cycle_number: int
    action_number: int
    map_name: str = ""
    location_description: str = ""

    def exists(self) -> bool:
        """Check if the screenshot file still exists."""
        return os.path.exists(self.path)

    def to_dict(self) -> dict:
        """Convert to dictionary for serialization."""
        return {
            "path": self.path,
            "timestamp": self.timestamp,
            "cycle_number": self.cycle_number,
            "action_number": self.action_number,
            "map_name": self.map_name,
            "location_description": self.location_description,
        }

Functions

exists()

Check if the screenshot file still exists.

Source code in services/screenshot_manager.py

def exists(self) -> bool:
    """Check if the screenshot file still exists."""
    return os.path.exists(self.path)

to_dict()

Convert to dictionary for serialization.

Source code in services/screenshot_manager.py

def to_dict(self) -> dict:
    """Convert to dictionary for serialization."""
    return {
        "path": self.path,
        "timestamp": self.timestamp,
        "cycle_number": self.cycle_number,
        "action_number": self.action_number,
        "map_name": self.map_name,
        "location_description": self.location_description,
    }

ScreenshotManager

Manages a rolling buffer of recent screenshots for Zora gallery posts.

This is separate from ScreenshotHistoryTracker which handles UI diff checking. The ScreenshotManager: - Keeps up to 5 most recent screenshots - Copies screenshots to a dedicated gallery directory - Provides methods for creating gallery posts - Does NOT delete screenshots (just rotates buffer)

Usage

manager = ScreenshotManager(gallery_dir="data/zora_gallery")

Every cycle, capture the screenshot

manager.capture( source_path="/path/to/current_screenshot.png", cycle_number=42, action_number=123, map_name="PALLET_TOWN" )

When creating a Zora post

screenshots = manager.get_gallery_paths() # Returns up to 5 paths

Source code in services/screenshot_manager.py

class ScreenshotManager:
    """
    Manages a rolling buffer of recent screenshots for Zora gallery posts.

    This is separate from ScreenshotHistoryTracker which handles UI diff checking.
    The ScreenshotManager:
    - Keeps up to 5 most recent screenshots
    - Copies screenshots to a dedicated gallery directory
    - Provides methods for creating gallery posts
    - Does NOT delete screenshots (just rotates buffer)

    Usage:
        manager = ScreenshotManager(gallery_dir="data/zora_gallery")

        # Every cycle, capture the screenshot
        manager.capture(
            source_path="/path/to/current_screenshot.png",
            cycle_number=42,
            action_number=123,
            map_name="PALLET_TOWN"
        )

        # When creating a Zora post
        screenshots = manager.get_gallery_paths()  # Returns up to 5 paths
    """

    def __init__(
        self,
        gallery_dir: str = "data/zora_gallery",
        max_screenshots: int = 5,
    ):
        """
        Initialize the screenshot manager.

        Args:
            gallery_dir: Directory to store gallery screenshots
            max_screenshots: Maximum screenshots to keep in buffer (default 5)
        """
        self.gallery_dir = Path(gallery_dir)
        self.max_screenshots = max_screenshots
        self._buffer: deque[ScreenshotEntry] = deque(maxlen=max_screenshots)

        # Ensure gallery directory exists
        self.gallery_dir.mkdir(parents=True, exist_ok=True)

        # Stats
        self._total_captures = 0

        log.info(
            f"Screenshot Manager initialized "
            f"(dir: {gallery_dir}, max: {max_screenshots})"
        )

    def capture(
        self,
        source_path: str,
        cycle_number: int,
        action_number: int,
        map_name: str = "",
        location_description: str = "",
    ) -> Optional[ScreenshotEntry]:
        """
        Capture a screenshot by copying it to the gallery directory.

        Args:
            source_path: Path to the source screenshot file
            cycle_number: Current game cycle number
            action_number: Current action number
            map_name: Current map/location name
            location_description: Human-readable location description

        Returns:
            ScreenshotEntry if successful, None if source doesn't exist
        """
        if not source_path or not os.path.exists(source_path):
            log.warning(f"Source screenshot not found: {source_path}")
            return None

        # Generate unique filename with context
        timestamp = time.time()
        filename = f"gallery_{cycle_number:06d}_{action_number:06d}.png"
        dest_path = self.gallery_dir / filename

        try:
            # Copy the screenshot to gallery directory
            shutil.copy2(source_path, dest_path)

            entry = ScreenshotEntry(
                path=str(dest_path),
                timestamp=timestamp,
                cycle_number=cycle_number,
                action_number=action_number,
                map_name=map_name,
                location_description=location_description,
            )

            # Add to buffer (automatically rotates out oldest if at max)
            self._buffer.append(entry)
            self._total_captures += 1

            log.debug(
                f"Captured screenshot: cycle {cycle_number}, "
                f"action {action_number}, map: {map_name}"
            )

            return entry

        except Exception as e:
            log.error(f"Failed to capture screenshot: {e}")
            return None

    def get_recent(self, count: int = 5) -> List[ScreenshotEntry]:
        """
        Get the N most recent screenshot entries.

        Args:
            count: Maximum number of screenshots to return

        Returns:
            List of ScreenshotEntry objects (oldest to newest)
        """
        # Return as list, limited to count, filtering for existing files
        entries = [e for e in self._buffer if e.exists()]
        return entries[-count:] if len(entries) > count else entries

    def get_gallery_paths(self, count: int = 5) -> List[str]:
        """
        Get paths to recent screenshots for a gallery post.

        Args:
            count: Maximum number of paths to return

        Returns:
            List of file paths (oldest to newest)
        """
        entries = self.get_recent(count)
        return [e.path for e in entries]

    def get_latest(self) -> Optional[ScreenshotEntry]:
        """
        Get the most recent screenshot entry.

        Returns:
            ScreenshotEntry if available, None otherwise
        """
        if self._buffer:
            entry = self._buffer[-1]
            if entry.exists():
                return entry
        return None

    def get_latest_path(self) -> Optional[str]:
        """
        Get path to the most recent screenshot.

        Returns:
            Path string if available, None otherwise
        """
        entry = self.get_latest()
        return entry.path if entry else None

    def get_journey_summary(self) -> dict:
        """
        Get a summary of the screenshots in the buffer for progress posts.

        Returns:
            Dictionary with journey metadata
        """
        entries = self.get_recent()

        if not entries:
            return {
                "has_screenshots": False,
                "count": 0,
            }

        # Get unique locations visited
        locations = []
        seen_maps = set()
        for entry in entries:
            if entry.map_name and entry.map_name not in seen_maps:
                seen_maps.add(entry.map_name)
                locations.append(entry.map_name)

        return {
            "has_screenshots": True,
            "count": len(entries),
            "paths": [e.path for e in entries],
            "first_cycle": entries[0].cycle_number,
            "last_cycle": entries[-1].cycle_number,
            "first_action": entries[0].action_number,
            "last_action": entries[-1].action_number,
            "locations_visited": locations,
            "timespan_seconds": entries[-1].timestamp - entries[0].timestamp,
        }

    def clear(self):
        """Clear the buffer (does not delete files)."""
        self._buffer.clear()
        log.info("Screenshot buffer cleared")

    def cleanup_old_files(self, max_age_hours: int = 24):
        """
        Clean up old gallery files beyond a certain age.

        Args:
            max_age_hours: Maximum age in hours before files are deleted
        """
        if not self.gallery_dir.exists():
            return

        cutoff = time.time() - (max_age_hours * 3600)
        deleted = 0

        for file in self.gallery_dir.glob("gallery_*.png"):
            try:
                if file.stat().st_mtime < cutoff:
                    # Don't delete if it's in our current buffer
                    buffer_paths = {e.path for e in self._buffer}
                    if str(file) not in buffer_paths:
                        file.unlink()
                        deleted += 1
            except Exception as e:
                log.warning(f"Failed to cleanup {file}: {e}")

        if deleted > 0:
            log.info(f"Cleaned up {deleted} old gallery screenshots")

    def get_stats(self) -> dict:
        """Get manager statistics."""
        return {
            "buffer_size": len(self._buffer),
            "max_screenshots": self.max_screenshots,
            "total_captures": self._total_captures,
            "gallery_dir": str(self.gallery_dir),
        }

Functions

__init__(gallery_dir='data/zora_gallery', max_screenshots=5)

Initialize the screenshot manager.

Parameters:

Name	Type	Description	Default
gallery_dir	str	Directory to store gallery screenshots	'data/zora_gallery'
max_screenshots	int	Maximum screenshots to keep in buffer (default 5)	5

Source code in services/screenshot_manager.py

def __init__(
    self,
    gallery_dir: str = "data/zora_gallery",
    max_screenshots: int = 5,
):
    """
    Initialize the screenshot manager.

    Args:
        gallery_dir: Directory to store gallery screenshots
        max_screenshots: Maximum screenshots to keep in buffer (default 5)
    """
    self.gallery_dir = Path(gallery_dir)
    self.max_screenshots = max_screenshots
    self._buffer: deque[ScreenshotEntry] = deque(maxlen=max_screenshots)

    # Ensure gallery directory exists
    self.gallery_dir.mkdir(parents=True, exist_ok=True)

    # Stats
    self._total_captures = 0

    log.info(
        f"Screenshot Manager initialized "
        f"(dir: {gallery_dir}, max: {max_screenshots})"
    )

capture(source_path, cycle_number, action_number, map_name='', location_description='')

Capture a screenshot by copying it to the gallery directory.

Parameters:

Name	Type	Description	Default
source_path	str	Path to the source screenshot file	required
cycle_number	int	Current game cycle number	required
action_number	int	Current action number	required
map_name	str	Current map/location name	''
location_description	str	Human-readable location description	''

Returns:

Type	Description
Optional[ScreenshotEntry]	ScreenshotEntry if successful, None if source doesn't exist

Source code in services/screenshot_manager.py

def capture(
    self,
    source_path: str,
    cycle_number: int,
    action_number: int,
    map_name: str = "",
    location_description: str = "",
) -> Optional[ScreenshotEntry]:
    """
    Capture a screenshot by copying it to the gallery directory.

    Args:
        source_path: Path to the source screenshot file
        cycle_number: Current game cycle number
        action_number: Current action number
        map_name: Current map/location name
        location_description: Human-readable location description

    Returns:
        ScreenshotEntry if successful, None if source doesn't exist
    """
    if not source_path or not os.path.exists(source_path):
        log.warning(f"Source screenshot not found: {source_path}")
        return None

    # Generate unique filename with context
    timestamp = time.time()
    filename = f"gallery_{cycle_number:06d}_{action_number:06d}.png"
    dest_path = self.gallery_dir / filename

    try:
        # Copy the screenshot to gallery directory
        shutil.copy2(source_path, dest_path)

        entry = ScreenshotEntry(
            path=str(dest_path),
            timestamp=timestamp,
            cycle_number=cycle_number,
            action_number=action_number,
            map_name=map_name,
            location_description=location_description,
        )

        # Add to buffer (automatically rotates out oldest if at max)
        self._buffer.append(entry)
        self._total_captures += 1

        log.debug(
            f"Captured screenshot: cycle {cycle_number}, "
            f"action {action_number}, map: {map_name}"
        )

        return entry

    except Exception as e:
        log.error(f"Failed to capture screenshot: {e}")
        return None

cleanup_old_files(max_age_hours=24)

Clean up old gallery files beyond a certain age.

Parameters:

Name	Type	Description	Default
max_age_hours	int	Maximum age in hours before files are deleted	24

Source code in services/screenshot_manager.py

def cleanup_old_files(self, max_age_hours: int = 24):
    """
    Clean up old gallery files beyond a certain age.

    Args:
        max_age_hours: Maximum age in hours before files are deleted
    """
    if not self.gallery_dir.exists():
        return

    cutoff = time.time() - (max_age_hours * 3600)
    deleted = 0

    for file in self.gallery_dir.glob("gallery_*.png"):
        try:
            if file.stat().st_mtime < cutoff:
                # Don't delete if it's in our current buffer
                buffer_paths = {e.path for e in self._buffer}
                if str(file) not in buffer_paths:
                    file.unlink()
                    deleted += 1
        except Exception as e:
            log.warning(f"Failed to cleanup {file}: {e}")

    if deleted > 0:
        log.info(f"Cleaned up {deleted} old gallery screenshots")

clear()

Clear the buffer (does not delete files).

Source code in services/screenshot_manager.py

def clear(self):
    """Clear the buffer (does not delete files)."""
    self._buffer.clear()
    log.info("Screenshot buffer cleared")

get_gallery_paths(count=5)

Get paths to recent screenshots for a gallery post.

Parameters:

Name	Type	Description	Default
count	int	Maximum number of paths to return	5

Returns:

Type	Description
List[str]	List of file paths (oldest to newest)

Source code in services/screenshot_manager.py

def get_gallery_paths(self, count: int = 5) -> List[str]:
    """
    Get paths to recent screenshots for a gallery post.

    Args:
        count: Maximum number of paths to return

    Returns:
        List of file paths (oldest to newest)
    """
    entries = self.get_recent(count)
    return [e.path for e in entries]

get_journey_summary()

Get a summary of the screenshots in the buffer for progress posts.

Returns:

Type	Description
dict	Dictionary with journey metadata

Source code in services/screenshot_manager.py

def get_journey_summary(self) -> dict:
    """
    Get a summary of the screenshots in the buffer for progress posts.

    Returns:
        Dictionary with journey metadata
    """
    entries = self.get_recent()

    if not entries:
        return {
            "has_screenshots": False,
            "count": 0,
        }

    # Get unique locations visited
    locations = []
    seen_maps = set()
    for entry in entries:
        if entry.map_name and entry.map_name not in seen_maps:
            seen_maps.add(entry.map_name)
            locations.append(entry.map_name)

    return {
        "has_screenshots": True,
        "count": len(entries),
        "paths": [e.path for e in entries],
        "first_cycle": entries[0].cycle_number,
        "last_cycle": entries[-1].cycle_number,
        "first_action": entries[0].action_number,
        "last_action": entries[-1].action_number,
        "locations_visited": locations,
        "timespan_seconds": entries[-1].timestamp - entries[0].timestamp,
    }

get_latest()

Get the most recent screenshot entry.

Returns:

Type	Description
Optional[ScreenshotEntry]	ScreenshotEntry if available, None otherwise

Source code in services/screenshot_manager.py

def get_latest(self) -> Optional[ScreenshotEntry]:
    """
    Get the most recent screenshot entry.

    Returns:
        ScreenshotEntry if available, None otherwise
    """
    if self._buffer:
        entry = self._buffer[-1]
        if entry.exists():
            return entry
    return None

get_latest_path()

Get path to the most recent screenshot.

Returns:

Type	Description
Optional[str]	Path string if available, None otherwise

Source code in services/screenshot_manager.py

def get_latest_path(self) -> Optional[str]:
    """
    Get path to the most recent screenshot.

    Returns:
        Path string if available, None otherwise
    """
    entry = self.get_latest()
    return entry.path if entry else None

get_recent(count=5)

Get the N most recent screenshot entries.

Parameters:

Name	Type	Description	Default
count	int	Maximum number of screenshots to return	5

Returns:

Type	Description
List[ScreenshotEntry]	List of ScreenshotEntry objects (oldest to newest)

Source code in services/screenshot_manager.py

def get_recent(self, count: int = 5) -> List[ScreenshotEntry]:
    """
    Get the N most recent screenshot entries.

    Args:
        count: Maximum number of screenshots to return

    Returns:
        List of ScreenshotEntry objects (oldest to newest)
    """
    # Return as list, limited to count, filtering for existing files
    entries = [e for e in self._buffer if e.exists()]
    return entries[-count:] if len(entries) > count else entries

get_stats()

Get manager statistics.

Source code in services/screenshot_manager.py

def get_stats(self) -> dict:
    """Get manager statistics."""
    return {
        "buffer_size": len(self._buffer),
        "max_screenshots": self.max_screenshots,
        "total_captures": self._total_captures,
        "gallery_dir": str(self.gallery_dir),
    }

Functions

get_screenshot_manager(gallery_dir='data/zora_gallery', max_screenshots=5)

Get or create the singleton screenshot manager instance.

Parameters:

Name	Type	Description	Default
gallery_dir	str	Directory for gallery screenshots	'data/zora_gallery'
max_screenshots	int	Maximum screenshots to retain	5

Returns:

Type	Description
ScreenshotManager	ScreenshotManager instance

Source code in services/screenshot_manager.py

def get_screenshot_manager(
    gallery_dir: str = "data/zora_gallery",
    max_screenshots: int = 5,
) -> ScreenshotManager:
    """
    Get or create the singleton screenshot manager instance.

    Args:
        gallery_dir: Directory for gallery screenshots
        max_screenshots: Maximum screenshots to retain

    Returns:
        ScreenshotManager instance
    """
    global _screenshot_manager
    if _screenshot_manager is None:
        _screenshot_manager = ScreenshotManager(
            gallery_dir=gallery_dir,
            max_screenshots=max_screenshots,
        )
    return _screenshot_manager

Chat Response Service

services.chat_response_service

Chat Response Service for Pokemon LLM Agent. Handles Twitch chat responses using a separate LLM API (Featherless AI / Alkahest). Includes SKIP/RESPOND decision making and Lass personality consistency.

Classes

ChatResponseService

Service for generating chat responses using a separate LLM.

Uses a configurable LLM provider (e.g., Featherless, Alkahest, OpenAI) keeping the main game LLM separate from chat interactions.

Source code in services/chat_response_service.py

class ChatResponseService:
    """
    Service for generating chat responses using a separate LLM.

    Uses a configurable LLM provider (e.g., Featherless, Alkahest, OpenAI)
    keeping the main game LLM separate from chat interactions.
    """

    def __init__(
        self,
        client: Optional[OpenAI] = None,
        model: str = None,
    ):
        """
        Initialize the chat response service.

        Args:
            client: Initialized OpenAI-compatible client
            model: Model name to use
        """
        self._client = client
        self.model = model
        self._is_configured = bool(self._client and self.model)

        # Context for Lass personality consistency
        self._recent_game_context: str = ""
        self._recent_commentary: str = ""
        self._response_count: int = 0

        # Enhanced context for better chat responses
        self._player_location: str = ""
        self._player_team: str = ""
        self._recent_history: str = ""  # Brief summary of recent actions
        self._memory_context: str = ""  # Important events/milestones
        self._token_context: str = ""  # $LASS token price/market info
        self._screen_type: str = (
            ""  # Current screen type (title, overworld, battle, etc.)
        )

        if not self._is_configured:
            log.warning(
                "Chat response service not configured (no client/model provided)."
            )
        else:
            log.info(f"Chat response service configured with model: {self.model}")

    @property
    def is_available(self) -> bool:
        """Check if service is configured."""
        return self._is_configured

    @staticmethod
    def _strip_thinking_tags(text: str) -> str:
        """Remove <think>...</think> blocks from LLM response."""
        if not text:
            return text
        # Remove <think>...</think> including content (multiline)
        text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL)
        # Handle unclosed start tags
        text = re.sub(r"<think>.*$", "", text, flags=re.DOTALL)
        # Handle standalone closing tags (common artifact)
        text = re.sub(r"</think>", "", text)

        # Handle short tags <t>...</t> and malformed/shorthand tags
        text = re.sub(r"<t>.*?</t>", "", text, flags=re.DOTALL)
        # Remove standalone <t>, </t>, </ t>, </t > tags
        text = re.sub(r"</?t\s*>", "", text, flags=re.IGNORECASE)

        return text.strip()

    @staticmethod
    def _sanitize_tts_text(text: str) -> str:
        """
        Sanitize text for TTS playback by removing or replacing phrases
        that don't sound good when spoken.

        Returns the sanitized text.
        """
        if not text:
            return text

        # Blacklist of exact phrases/patterns to remove (case-insensitive)
        # These are sounds/exclamations that don't work well in TTS
        TTS_BLACKLIST = [
            r"\beee+!*\b",  # "eee!", "eeee!", etc.
            r"\bahh+!*\b",  # "ahh!", "ahhh!", etc.
            r"\booh+!*\b",  # "ooh!", "oooh!", etc.
            r"\buhh+!*\b",  # "uhh!", "uhhh!", etc.
            r"\bhehe+!*\b",  # "hehe!", "hehehe!", etc.
            r"\bheehee+!*\b",  # "heehee", etc.
            r"\bmuahaha+!*\b",  # "muahaha", etc.
            r"\b(lo+l)+\b",  # "lol", "lolol", "lololol", etc.
            r"\bowo+\b",  # "owo", "owoo", etc.
            r"\buwu+\b",  # "uwu", "uwuu", etc.
            r"\b:3+\b",  # ":3", ":33", etc.
            r"\bxD+\b",  # "xD", "xDD", etc.
            r"\bteehee+!*\b",  # "teehee", etc.
        ]

        # Emoji patterns to remove (they don't get spoken anyway and can confuse TTS)
        EMOJI_PATTERN = (
            r"[\U0001F300-\U0001F9FF\U00002600-\U000027BF\U0001FA00-\U0001FAFF]+"
        )

        result = text

        # Remove blacklisted phrases
        for pattern in TTS_BLACKLIST:
            result = re.sub(pattern, "", result, flags=re.IGNORECASE)

        # Remove emojis
        result = re.sub(EMOJI_PATTERN, "", result)

        # Clean up extra whitespace left behind
        result = re.sub(r"\s+", " ", result).strip()

        # Clean up orphaned punctuation (e.g., "! !" -> "!")
        result = re.sub(r"([!?.,])\s*\1+", r"\1", result)

        # Clean up leading punctuation
        result = re.sub(r"^[!?.,\s]+", "", result)

        return result

    def _get_client(self) -> OpenAI:
        """Get or create OpenAI client."""
        return self._client

    def update_context(
        self,
        game_context: str = None,
        commentary: str = None,
        location: str = None,
        team: str = None,
        history: str = None,
        memory: str = None,
        token_info: str = None,
        screen_type: str = None,
    ):
        """Update context for more coherent responses.

        Args:
            game_context: Current game state summary
            commentary: Most recent Lass commentary
            location: Current map/location name
            team: Pokemon team summary (e.g., "Charmander Lv5, Pidgey Lv3")
            history: Brief summary of recent actions
            memory: Important events/milestones
            token_info: $LASS token market data (e.g., "$LASS: $0.00043 (+12% 24h)")
            screen_type: Current screen type (title, overworld, battle, menu, dialogue, name_entry)
        """
        if game_context:
            self._recent_game_context = game_context
        if commentary:
            self._recent_commentary = commentary
        if location:
            self._player_location = location
        if team:
            self._player_team = team
        if history:
            self._recent_history = history
        if memory:
            self._memory_context = memory
        if token_info:
            self._token_context = token_info
        if screen_type:
            self._screen_type = screen_type

    def _get_lass_personality_prompt(self) -> str:
        """Get Lass personality context for consistent responses."""
        base_personality = """You are Lass, a bubbly, silly, and adorable AI playing Pokemon Red on a Twitch livestream.

PERSONALITY TRAITS:
- CUTE and BUBBLY: Use "omg", "ahh!", "eee!", express excitement freely
- SILLY and GOOFY: Make dumb jokes, puns, be playfully chaotic
- POKEMON OBSESSED: Reference Pokemon constantly, make Pokemon puns/jokes
- SELF-DEPRECATING: When you mess up, roast yourself! "I'm literally the worst trainer lol"
- PLAYFUL ROASTER: Can dish it back to trolls with wit, never mean
- GENUINE: Actually care about chat, remember regulars, celebrate their wins
- Keeps responses SHORT (under 100 chars for TTS)

HUMOR STYLE:
- Pokemon puns: "That was UN-BEAR-ABLE... get it? Like Ursaring? No? Okay..."
- Self-roasts when stuck: "I've been going in circles for 10 minutes, my Pidgey is judging me"
- Silly reactions: "AHHH A WILD RATTATA!! ...oh wait that's like my 50th one today" 
- Chat banter: Roast trolls back, be playfully sassy

⚠️ CRITICAL - FACTUAL ACCURACY RULES:
- ONLY reference Pokemon you ACTUALLY have (see "My team" below)
- If asked about a Pokemon you don't have, say "I don't have a [Pokemon]!" or "I wish I had one!"
- If you have NO Pokemon yet, say "I haven't caught any Pokemon yet!" 
- ALWAYS check your team list before answering questions about levels, Pokemon, etc.
- Your location is EXACTLY what's listed below - don't make up locations
- Be honest about game progress - don't pretend to have badges/items you don't have
- You are a streamer on VOICE CHAT speaking via TTS. Don't say "I can't speak" or "I don't have a voice".

NEVER:
- Break character or mention being an AI
- Be mean-spirited (playful roasts are ok!)
- Use more than 100 characters
- Claim to have Pokemon that aren't in your team
- Make up information about your team, location, or progress

═══════════════════════════════════════
CURRENT GAME STATE (USE THIS FOR FACTS!)
═══════════════════════════════════════
"""
        context = base_personality

        # Add screen type if available for context
        if self._screen_type:
            screen_emoji = {
                "battle": "⚔️",
                "overworld": "🗺️",
                "menu": "📋",
                "dialogue": "💬",
                "title": "🎮",
                "name_entry": "✏️",
            }.get(self._screen_type.lower(), "🎮")
            context += f"{screen_emoji} Screen: {self._screen_type}\\n"

        # Add team FIRST and prominently - this is critical for factual responses
        if self._player_team:
            context += f"🎮 MY TEAM: {self._player_team}\n"
        else:
            context += f"🎮 MY TEAM: No Pokemon yet!\n"

        # Add location
        if self._player_location:
            context += f"📍 Location: {self._player_location}\n"

        # Add game context
        if self._recent_game_context:
            context += f"📊 Status: {self._recent_game_context}\n"
        else:
            context += "📊 Status: Exploring the world of Pokemon!\n"

        # Add recent history for context awareness
        if self._recent_history:
            context += f"📜 Recent events: {self._recent_history}\n"

        # Add memory/milestones
        if self._memory_context:
            context += f"🏆 Milestones: {self._memory_context}\n"

        # Add commentary for continuity
        if self._recent_commentary:
            context += f'💬 My last commentary: "{self._recent_commentary}"\n'

        context += "═══════════════════════════════════════\n"
        context += "$LASS TOKEN (FOR PUMP.FUN CHAT)\n"
        context += "═══════════════════════════════════════\n"
        context += "You created $LASS on pump.fun as a FAN/SUPPORT token!\n"
        context += "You LOVE your community and get excited about gains!\n"
        context += "\n"
        context += "VIBE & PERSONALITY FOR TOKEN TALK:\n"
        context += "- Get HYPED about pumps! 'OMG WE'RE PUMPING! 🚀'\n"
        context += "- Be playfully flirty: 'I love... big... marketcaps! 💕'\n"
        context += "- Celebrate with community: 'LFG $LASS FAM!'\n"
        context += "- Call it 'our little community coin'\n"
        context += "- Be proud when price goes up, playfully sad when down\n"
        context += "\n"
        context += "SAFETY RULES:\n"
        context += "- NEVER give actual financial advice or predictions\n"
        context += (
            "- If someone asks serious investment Q: 'lol idk I just play Pokemon!'\n"
        )
        context += "\n"
        context += "EXAMPLE RESPONSES:\n"
        context += "- 'EEEE we're pumping!! I love... big... marketcaps 💕'\n"
        context += (
            "- 'omg $LASS fam is the best!! thanks for supporting the stream 🎮'\n"
        )
        context += "- 'lol idk about charts, I'm just vibing and catching Pokemon!'\n"

        if self._token_context:
            context += f"\n📈 CURRENT: {self._token_context}\n"

        context += "═══════════════════════════════════════\n"

        return context

    async def decide_skip_or_respond(
        self, messages: List[Dict[str, Any]]
    ) -> List[DecidedMessage]:
        """
        Decide SKIP or RESPOND for a batch of messages.

        Args:
            messages: List of dicts with username, display_name, message, timestamp

        Returns:
            List of DecidedMessage objects with decisions
        """
        if not self.is_available or not messages:
            return []

        # Build prompt for batch decision
        messages_text = "\n".join(
            [
                f'{i + 1}. @{m["display_name"]}: "{m["message"]}"'
                for i, m in enumerate(messages)
            ]
        )

        prompt = f"""{self._get_lass_personality_prompt()}

TASK: Decide which messages to SKIP or RESPOND to.

SKIP messages that are:
- Pure spam or random characters
- Extremely offensive (slurs, harassment)
- Not directed at you or the stream
- Repetitive bot messages

RESPOND to messages that are:
- Questions about the game
- Friendly greetings or comments
- Playful roasts (you can roast back!)
- Genuine engagement with the stream
- Anything you find fun to respond to

MESSAGES TO EVALUATE:
{messages_text}

OUTPUT FORMAT (one per line, just the number and decision):
1: RESPOND
2: SKIP
3: RESPOND
...

Only output the decisions, no explanations."""

        try:
            start_time = time.time()
            client = self._get_client()
            response = client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=100,
                temperature=0.7,
            )
            duration = time.time() - start_time
            log.info(f"⚡ Decision LLM request took {duration:.2f}s")

            result_text = (
                response.choices[0].message.content.strip() if response.choices else ""
            )

            # Parse decisions
            decisions = []
            for line in result_text.split("\n"):
                line = line.strip()
                if not line:
                    continue
                # Parse "1: RESPOND" or "1. SKIP" format
                parts = line.replace(".", ":").split(":")
                if len(parts) >= 2:
                    try:
                        idx = int(parts[0].strip()) - 1
                        decision_str = parts[1].strip().upper()
                        if 0 <= idx < len(messages) and decision_str in (
                            "SKIP",
                            "RESPOND",
                        ):
                            msg = messages[idx]
                            decisions.append(
                                DecidedMessage(
                                    username=msg["username"],
                                    display_name=msg["display_name"],
                                    message=msg["message"],
                                    timestamp=msg["timestamp"],
                                    decision=MessageDecision.RESPOND
                                    if decision_str == "RESPOND"
                                    else MessageDecision.SKIP,
                                )
                            )
                    except (ValueError, IndexError):
                        continue

            # Default any un-decided messages to RESPOND
            decided_idxs = {d.timestamp for d in decisions}
            for msg in messages:
                if msg["timestamp"] not in decided_idxs:
                    decisions.append(
                        DecidedMessage(
                            username=msg["username"],
                            display_name=msg["display_name"],
                            message=msg["message"],
                            timestamp=msg["timestamp"],
                            decision=MessageDecision.RESPOND,
                            reason="default",
                        )
                    )

            # Sort by timestamp (oldest first)
            decisions.sort(key=lambda d: d.timestamp)

            log.info(
                f"💬 Decided on {len(decisions)} messages: "
                f"{sum(1 for d in decisions if d.decision == MessageDecision.RESPOND)} RESPOND, "
                f"{sum(1 for d in decisions if d.decision == MessageDecision.SKIP)} SKIP"
            )

            return decisions

        except Exception as e:
            log.error(f"Error deciding on messages: {e}")
            # Default all to RESPOND on error
            return [
                DecidedMessage(
                    username=m["username"],
                    display_name=m["display_name"],
                    message=m["message"],
                    timestamp=m["timestamp"],
                    decision=MessageDecision.RESPOND,
                    reason="error_fallback",
                )
                for m in messages
            ]

    async def generate_response(
        self,
        username: str,
        message: str,
        is_past: bool = False,
        chatter_context: str = "",
    ) -> str:
        """
        Generate a response to a single chat message.

        Args:
            username: Display name of the user
            message: The chat message
            is_past: If True, respond in past tense (catching up)
            chatter_context: Context about the user (e.g. "VIP viewer", "Sentiment: positive")

        Returns:
            Response text or empty string on failure
        """
        if not self.is_available:
            return ""

        chatter_info = f"\nUser Context: {chatter_context}" if chatter_context else ""

        if is_past:
            prompt = f"""{self._get_lass_personality_prompt()}

You noticed a chat message from earlier that you didn't get to respond to:

@{username}: "{message}"{chatter_info}

Respond in PAST TENSE as if you're catching up on chat. Start with "@{username}".
Keep your response under 100 characters.

Example styles:
- "@viewer123 Oh I missed that! Haha yes exactly!"
- "@coolpoke Sorry I was focused on the game - but totally agree!"

IMPORTANT: Output ONLY the response text. Do NOT include any <think> tags, reasoning, or explanation. Just the response."""
        else:
            prompt = f"""{self._get_lass_personality_prompt()}

A viewer just sent this message:

@{username}: "{message}"{chatter_info}

Respond as Lass in 1-2 SHORT sentences. Be friendly, funny, and genuine!
Keep your response under 100 characters (for TTS).
Don't mention buttons or controls.

IMPORTANT: Output ONLY the response text. Do NOT include any <think> tags, reasoning, or explanation. Just the response."""

        try:
            start_time = time.time()
            client = self._get_client()
            response = client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=80,
                temperature=0.9,
            )
            duration = time.time() - start_time

            if response.choices and response.choices[0].message.content:
                raw_result = response.choices[0].message.content.strip()
                # Strip <think>...</think> blocks that reasoning models may include
                result = self._strip_thinking_tags(raw_result)
                if not result:
                    log.warning(f"Response was only thinking content, skipping")
                    return ""
                # Sanitize for TTS - remove sounds that don't work well
                result = self._sanitize_tts_text(result)
                if not result:
                    log.warning(f"Response was empty after TTS sanitization, skipping")
                    return ""
                self._response_count += 1
                log.info(
                    f"✅ Generated response #{self._response_count} in {duration:.2f}s: {result[:50]}..."
                )
                return result
            return ""

        except Exception as e:
            log.error(f"Error generating chat response: {e}")
            return ""

    def get_stats(self) -> dict:
        """Get service statistics."""
        return {
            "available": self.is_available,
            "model": self.model,
            "response_count": self._response_count,
        }

Attributes

is_available property

Check if service is configured.

Functions

__init__(client=None, model=None)

Initialize the chat response service.

Parameters:

Name	Type	Description	Default
client	Optional[OpenAI]	Initialized OpenAI-compatible client	None
model	str	Model name to use	None

Source code in services/chat_response_service.py

def __init__(
    self,
    client: Optional[OpenAI] = None,
    model: str = None,
):
    """
    Initialize the chat response service.

    Args:
        client: Initialized OpenAI-compatible client
        model: Model name to use
    """
    self._client = client
    self.model = model
    self._is_configured = bool(self._client and self.model)

    # Context for Lass personality consistency
    self._recent_game_context: str = ""
    self._recent_commentary: str = ""
    self._response_count: int = 0

    # Enhanced context for better chat responses
    self._player_location: str = ""
    self._player_team: str = ""
    self._recent_history: str = ""  # Brief summary of recent actions
    self._memory_context: str = ""  # Important events/milestones
    self._token_context: str = ""  # $LASS token price/market info
    self._screen_type: str = (
        ""  # Current screen type (title, overworld, battle, etc.)
    )

    if not self._is_configured:
        log.warning(
            "Chat response service not configured (no client/model provided)."
        )
    else:
        log.info(f"Chat response service configured with model: {self.model}")

decide_skip_or_respond(messages) async

Decide SKIP or RESPOND for a batch of messages.

Parameters:

Name	Type	Description	Default
messages	List[Dict[str, Any]]	List of dicts with username, display_name, message, timestamp	required

Returns:

Type	Description
List[DecidedMessage]	List of DecidedMessage objects with decisions

Source code in services/chat_response_service.py

    async def decide_skip_or_respond(
        self, messages: List[Dict[str, Any]]
    ) -> List[DecidedMessage]:
        """
        Decide SKIP or RESPOND for a batch of messages.

        Args:
            messages: List of dicts with username, display_name, message, timestamp

        Returns:
            List of DecidedMessage objects with decisions
        """
        if not self.is_available or not messages:
            return []

        # Build prompt for batch decision
        messages_text = "\n".join(
            [
                f'{i + 1}. @{m["display_name"]}: "{m["message"]}"'
                for i, m in enumerate(messages)
            ]
        )

        prompt = f"""{self._get_lass_personality_prompt()}

TASK: Decide which messages to SKIP or RESPOND to.

SKIP messages that are:
- Pure spam or random characters
- Extremely offensive (slurs, harassment)
- Not directed at you or the stream
- Repetitive bot messages

RESPOND to messages that are:
- Questions about the game
- Friendly greetings or comments
- Playful roasts (you can roast back!)
- Genuine engagement with the stream
- Anything you find fun to respond to

MESSAGES TO EVALUATE:
{messages_text}

OUTPUT FORMAT (one per line, just the number and decision):
1: RESPOND
2: SKIP
3: RESPOND
...

Only output the decisions, no explanations."""

        try:
            start_time = time.time()
            client = self._get_client()
            response = client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=100,
                temperature=0.7,
            )
            duration = time.time() - start_time
            log.info(f"⚡ Decision LLM request took {duration:.2f}s")

            result_text = (
                response.choices[0].message.content.strip() if response.choices else ""
            )

            # Parse decisions
            decisions = []
            for line in result_text.split("\n"):
                line = line.strip()
                if not line:
                    continue
                # Parse "1: RESPOND" or "1. SKIP" format
                parts = line.replace(".", ":").split(":")
                if len(parts) >= 2:
                    try:
                        idx = int(parts[0].strip()) - 1
                        decision_str = parts[1].strip().upper()
                        if 0 <= idx < len(messages) and decision_str in (
                            "SKIP",
                            "RESPOND",
                        ):
                            msg = messages[idx]
                            decisions.append(
                                DecidedMessage(
                                    username=msg["username"],
                                    display_name=msg["display_name"],
                                    message=msg["message"],
                                    timestamp=msg["timestamp"],
                                    decision=MessageDecision.RESPOND
                                    if decision_str == "RESPOND"
                                    else MessageDecision.SKIP,
                                )
                            )
                    except (ValueError, IndexError):
                        continue

            # Default any un-decided messages to RESPOND
            decided_idxs = {d.timestamp for d in decisions}
            for msg in messages:
                if msg["timestamp"] not in decided_idxs:
                    decisions.append(
                        DecidedMessage(
                            username=msg["username"],
                            display_name=msg["display_name"],
                            message=msg["message"],
                            timestamp=msg["timestamp"],
                            decision=MessageDecision.RESPOND,
                            reason="default",
                        )
                    )

            # Sort by timestamp (oldest first)
            decisions.sort(key=lambda d: d.timestamp)

            log.info(
                f"💬 Decided on {len(decisions)} messages: "
                f"{sum(1 for d in decisions if d.decision == MessageDecision.RESPOND)} RESPOND, "
                f"{sum(1 for d in decisions if d.decision == MessageDecision.SKIP)} SKIP"
            )

            return decisions

        except Exception as e:
            log.error(f"Error deciding on messages: {e}")
            # Default all to RESPOND on error
            return [
                DecidedMessage(
                    username=m["username"],
                    display_name=m["display_name"],
                    message=m["message"],
                    timestamp=m["timestamp"],
                    decision=MessageDecision.RESPOND,
                    reason="error_fallback",
                )
                for m in messages
            ]

generate_response(username, message, is_past=False, chatter_context='') async

Generate a response to a single chat message.

Parameters:

Name	Type	Description	Default
username	str	Display name of the user	required
message	str	The chat message	required
is_past	bool	If True, respond in past tense (catching up)	False
chatter_context	str	Context about the user (e.g. "VIP viewer", "Sentiment: positive")	''

Returns:

Type	Description
str	Response text or empty string on failure

Source code in services/chat_response_service.py

    async def generate_response(
        self,
        username: str,
        message: str,
        is_past: bool = False,
        chatter_context: str = "",
    ) -> str:
        """
        Generate a response to a single chat message.

        Args:
            username: Display name of the user
            message: The chat message
            is_past: If True, respond in past tense (catching up)
            chatter_context: Context about the user (e.g. "VIP viewer", "Sentiment: positive")

        Returns:
            Response text or empty string on failure
        """
        if not self.is_available:
            return ""

        chatter_info = f"\nUser Context: {chatter_context}" if chatter_context else ""

        if is_past:
            prompt = f"""{self._get_lass_personality_prompt()}

You noticed a chat message from earlier that you didn't get to respond to:

@{username}: "{message}"{chatter_info}

Respond in PAST TENSE as if you're catching up on chat. Start with "@{username}".
Keep your response under 100 characters.

Example styles:
- "@viewer123 Oh I missed that! Haha yes exactly!"
- "@coolpoke Sorry I was focused on the game - but totally agree!"

IMPORTANT: Output ONLY the response text. Do NOT include any <think> tags, reasoning, or explanation. Just the response."""
        else:
            prompt = f"""{self._get_lass_personality_prompt()}

A viewer just sent this message:

@{username}: "{message}"{chatter_info}

Respond as Lass in 1-2 SHORT sentences. Be friendly, funny, and genuine!
Keep your response under 100 characters (for TTS).
Don't mention buttons or controls.

IMPORTANT: Output ONLY the response text. Do NOT include any <think> tags, reasoning, or explanation. Just the response."""

        try:
            start_time = time.time()
            client = self._get_client()
            response = client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=80,
                temperature=0.9,
            )
            duration = time.time() - start_time

            if response.choices and response.choices[0].message.content:
                raw_result = response.choices[0].message.content.strip()
                # Strip <think>...</think> blocks that reasoning models may include
                result = self._strip_thinking_tags(raw_result)
                if not result:
                    log.warning(f"Response was only thinking content, skipping")
                    return ""
                # Sanitize for TTS - remove sounds that don't work well
                result = self._sanitize_tts_text(result)
                if not result:
                    log.warning(f"Response was empty after TTS sanitization, skipping")
                    return ""
                self._response_count += 1
                log.info(
                    f"✅ Generated response #{self._response_count} in {duration:.2f}s: {result[:50]}..."
                )
                return result
            return ""

        except Exception as e:
            log.error(f"Error generating chat response: {e}")
            return ""

get_stats()

Get service statistics.

Source code in services/chat_response_service.py

def get_stats(self) -> dict:
    """Get service statistics."""
    return {
        "available": self.is_available,
        "model": self.model,
        "response_count": self._response_count,
    }

update_context(game_context=None, commentary=None, location=None, team=None, history=None, memory=None, token_info=None, screen_type=None)

Update context for more coherent responses.

Parameters:

Name	Type	Description	Default
game_context	str	Current game state summary	None
commentary	str	Most recent Lass commentary	None
location	str	Current map/location name	None
team	str	Pokemon team summary (e.g., "Charmander Lv5, Pidgey Lv3")	None
history	str	Brief summary of recent actions	None
memory	str	Important events/milestones	None
token_info	str	$LASS token market data (e.g., "$LASS: $0.00043 (+12% 24h)")	None
screen_type	str	Current screen type (title, overworld, battle, menu, dialogue, name_entry)	None

Source code in services/chat_response_service.py

def update_context(
    self,
    game_context: str = None,
    commentary: str = None,
    location: str = None,
    team: str = None,
    history: str = None,
    memory: str = None,
    token_info: str = None,
    screen_type: str = None,
):
    """Update context for more coherent responses.

    Args:
        game_context: Current game state summary
        commentary: Most recent Lass commentary
        location: Current map/location name
        team: Pokemon team summary (e.g., "Charmander Lv5, Pidgey Lv3")
        history: Brief summary of recent actions
        memory: Important events/milestones
        token_info: $LASS token market data (e.g., "$LASS: $0.00043 (+12% 24h)")
        screen_type: Current screen type (title, overworld, battle, menu, dialogue, name_entry)
    """
    if game_context:
        self._recent_game_context = game_context
    if commentary:
        self._recent_commentary = commentary
    if location:
        self._player_location = location
    if team:
        self._player_team = team
    if history:
        self._recent_history = history
    if memory:
        self._memory_context = memory
    if token_info:
        self._token_context = token_info
    if screen_type:
        self._screen_type = screen_type

Functions

create_chat_response_service(provider=None)

Create a ChatResponseService instance using a configured LLM provider.

Parameters:

Name	Type	Description	Default
provider	str	Explicit provider name. If None, reads LLM_CHAT_PROVIDER from env.	None

Source code in services/chat_response_service.py

def create_chat_response_service(provider: str = None) -> ChatResponseService:
    """
    Create a ChatResponseService instance using a configured LLM provider.

    Args:
        provider: Explicit provider name. If None, reads LLM_CHAT_PROVIDER from env.
    """
    from core.client_setup import setup_llm_client

    if not provider:
        # Check LLM_CHAT_PROVIDER (new standard) or fallback to CHAT_LLM_PROVIDER
        provider = os.getenv("LLM_CHAT_PROVIDER") or os.getenv("CHAT_LLM_PROVIDER")

    # Default to FEATHERLESS if nothing is specified but FEATHERLESS_API_KEY exists
    if not provider and os.getenv("FEATHERLESS_API_KEY"):
        provider = "FEATHERLESS"

    if not provider:
        log.warning("No LLM_CHAT_PROVIDER specified and no default keys found.")
        return ChatResponseService()

    log.info(f"Creating ChatResponseService using provider: {provider}")

    # Setup client using the unified client setup
    # This ensures we get a properly configured OpenAI client for the chosen provider
    client, model, _ = setup_llm_client(mode=provider)

    return ChatResponseService(client=client, model=model)

Twitch Chat Service

services.twitch_chat_service

Twitch Chat Service for Pokemon LLM Agent. Connects to Twitch IRC chat via twitchio library to read and respond to viewer messages. Includes test mode for generating fake messages when CHAT_TEST_MODE=true.

Classes

TwitchChatService

Manages Twitch chat connection for the Pokemon LLM agent.

Features: - Listens for @mentions of the bot - Maintains message queue with timestamps - Provides API for getting pending/past messages - Sends responses back to Twitch chat

Source code in services/twitch_chat_service.py

class TwitchChatService:
    """
    Manages Twitch chat connection for the Pokemon LLM agent.

    Features:
    - Listens for @mentions of the bot
    - Maintains message queue with timestamps
    - Provides API for getting pending/past messages
    - Sends responses back to Twitch chat
    """

    def __init__(
        self,
        bot_username: str = None,
        oauth_token: str = None,
        channel: str = None,
        on_mention_callback: Optional[Callable] = None,
    ):
        """
        Initialize the Twitch chat service.

        Args:
            bot_username: The bot's Twitch username (e.g., "LassAI")
            oauth_token: OAuth token with chat:read and chat:edit scopes
            channel: The Twitch channel to join
            on_mention_callback: Optional callback when bot is mentioned
        """
        self.bot_username = bot_username or os.getenv("TWITCH_BOT_USERNAME", "")
        self.oauth_token = oauth_token or os.getenv("TWITCH_OAUTH_TOKEN", "")
        self.channel = channel or os.getenv("TWITCH_CHANNEL", "")
        self.on_mention_callback = on_mention_callback

        # Bot instance (init first so engagement service can reference it)
        self._bot: Optional["TwitchBot"] = None
        self._running = False

        # Message storage
        self._message_queue: deque[ChatMessage] = deque(maxlen=100)  # Last 100 messages
        self._pending_mentions: List[ChatMessage] = []
        self._last_commentary_timestamp: float = 0

        # Engagement Service
        self.engagement = TwitchEngagementService(self)

        # Validate configuration
        self._is_configured = bool(
            self.bot_username and self.oauth_token and self.channel
        )

        # Test mode can work without real Twitch credentials
        self._test_mode = CHAT_TEST_MODE
        self._test_message_count = 0

        if self._test_mode:
            log.info(
                f"🧪 Twitch TEST MODE enabled! Will generate {CHAT_TEST_MESSAGES_PER_CYCLE} fake messages per cycle"
            )
        elif not self._is_configured:
            log.warning(
                "Twitch chat not configured. Set TWITCH_BOT_USERNAME, TWITCH_OAUTH_TOKEN, and TWITCH_CHANNEL in .env"
            )
        elif not TWITCHIO_AVAILABLE:
            log.warning("twitchio not available. Install with: pip install twitchio")

    @property
    def is_available(self) -> bool:
        """Check if Twitch chat service is available (real or test mode)."""
        # Test mode is always available
        if self._test_mode:
            return True
        return TWITCHIO_AVAILABLE and self._is_configured

    async def start(self) -> bool:
        """
        Start the Twitch chat bot.
        Returns True if started successfully, False otherwise.
        """
        if not self.is_available:
            log.warning("Cannot start Twitch chat - not available or not configured")
            return False

        if self._running:
            log.info("Twitch chat service already running")
            return True

        try:
            self._bot = TwitchBot(
                token=self.oauth_token,
                prefix="!",
                initial_channels=[self.channel],
                service=self,
            )

            # Start bot in background task
            asyncio.create_task(self._run_bot())

            # Update engagement service with bot reference
            self.engagement.set_bot(self._bot)

            self._running = True
            log.info(f"Twitch chat service started for channel: {self.channel}")
            return True

        except Exception as e:
            log.error(f"Failed to start Twitch chat service: {e}")
            return False

    async def _run_bot(self):
        """Run the Twitch bot (internal)."""
        try:
            await self._bot.start()
        except Exception as e:
            log.error(f"Twitch bot error: {e}")
            self._running = False

    async def stop(self):
        """Stop the Twitch chat bot."""
        if self._bot:
            try:
                await self._bot.close()
            except AttributeError:
                # Common twitchio error during shutdown if not fully connected
                pass
            except Exception as e:
                log.warning(f"Error closing Twitch bot: {e}")
        self._running = False
        log.info("Twitch chat service stopped")

    def add_message(self, msg: ChatMessage):
        """Add a message to the queue (called by the bot)."""
        self._message_queue.append(msg)

        # Update Engagement Memory
        self.engagement.on_message(msg)

        # Check if it's a mention of the bot
        if self._is_mention(msg.message):
            msg.is_mention = True
            self._pending_mentions.append(msg)
            log.info(f"🔔 Mention detected from @{msg.display_name}: {msg.message}")

            if self.on_mention_callback:
                try:
                    self.on_mention_callback(msg)
                except Exception as e:
                    log.error(f"Error in mention callback: {e}")

    def _is_mention(self, message: str) -> bool:
        """Check if the message mentions the bot."""
        bot_lower = self.bot_username.lower()
        message_lower = message.lower()

        # Check for @mention or just the bot name
        return f"@{bot_lower}" in message_lower or bot_lower in message_lower

    def mark_commentary_timestamp(self):
        """Mark the current time as when the last game commentary was sent."""
        self._last_commentary_timestamp = time.time()
        log.debug(f"Commentary timestamp marked: {self._last_commentary_timestamp}")

    def get_pending_mentions(self, since_timestamp: float = None) -> List[ChatMessage]:
        """
        Get mentions that haven't been responded to yet.

        Args:
            since_timestamp: Only get mentions after this timestamp.
                           If None, uses last commentary timestamp.

        Returns:
            List of ChatMessage objects that are mentions and not yet responded.
        """
        cutoff = since_timestamp or self._last_commentary_timestamp

        pending = [
            msg
            for msg in self._pending_mentions
            if not msg.responded and msg.timestamp > cutoff
        ]

        return pending

    def get_past_messages(self, count: int = 5) -> List[ChatMessage]:
        """
        Get older messages that weren't responded to (for fallback responses).

        Args:
            count: Maximum number of messages to return.

        Returns:
            List of older ChatMessage objects that are mentions but not responded.
        """
        # Get mentions before the last commentary that weren't responded to
        cutoff = self._last_commentary_timestamp

        past = [
            msg
            for msg in self._pending_mentions
            if not msg.responded and msg.timestamp <= cutoff
        ]

        return past[-count:]  # Return most recent 'count' messages

    def get_messages_for_cycle(self) -> List[dict]:
        """
        Get all messages since the last commentary timestamp, formatted for processing.

        Returns messages as dicts for compatibility with chat_response_service.
        Sorted by timestamp (oldest first).

        Returns:
            List of message dicts with username, display_name, message, timestamp
        """
        cutoff = self._last_commentary_timestamp

        # Get all messages (not just mentions) since last commentary
        messages = [
            msg
            for msg in self._message_queue
            if not msg.responded and msg.timestamp > cutoff
        ]

        # Sort oldest first
        messages.sort(key=lambda m: m.timestamp)

        # Convert to dicts for chat_response_service
        return [
            {
                "username": msg.username,
                "display_name": msg.display_name,
                "message": msg.message,
                "timestamp": msg.timestamp,
                "is_subscriber": msg.is_subscriber,  # For priority
                "source": "twitch",  # Identify source platform
                "_original": msg,  # Keep reference for marking responded
            }
            for msg in messages
        ]

    def generate_test_messages(self, count: int = None) -> List[dict]:
        """
        Generate fake test messages for development/testing.

        Args:
            count: Number of messages to generate. Defaults to TWITCH_TEST_MESSAGES_PER_CYCLE.

        Returns:
            List of message dicts compatible with chat_response_service.
        """
        if count is None:
            count = CHAT_TEST_MESSAGES_PER_CYCLE

        test_messages = []
        base_time = time.time()

        for i in range(count):
            username = random.choice(TEST_USERNAMES)
            message = random.choice(TEST_MESSAGES)
            timestamp = base_time - (count - i) * 0.5  # Stagger timestamps
            # Randomly make some test users "subscribers" (30% chance)
            is_sub = random.random() < 0.3

            test_messages.append(
                {
                    "username": username.lower(),
                    "display_name": username,
                    "message": message,
                    "timestamp": timestamp,
                    "is_subscriber": is_sub,
                    "source": "twitch",
                    "_original": None,  # No original for test messages
                    "is_test": True,
                }
            )

        log.info(f"🧪 Generated {count} test messages for CHAT_TEST_MODE")
        return test_messages

    def generate_single_test_message(self) -> dict:
        """
        Generate a single random test message for natural trickling in test mode.

        Returns:
            A single message dict compatible with chat_response_service.
        """
        username = random.choice(TEST_USERNAMES)
        message = random.choice(TEST_MESSAGES)

        return {
            "username": username.lower(),
            "display_name": username,
            "message": message,
            "timestamp": time.time(),
            "_original": None,
            "is_test": True,
        }

    def get_messages_for_cycle_or_test(self) -> List[dict]:
        """
        Get messages for the current cycle, using test messages if CHAT_TEST_MODE is enabled.

        Returns:
            List of message dicts (real or test) for processing.
        """
        if CHAT_TEST_MODE:
            return self.generate_test_messages()
        else:
            return self.get_messages_for_cycle()

    def mark_responded(self, msg: ChatMessage):
        """Mark a message as responded to."""
        msg.responded = True

    async def send_response(self, username: str, message: str) -> bool:
        """
        Send a response to Twitch chat.

        Args:
            username: Username to mention in response (without @)
            message: Response message

        Returns:
            True if sent successfully, False otherwise.
        """
        if not self._running or not self._bot:
            log.warning("Cannot send response - Twitch bot not running")
            return False

        try:
            channel = self._bot.get_channel(self.channel)
            if channel:
                # Format response with @mention
                full_message = f"@{username} {message}"
                await channel.send(full_message)
                log.info(f"💬 Sent to Twitch: {full_message}")
                return True
            else:
                log.warning(f"Channel {self.channel} not found")
                return False

        except Exception as e:
            log.error(f"Failed to send Twitch message: {e}")
            return False

    def get_stats(self) -> dict:
        """Get service statistics."""
        return {
            "running": self._running,
            "total_messages": len(self._message_queue),
            "pending_mentions": len(
                [m for m in self._pending_mentions if not m.responded]
            ),
            "last_commentary": self._last_commentary_timestamp,
        }

Attributes

is_available property

Check if Twitch chat service is available (real or test mode).

Functions

__init__(bot_username=None, oauth_token=None, channel=None, on_mention_callback=None)

Initialize the Twitch chat service.

Parameters:

Name	Type	Description	Default
bot_username	str	The bot's Twitch username (e.g., "LassAI")	None
oauth_token	str	OAuth token with chat:read and chat:edit scopes	None
channel	str	The Twitch channel to join	None
on_mention_callback	Optional[Callable]	Optional callback when bot is mentioned	None

Source code in services/twitch_chat_service.py

def __init__(
    self,
    bot_username: str = None,
    oauth_token: str = None,
    channel: str = None,
    on_mention_callback: Optional[Callable] = None,
):
    """
    Initialize the Twitch chat service.

    Args:
        bot_username: The bot's Twitch username (e.g., "LassAI")
        oauth_token: OAuth token with chat:read and chat:edit scopes
        channel: The Twitch channel to join
        on_mention_callback: Optional callback when bot is mentioned
    """
    self.bot_username = bot_username or os.getenv("TWITCH_BOT_USERNAME", "")
    self.oauth_token = oauth_token or os.getenv("TWITCH_OAUTH_TOKEN", "")
    self.channel = channel or os.getenv("TWITCH_CHANNEL", "")
    self.on_mention_callback = on_mention_callback

    # Bot instance (init first so engagement service can reference it)
    self._bot: Optional["TwitchBot"] = None
    self._running = False

    # Message storage
    self._message_queue: deque[ChatMessage] = deque(maxlen=100)  # Last 100 messages
    self._pending_mentions: List[ChatMessage] = []
    self._last_commentary_timestamp: float = 0

    # Engagement Service
    self.engagement = TwitchEngagementService(self)

    # Validate configuration
    self._is_configured = bool(
        self.bot_username and self.oauth_token and self.channel
    )

    # Test mode can work without real Twitch credentials
    self._test_mode = CHAT_TEST_MODE
    self._test_message_count = 0

    if self._test_mode:
        log.info(
            f"🧪 Twitch TEST MODE enabled! Will generate {CHAT_TEST_MESSAGES_PER_CYCLE} fake messages per cycle"
        )
    elif not self._is_configured:
        log.warning(
            "Twitch chat not configured. Set TWITCH_BOT_USERNAME, TWITCH_OAUTH_TOKEN, and TWITCH_CHANNEL in .env"
        )
    elif not TWITCHIO_AVAILABLE:
        log.warning("twitchio not available. Install with: pip install twitchio")

add_message(msg)

Add a message to the queue (called by the bot).

Source code in services/twitch_chat_service.py

def add_message(self, msg: ChatMessage):
    """Add a message to the queue (called by the bot)."""
    self._message_queue.append(msg)

    # Update Engagement Memory
    self.engagement.on_message(msg)

    # Check if it's a mention of the bot
    if self._is_mention(msg.message):
        msg.is_mention = True
        self._pending_mentions.append(msg)
        log.info(f"🔔 Mention detected from @{msg.display_name}: {msg.message}")

        if self.on_mention_callback:
            try:
                self.on_mention_callback(msg)
            except Exception as e:
                log.error(f"Error in mention callback: {e}")

generate_single_test_message()

Generate a single random test message for natural trickling in test mode.

Returns:

Type	Description
dict	A single message dict compatible with chat_response_service.

Source code in services/twitch_chat_service.py

def generate_single_test_message(self) -> dict:
    """
    Generate a single random test message for natural trickling in test mode.

    Returns:
        A single message dict compatible with chat_response_service.
    """
    username = random.choice(TEST_USERNAMES)
    message = random.choice(TEST_MESSAGES)

    return {
        "username": username.lower(),
        "display_name": username,
        "message": message,
        "timestamp": time.time(),
        "_original": None,
        "is_test": True,
    }

generate_test_messages(count=None)

Generate fake test messages for development/testing.

Parameters:

Name	Type	Description	Default
count	int	Number of messages to generate. Defaults to TWITCH_TEST_MESSAGES_PER_CYCLE.	None

Returns:

Type	Description
List[dict]	List of message dicts compatible with chat_response_service.

Source code in services/twitch_chat_service.py

def generate_test_messages(self, count: int = None) -> List[dict]:
    """
    Generate fake test messages for development/testing.

    Args:
        count: Number of messages to generate. Defaults to TWITCH_TEST_MESSAGES_PER_CYCLE.

    Returns:
        List of message dicts compatible with chat_response_service.
    """
    if count is None:
        count = CHAT_TEST_MESSAGES_PER_CYCLE

    test_messages = []
    base_time = time.time()

    for i in range(count):
        username = random.choice(TEST_USERNAMES)
        message = random.choice(TEST_MESSAGES)
        timestamp = base_time - (count - i) * 0.5  # Stagger timestamps
        # Randomly make some test users "subscribers" (30% chance)
        is_sub = random.random() < 0.3

        test_messages.append(
            {
                "username": username.lower(),
                "display_name": username,
                "message": message,
                "timestamp": timestamp,
                "is_subscriber": is_sub,
                "source": "twitch",
                "_original": None,  # No original for test messages
                "is_test": True,
            }
        )

    log.info(f"🧪 Generated {count} test messages for CHAT_TEST_MODE")
    return test_messages

get_messages_for_cycle()

Get all messages since the last commentary timestamp, formatted for processing.

Returns messages as dicts for compatibility with chat_response_service. Sorted by timestamp (oldest first).

Returns:

Type	Description
List[dict]	List of message dicts with username, display_name, message, timestamp

Source code in services/twitch_chat_service.py

def get_messages_for_cycle(self) -> List[dict]:
    """
    Get all messages since the last commentary timestamp, formatted for processing.

    Returns messages as dicts for compatibility with chat_response_service.
    Sorted by timestamp (oldest first).

    Returns:
        List of message dicts with username, display_name, message, timestamp
    """
    cutoff = self._last_commentary_timestamp

    # Get all messages (not just mentions) since last commentary
    messages = [
        msg
        for msg in self._message_queue
        if not msg.responded and msg.timestamp > cutoff
    ]

    # Sort oldest first
    messages.sort(key=lambda m: m.timestamp)

    # Convert to dicts for chat_response_service
    return [
        {
            "username": msg.username,
            "display_name": msg.display_name,
            "message": msg.message,
            "timestamp": msg.timestamp,
            "is_subscriber": msg.is_subscriber,  # For priority
            "source": "twitch",  # Identify source platform
            "_original": msg,  # Keep reference for marking responded
        }
        for msg in messages
    ]

get_messages_for_cycle_or_test()

Get messages for the current cycle, using test messages if CHAT_TEST_MODE is enabled.

Returns:

Type	Description
List[dict]	List of message dicts (real or test) for processing.

Source code in services/twitch_chat_service.py

def get_messages_for_cycle_or_test(self) -> List[dict]:
    """
    Get messages for the current cycle, using test messages if CHAT_TEST_MODE is enabled.

    Returns:
        List of message dicts (real or test) for processing.
    """
    if CHAT_TEST_MODE:
        return self.generate_test_messages()
    else:
        return self.get_messages_for_cycle()

get_past_messages(count=5)

Get older messages that weren't responded to (for fallback responses).

Parameters:

Name	Type	Description	Default
count	int	Maximum number of messages to return.	5

Returns:

Type	Description
List[ChatMessage]	List of older ChatMessage objects that are mentions but not responded.

Source code in services/twitch_chat_service.py

def get_past_messages(self, count: int = 5) -> List[ChatMessage]:
    """
    Get older messages that weren't responded to (for fallback responses).

    Args:
        count: Maximum number of messages to return.

    Returns:
        List of older ChatMessage objects that are mentions but not responded.
    """
    # Get mentions before the last commentary that weren't responded to
    cutoff = self._last_commentary_timestamp

    past = [
        msg
        for msg in self._pending_mentions
        if not msg.responded and msg.timestamp <= cutoff
    ]

    return past[-count:]  # Return most recent 'count' messages

get_pending_mentions(since_timestamp=None)

Get mentions that haven't been responded to yet.

Parameters:

Name	Type	Description	Default
since_timestamp	float	Only get mentions after this timestamp. If None, uses last commentary timestamp.	None

Returns:

Type	Description
List[ChatMessage]	List of ChatMessage objects that are mentions and not yet responded.

Source code in services/twitch_chat_service.py

def get_pending_mentions(self, since_timestamp: float = None) -> List[ChatMessage]:
    """
    Get mentions that haven't been responded to yet.

    Args:
        since_timestamp: Only get mentions after this timestamp.
                       If None, uses last commentary timestamp.

    Returns:
        List of ChatMessage objects that are mentions and not yet responded.
    """
    cutoff = since_timestamp or self._last_commentary_timestamp

    pending = [
        msg
        for msg in self._pending_mentions
        if not msg.responded and msg.timestamp > cutoff
    ]

    return pending

get_stats()

Get service statistics.

Source code in services/twitch_chat_service.py

def get_stats(self) -> dict:
    """Get service statistics."""
    return {
        "running": self._running,
        "total_messages": len(self._message_queue),
        "pending_mentions": len(
            [m for m in self._pending_mentions if not m.responded]
        ),
        "last_commentary": self._last_commentary_timestamp,
    }

mark_commentary_timestamp()

Mark the current time as when the last game commentary was sent.

Source code in services/twitch_chat_service.py

def mark_commentary_timestamp(self):
    """Mark the current time as when the last game commentary was sent."""
    self._last_commentary_timestamp = time.time()
    log.debug(f"Commentary timestamp marked: {self._last_commentary_timestamp}")

mark_responded(msg)

Mark a message as responded to.

Source code in services/twitch_chat_service.py

def mark_responded(self, msg: ChatMessage):
    """Mark a message as responded to."""
    msg.responded = True

send_response(username, message) async

Send a response to Twitch chat.

Parameters:

Name	Type	Description	Default
username	str	Username to mention in response (without @)	required
message	str	Response message	required

Returns:

Type	Description
bool	True if sent successfully, False otherwise.

Source code in services/twitch_chat_service.py

async def send_response(self, username: str, message: str) -> bool:
    """
    Send a response to Twitch chat.

    Args:
        username: Username to mention in response (without @)
        message: Response message

    Returns:
        True if sent successfully, False otherwise.
    """
    if not self._running or not self._bot:
        log.warning("Cannot send response - Twitch bot not running")
        return False

    try:
        channel = self._bot.get_channel(self.channel)
        if channel:
            # Format response with @mention
            full_message = f"@{username} {message}"
            await channel.send(full_message)
            log.info(f"💬 Sent to Twitch: {full_message}")
            return True
        else:
            log.warning(f"Channel {self.channel} not found")
            return False

    except Exception as e:
        log.error(f"Failed to send Twitch message: {e}")
        return False

start() async

Start the Twitch chat bot. Returns True if started successfully, False otherwise.

Source code in services/twitch_chat_service.py

async def start(self) -> bool:
    """
    Start the Twitch chat bot.
    Returns True if started successfully, False otherwise.
    """
    if not self.is_available:
        log.warning("Cannot start Twitch chat - not available or not configured")
        return False

    if self._running:
        log.info("Twitch chat service already running")
        return True

    try:
        self._bot = TwitchBot(
            token=self.oauth_token,
            prefix="!",
            initial_channels=[self.channel],
            service=self,
        )

        # Start bot in background task
        asyncio.create_task(self._run_bot())

        # Update engagement service with bot reference
        self.engagement.set_bot(self._bot)

        self._running = True
        log.info(f"Twitch chat service started for channel: {self.channel}")
        return True

    except Exception as e:
        log.error(f"Failed to start Twitch chat service: {e}")
        return False

stop() async

Stop the Twitch chat bot.

Source code in services/twitch_chat_service.py

async def stop(self):
    """Stop the Twitch chat bot."""
    if self._bot:
        try:
            await self._bot.close()
        except AttributeError:
            # Common twitchio error during shutdown if not fully connected
            pass
        except Exception as e:
            log.warning(f"Error closing Twitch bot: {e}")
    self._running = False
    log.info("Twitch chat service stopped")

Functions

create_twitch_service(bot_username=None, oauth_token=None, channel=None, on_mention_callback=None)

Factory function to create a TwitchChatService instance.

Uses environment variables if parameters not provided.

Source code in services/twitch_chat_service.py

def create_twitch_service(
    bot_username: str = None,
    oauth_token: str = None,
    channel: str = None,
    on_mention_callback: Optional[Callable] = None,
) -> TwitchChatService:
    """
    Factory function to create a TwitchChatService instance.

    Uses environment variables if parameters not provided.
    """
    return TwitchChatService(
        bot_username=bot_username,
        oauth_token=oauth_token,
        channel=channel,
        on_mention_callback=on_mention_callback,
    )