feat(cast): add Cast TTS Gateway with multi-room audio, scheduling, and security - #926
Conversation
…nd security Self-contained Google Cast TTS gateway service providing: - Multi-room audio orchestration with device discovery and groups - Priority queue system with concurrent playback support - Scheduled announcements (cron-based) and voice rotation - Security layer with API key auth, rate limiting, and IP allowlists - Circuit breaker recovery with exponential backoff - Performance optimization with caching and connection pooling - Integration with Flute-Gateway for prosodic TTS synthesis Includes Dockerfile, docker-compose, quickstart guide, and test scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
📝 WalkthroughWalkthroughThis PR introduces a comprehensive Cast TTS Gateway microservice that integrates Google Cast device audio streaming with text-to-speech synthesis. The addition includes device management, priority announcement queuing, health monitoring, JWT authentication, circuit breaker resilience patterns, rate limiting, cron-based scheduling, voice profile management, concurrent multi-device casting, and a full HTTP API with Prometheus metrics integration. Changes
Sequence DiagramsequenceDiagram
participant Client
participant Gateway as Cast TTS Gateway
participant TTSFallback as TTS Fallback Chain
participant DeviceManager as Device Manager
participant Queue as Announcement Queue
participant NATS as NATS Bus
Client->>Gateway: POST /cast/speech<br/>(text, device, voice)
Gateway->>Queue: enqueue announcement<br/>(priority-based)
Queue-->>Gateway: queued response
Gateway->>TTSFallback: synthesize_with_fallback<br/>(text, voice)
alt TTS Provider Success
TTSFallback->>TTSFallback: try Flute/Ultimate/Google
TTSFallback-->>Gateway: audio bytes
else All Providers Failed
TTSFallback-->>Gateway: fallback error
end
Gateway->>DeviceManager: cast_audio<br/>(audio_path, device)
DeviceManager->>DeviceManager: execute catt subprocess
DeviceManager-->>Gateway: cast result
Gateway->>NATS: publish event<br/>cast.speech.completed
NATS-->>Gateway: published
Gateway-->>Client: success response<br/>(duration, device)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
📝 Coding Plan
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Docker Hardening ValidationHardening Validation ReportValidated: Sat Mar 14 18:35:46 UTC 2026Services CheckedPMOVES.AI Docker Hardening Validation[INFO] Checking: pmoves/docker-compose.hardened.yml [INFO] Validating: hi-rag-gateway-v2 [INFO] Validating: extract-worker [INFO] Validating: langextract [INFO] Validating: presign [INFO] Validating: render-webhook [INFO] Validating: retrieval-eval [INFO] Validating: pdf-ingest [INFO] Validating: jellyfin-bridge [INFO] Validating: invidious-companion-proxy [INFO] Validating: ffmpeg-whisper [INFO] Validating: media-video [INFO] Validating: media-audio [INFO] Validating: hi-rag-gateway-v2-gpu [INFO] Validating: hi-rag-gateway-gpu [INFO] Validating: deepresearch [INFO] Validating: supaserch [INFO] Validating: publisher-discord [INFO] Validating: mesh-agent [INFO] Validating: nats-echo-req [INFO] Validating: nats-echo-res [INFO] Validating: publisher [INFO] Validating: analysis-echo [INFO] Validating: graph-linker [INFO] Validating: comfy-watcher [INFO] Validating: grayjay-plugin-host [INFO] Validating: agent-zero [INFO] Validating: archon [INFO] Validating: channel-monitor [INFO] Validating: pmoves-yt [INFO] Validating: notebook-sync [INFO] Validating: supabase_service_role_key [INFO] Validating: supabase_jwt_secret ====================================== |
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
- Dockerfile: use `COPY *.py .` to include all 14 Python modules (was only copying 3) - health.py: remove duplicate `record_check` method in DeviceHealth - auth.py: change CAST_AUTH_REQUIRED default from "false" to "true" (fail-closed) - service.py: add path traversal protection for POST /cast/audio endpoint - docker-compose.yml: add tmpfs /tmp mount for read_only container - docker-compose.yml: remove deprecated `version: '3.8'` directive Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Docker Hardening ValidationHardening Validation ReportValidated: Sat Mar 14 20:23:22 UTC 2026Services CheckedPMOVES.AI Docker Hardening Validation[INFO] Checking: pmoves/docker-compose.hardened.yml [INFO] Validating: hi-rag-gateway-v2 [INFO] Validating: extract-worker [INFO] Validating: langextract [INFO] Validating: presign [INFO] Validating: render-webhook [INFO] Validating: retrieval-eval [INFO] Validating: pdf-ingest [INFO] Validating: jellyfin-bridge [INFO] Validating: invidious-companion-proxy [INFO] Validating: ffmpeg-whisper [INFO] Validating: media-video [INFO] Validating: media-audio [INFO] Validating: hi-rag-gateway-v2-gpu [INFO] Validating: hi-rag-gateway-gpu [INFO] Validating: deepresearch [INFO] Validating: supaserch [INFO] Validating: publisher-discord [INFO] Validating: mesh-agent [INFO] Validating: nats-echo-req [INFO] Validating: nats-echo-res [INFO] Validating: publisher [INFO] Validating: analysis-echo [INFO] Validating: graph-linker [INFO] Validating: comfy-watcher [INFO] Validating: grayjay-plugin-host [INFO] Validating: agent-zero [INFO] Validating: archon [INFO] Validating: channel-monitor [INFO] Validating: pmoves-yt [INFO] Validating: notebook-sync [INFO] Validating: supabase_service_role_key [INFO] Validating: supabase_jwt_secret ====================================== |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 14
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
🟡 Minor comments (14)
pmoves/services/cast-tts-gateway/device_manager.py-57-65 (1)
57-65:⚠️ Potential issue | 🟡 MinorAdd timeout to subprocess to prevent hanging.
The
catt scansubprocess has no timeout. Ifcatthangs (e.g., network issues), this will block indefinitely. Consider adding a timeout.🛡️ Proposed fix
proc = await asyncio.create_subprocess_exec( "catt", "scan", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) - stdout, stderr = await proc.communicate() + try: + stdout, _ = await asyncio.wait_for( + proc.communicate(), + timeout=30.0 # 30 second timeout for scan + ) + except asyncio.TimeoutError: + proc.kill() + return []🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/device_manager.py` around lines 57 - 65, The subprocess call that runs "catt scan" uses await proc.communicate() with no timeout and can hang; wrap the communicate call in an asyncio timeout (e.g., asyncio.wait_for(proc.communicate(), timeout=SECONDS)) and handle asyncio.TimeoutError by terminating/killing the process (proc.kill() / proc.terminate() and await proc.wait()), then return an empty list or appropriate error; update the error path where proc.returncode is checked so it also handles the timeout case. Ensure references: the asyncio.create_subprocess_exec(...) call that assigns proc and the subsequent proc.communicate(), proc.kill()/proc.terminate(), and proc.returncode checks are updated accordingly.pmoves/scripts/cast_tts.py-33-37 (1)
33-37:⚠️ Potential issue | 🟡 MinorUser input passed to subprocess without sanitization.
The
deviceparameter from command-line arguments is passed directly to subprocess. While this is a local CLI tool, consider basic validation to prevent accidental command injection if device names contain shell metacharacters.🛡️ Optional sanitization
def cast_gtts_fallback(text: str, device: str): """Fallback: cast via Google Translate TTS (max ~200 chars, low quality).""" + # Basic validation - device names should be alphanumeric with spaces + if not all(c.isalnum() or c in ' -_' for c in device): + raise ValueError(f"Invalid device name: {device}") encoded = urllib.parse.quote(text[:200]) url = f"https://translate.google.com/translate_tts?ie=UTF-8&client=tw-ob&tl=en&q={encoded}" subprocess.run(["catt", "-d", device, "cast_site", url], check=True)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/scripts/cast_tts.py` around lines 33 - 37, The cast_gtts_fallback function currently passes the device argument directly into subprocess.run; validate or sanitize device before using it to prevent accidental injection via special characters—e.g., ensure device matches a safe pattern (allow only alphanumerics, dashes and underscores) or check it against a known list of devices, raise an error on invalid names, then use the validated value (e.g., validated_device) in the subprocess.run call; keep using the argument-list form of subprocess.run (no shell) but enforce the device check in cast_gtts_fallback to reject unsafe input.pmoves/scripts/cast_tts.py-49-56 (1)
49-56:⚠️ Potential issue | 🟡 MinorValidate URL scheme to prevent SSRF via environment variable.
ULTIMATE_TTS_URLfrom environment is used directly without scheme validation. Enforcehttp://orhttps://schemes.🛡️ Proposed fix
+from urllib.parse import urlparse def cast_via_ultimate_tts(text: str, device: str): """Generate TTS via Ultimate-TTS-Studio API, then cast to speaker.""" + parsed = urlparse(ULTIMATE_TTS_URL) + if parsed.scheme not in ("http", "https"): + print(f"Invalid URL scheme: {parsed.scheme}, falling back to gTTS") + cast_gtts_fallback(text, device) + return + # Use Gradio client API to generate audio api_url = f"{ULTIMATE_TTS_URL}/gradio_api/call/synthesize"🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/scripts/cast_tts.py` around lines 49 - 56, The code uses api_url (derived from ULTIMATE_TTS_URL) directly when building the Request, which can enable SSRF; before constructing the Request in cast_tts.py validate the URL's scheme by parsing ULTIMATE_TTS_URL (or api_url) with urllib.parse.urlparse and ensure scheme is exactly 'http' or 'https', otherwise log/raise an error and abort the request; perform this check where api_url is set and reject empty or non-http(s) schemes so the urllib.request.urlopen call only ever receives a safe http/https URL.pmoves/services/cast-tts-gateway/groups.py-28-28 (1)
28-28:⚠️ Potential issue | 🟡 MinorIncorrect timezone handling in ISO timestamp.
Same issue as in
voices.py-datetime.fromtimestamp()returns local time but "Z" suffix indicates UTC.🐛 Proposed fix
+from datetime import datetime, timezone - "created_at_iso": datetime.fromtimestamp(self.created_at).isoformat() + "Z", + "created_at_iso": datetime.fromtimestamp(self.created_at, tz=timezone.utc).isoformat().replace("+00:00", "Z"),🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/groups.py` at line 28, The timestamp for "created_at_iso" is using datetime.fromtimestamp(self.created_at) which yields local time but appends "Z" (UTC); change this to produce a true UTC ISO timestamp by calling datetime.fromtimestamp(self.created_at, timezone.utc) (or datetime.utcfromtimestamp(...)+tzinfo=timezone.utc) and then outputting the ISO string (e.g., .isoformat() with the +00:00 replaced by "Z" if needed) so that the created_at_iso field correctly represents UTC; update the code that builds created_at_iso (referencing created_at_iso and self.created_at in groups.py) accordingly.pmoves/scripts/test_cast_tts_gateway_pr_fixes.sh-170-179 (1)
170-179:⚠️ Potential issue | 🟡 MinorTest counter inconsistency on SKIP.
When mypy test is skipped (line 178),
TESTS_RUNis incremented (line 171) but neitherTESTS_PASSEDnorTESTS_FAILEDis updated. This causes the final summary totals to not add up. Consider adding aTESTS_SKIPPEDcounter or not incrementingTESTS_RUNfor skippable tests.🐛 Proposed fix
+TESTS_SKIPPED=0 # Test type annotations echo -n "Testing: Type annotations (mypy)... " -TESTS_RUN=$((TESTS_RUN + 1)) if python3 -m mypy pmoves/services/cast-tts-gateway/*.py \ --ignore-missing-imports \ --no-error-summary > /dev/null 2>&1; then + TESTS_RUN=$((TESTS_RUN + 1)) echo -e "${GREEN}PASS${NC} (no type errors)" TESTS_PASSED=$((TESTS_PASSED + 1)) else echo -e "${YELLOW}SKIP${NC} (mypy not available or has errors)" + TESTS_SKIPPED=$((TESTS_SKIPPED + 1)) fi🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/scripts/test_cast_tts_gateway_pr_fixes.sh` around lines 170 - 179, The mypy branch increments TESTS_RUN but doesn't update results on skip, causing totals to misreport; either don't increment TESTS_RUN until the test actually runs or add a TESTS_SKIPPED counter and increment it in the else branch. Specifically, initialize TESTS_SKIPPED at script start (if not present), and in the mypy else branch increment TESTS_SKIPPED (or decrement TESTS_RUN) and ensure the final summary prints TESTS_SKIPPED; update references to TESTS_RUN/TESTS_PASSED/TESTS_FAILED in the summary accordingly so totals add up.pmoves/services/cast-tts-gateway/voices.py-37-37 (1)
37-37:⚠️ Potential issue | 🟡 MinorIncorrect timezone handling in ISO timestamp.
datetime.fromtimestamp()returns local time, but appending "Z" implies UTC. Usedatetime.utcfromtimestamp()ordatetime.fromtimestamp(..., tz=timezone.utc).🐛 Proposed fix
+from datetime import datetime, timezone - "created_at_iso": datetime.fromtimestamp(self.created_at).isoformat() + "Z", + "created_at_iso": datetime.fromtimestamp(self.created_at, tz=timezone.utc).isoformat().replace("+00:00", "Z"),🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/voices.py` at line 37, The created_at_iso field currently uses datetime.fromtimestamp(self.created_at).isoformat() + "Z" which produces a local-time ISO string but falsely labels it UTC; change the generation in voices.py (the created_at_iso computation) to produce a true UTC ISO timestamp by using either datetime.utcfromtimestamp(self.created_at).isoformat() + "Z" or datetime.fromtimestamp(self.created_at, tz=timezone.utc).isoformat() (and add an import for timezone if you use the latter), ensuring self.created_at remains the input epoch seconds.pmoves/services/cast-tts-gateway/requirements.txt-6-6 (1)
6-6:⚠️ Potential issue | 🟡 MinorClarify ecdsa vulnerability exposure in python-jose dependency.
The
python-josepackage includesecdsaas a required dependency. While GHSA-wj6h-64fc-37mp (Minerva timing attack on P-256) is real, no patched version of python-ecdsa exists (the project considers side-channel attacks out of scope). However, since you are usingpython-jose[cryptography], thecryptographybackend is used at runtime andecdsaremains unused despite being installed. Verify whether your code performs JWT signing operations (vulnerable to Minerva if timing can be measured) or only verification (not vulnerable); if only JWT verification is performed, this poses no practical risk.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/requirements.txt` at line 6, requirements.txt pins python-jose[cryptography] which pulls in python-ecdsa (GHSA-wj6h-64fc-37mp) but the cryptography backend is used at runtime; confirm whether our codebase performs JWT signing (which could be vulnerable to Minerva timing attacks) or only verification so we can decide remediation. Search for JWT-related symbols (calls to jwt.encode, jwt.decode, JoseError usage, any usage of python-jose in cast-tts-gateway service) and if we only verify tokens, document this finding in the repository and add a comment in requirements.txt noting that python-ecdsa is unused at runtime due to the cryptography backend; if we sign tokens, remove python-jose or replace signing with the cryptography backend explicitly (or use a library without python-ecdsa) and add a ticket to remediate.pmoves/services/cast-tts-gateway/queue.py-175-195 (1)
175-195:⚠️ Potential issue | 🟡 MinorReport the true head of the priority queue.
get_queue_status()only serializesself.queues[Priority.NORMAL][0]. If the next item isURGENT,HIGH, orLOW, this endpoint returns the wrong announcement orNone, even thoughdequeue()will play something else first.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/queue.py` around lines 175 - 195, get_queue_status currently always returns the head of self.queues[Priority.NORMAL] for "next_announcement", which is wrong when other priority queues (URGENT, HIGH, LOW) have items; update get_queue_status to find the true next item by iterating priorities in the dequeue order (e.g., check Priority in order URGENT, HIGH, NORMAL, LOW or use the same Priority iteration used by dequeue) and select the first non-empty queue's [0].to_dict() (or None if all empty), and return that as "next_announcement"; reference functions/variables: get_queue_status, dequeue, self.queues, Priority, next_announcement.pmoves/docs/voice/QUICKSTART_CAST.md-170-175 (1)
170-175:⚠️ Potential issue | 🟡 MinorSecurity claim may be inaccurate.
The documentation states "No authentication required" but the service includes JWT authentication middleware (
auth_middlewareinservice.py). The actual behavior depends onCAST_AUTH_REQUIREDenvironment variable, which defaults to"true"per PR objectives. Consider updating this section to reflect the actual authentication model.📝 Suggested update
## Security - ✅ Local control only (no Google cloud) -- ✅ No authentication required +- ✅ JWT authentication (optional, enabled by default) - ✅ Non-root container - ✅ Read-only filesystem🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/docs/voice/QUICKSTART_CAST.md` around lines 170 - 175, Update the "Security" section text to reflect that authentication is conditionally enabled: replace "No authentication required" with a statement that the service uses JWT auth via the auth_middleware in service.py and that authentication is controlled by the CAST_AUTH_REQUIRED environment variable (which defaults to "true"); mention that auth can be disabled by setting CAST_AUTH_REQUIRED appropriately so docs match runtime behavior.pmoves/services/cast-tts-gateway/service.py-566-578 (1)
566-578:⚠️ Potential issue | 🟡 MinorUnreachable code after
return.Line 578 (
return result) is unreachable because lines 566-576 have a try-except block that always returns.🐛 Proposed fix
# Use recovery manager for retry logic try: return await self.recovery_manager.execute_with_retry( _cast, circuit_breaker_key=device_name, ) except Exception as e: return { "success": False, "device": device_name, "error": str(e), } - return result🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/service.py` around lines 566 - 578, The trailing "return result" is unreachable because the try/except around self.recovery_manager.execute_with_retry(_cast, circuit_breaker_key=device_name) always returns on success or handles exceptions via the except block; remove the unreachable "return result" line (or instead assign the call to result and return it once after the try/except) to ensure the function flow is correct—look for the recovery_manager.execute_with_retry call, the _cast callback, device_name variable, and the result identifier in service.py and either delete the final "return result" or refactor the try block to set result then return it after the try/except.pmoves/services/cast-tts-gateway/service.py-320-321 (1)
320-321:⚠️ Potential issue | 🟡 MinorBare
exceptcatches all exceptions including system exits.Replace bare
except:withexcept Exception:to avoid catchingKeyboardInterrupt,SystemExit, etc.🐛 Proposed fix
try: body = await request.json() force = body.get("force", False) - except: + except Exception: force = False🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/service.py` around lines 320 - 321, The bare except in the try/except block that sets force = False should be changed to catch only standard exceptions: replace "except:" with "except Exception:" so you don't swallow SystemExit/KeyboardInterrupt; update the same try/except block (the one that assigns force = False) to use "except Exception:" and preserve the existing behavior (set force = False) — add optional logging of the caught exception if desired.pmoves/services/cast-tts-gateway/docker-compose.yml-25-29 (1)
25-29:⚠️ Potential issue | 🟡 MinorHardcoded NATS credentials violate secret hardening conventions.
The
NATS_URLcontains hardcoded credentials (nats:pmoves@). Per coding guidelines, prefer central env helpers and*_FILEsecret loading paths for critical secrets.📝 Suggested fix using environment variable
environment: - PORT=8060 - FLUTE_GATEWAY_URL=http://flute-gateway:8055 - ULTIMATE_TTS_URL=http://ultimate-tts-studio:7861 - - NATS_URL=nats://nats:pmoves@nats:4222 + - NATS_URL=${NATS_URL:-nats://nats:4222}Or use Docker secrets with
*_FILEpattern for production deployments.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/docker-compose.yml` around lines 25 - 29, The NATS credentials are hardcoded in the environment NATS_URL inside docker-compose.yml; remove the embedded username/password and switch to secret-safe loading by replacing the direct value with a reference to a secret or env-file pattern (e.g., use NATS_URL without credentials and provide NATS_USER/NATS_PASSWORD via central env helpers or use NATS_URL_FILE / NATS_PASSWORD_FILE docker secret entries). Update the environment block that contains NATS_URL to read credentials from secure sources (environment variables managed by your env helper or Docker secrets using the *_FILE convention) and ensure any service that constructs the final connection string (the startup script or service entrypoint) composes it from those secure inputs rather than from a hardcoded value.pmoves/services/cast-tts-gateway/health.py-132-145 (1)
132-145:⚠️ Potential issue | 🟡 MinorIncorrect UTC timestamp formatting.
datetime.fromtimestamp()returns local time, but the code appends"Z"(Zulu/UTC indicator). This produces incorrect ISO timestamps in non-UTC timezones.🐛 Proposed fix
+from datetime import datetime, timezone + def to_dict(self) -> dict: """Convert to dictionary.""" return { # ... "last_check": self.last_check, - "last_check_iso": datetime.fromtimestamp(self.last_check).isoformat() + "Z" + "last_check_iso": datetime.fromtimestamp(self.last_check, tz=timezone.utc).isoformat().replace("+00:00", "Z") if self.last_check else None, "last_success": self.last_success, - "last_success_iso": datetime.fromtimestamp(self.last_success).isoformat() + "Z" + "last_success_iso": datetime.fromtimestamp(self.last_success, tz=timezone.utc).isoformat().replace("+00:00", "Z") if self.last_success else None, # similar for other timestamps... }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/health.py` around lines 132 - 145, The ISO timestamp fields (last_check_iso, last_success_iso, last_failure_iso, created_at_iso) are built using datetime.fromtimestamp(...) and then appending "Z", which yields local-time values with a misleading UTC indicator; change these to create true UTC-aware datetimes by using datetime.fromtimestamp(<ts>, tz=timezone.utc).isoformat() (or datetime.utcfromtimestamp(...).replace(tzinfo=timezone.utc).isoformat()) and remove the manual +"Z" concatenation so each field is a correct RFC3339/ISO8601 UTC timestamp; update the code that sets last_check_iso, last_success_iso, last_failure_iso, and created_at_iso accordingly and import timezone from datetime if not already imported.pmoves/services/cast-tts-gateway/fallback.py-154-167 (1)
154-167:⚠️ Potential issue | 🟡 MinorBlocking I/O in async method.
gTTS.write_to_fp()performs blocking I/O which will block the event loop. Useasyncio.to_thread()to run it in a thread pool.🔧 Suggested fix
async def synthesize(self, text: str, voice: str = "default") -> Optional[bytes]: """Synthesize using Google TTS (gtts library).""" try: from gtts import gTTS import io - tts = gTTS(text=text, lang="en") - audio_fp = io.BytesIO() - tts.write_to_fp(audio_fp) - audio_fp.seek(0) - return audio_fp.read() + def _synthesize(): + tts = gTTS(text=text, lang="en") + audio_fp = io.BytesIO() + tts.write_to_fp(audio_fp) + audio_fp.seek(0) + return audio_fp.read() + + return await asyncio.to_thread(_synthesize)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/fallback.py` around lines 154 - 167, The synthesize async method uses blocking I/O via gTTS.write_to_fp which will block the event loop; modify synthesize to run the blocking parts in a thread using asyncio.to_thread (ensure asyncio is imported) by moving the gTTS creation and write_to_fp call into a synchronous helper or inline call passed to asyncio.to_thread (e.g., create an io.BytesIO, create gTTS(text=...), call write_to_fp on that buffer inside to_thread), then await the thread result, seek/read the buffer and return bytes; reference synthesize and gTTS.write_to_fp when making the change.
🧹 Nitpick comments (14)
pmoves/services/cast-tts-gateway/concurrent.py (3)
66-71: Type hint should indicate async callable.
cast_fnis awaited at line 95, but the type hintCallable[[str], dict]doesn't reflect this. UseCallable[[str], Awaitable[dict]]for accurate typing.♻️ Proposed fix
-from typing import Optional, Callable +from typing import Optional, Callable, Awaitable ... async def cast_to_devices( self, - cast_fn: Callable[[str], dict], + cast_fn: Callable[[str], Awaitable[dict]], devices: list[str], text: str = "", ) -> MultiCastResult:🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/concurrent.py` around lines 66 - 71, The type hint for cast_fn in cast_to_devices is incorrect: cast_fn is awaited inside cast_to_devices, so change its annotation from Callable[[str], dict] to Callable[[str], Awaitable[dict]] and add the necessary Awaitable import from typing (or typing_extensions) so the signature accurately reflects an async callable returning a dict; update any references to cast_fn in cast_to_devices accordingly.
83-83: Move import to module level.
import timeinside the method body should be moved to the top of the file with other imports.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/concurrent.py` at line 83, The inline "import time" found inside the function in concurrent.py should be moved to the module level with the other imports; remove the in-function import and add "import time" at the top of the file so functions like the one containing the original inline import (the method around line with the current inline import) use the module-level time import instead.
21-21:datetime.utcnow()is deprecated in Python 3.12+.Use
datetime.now(timezone.utc)instead for future compatibility.♻️ Proposed fix
from dataclasses import dataclass, field from typing import Optional, Callable -from datetime import datetime +from datetime import datetime, timezone `@dataclass` class CastResult: ... - timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z") + timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"))🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/concurrent.py` at line 21, The timestamp field's default_factory lambda uses deprecated datetime.utcnow(); update the lambda in the timestamp field (the default_factory for variable named timestamp) to use datetime.now(timezone.utc).isoformat() instead (and import timezone from datetime if not already imported) so the generated ISO timestamp remains UTC and compatible with Python 3.12+.pmoves/services/cast-tts-gateway/types.py (1)
7-7: Use built-inlistinstead oftyping.List.Python 3.9+ supports generic subscripting on built-in types. Since the project targets Python 3.11+, use
list[dict]directly instead of importingListfromtyping.♻️ Proposed fix
-from typing import TypedDict, Optional, Any, List +from typing import TypedDict, Optional, Any # Then replace all List[...] with list[...] - queue: List[dict] + queue: list[dict]As per coding guidelines: "Use Python 3.11+, 4-space indentation, and prefer type hints"
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/types.py` at line 7, Replace the use of typing.List with the builtin generic list in the import and any type annotations: remove List from the import line in types.py (the existing "from typing import TypedDict, Optional, Any, List") and update annotations that used List[...] to use list[...] (e.g., change List[dict] to list[dict]); keep other typing imports (TypedDict, Optional, Any) unchanged.pmoves/scripts/test_cast_tts_gateway_pr_fixes.sh (1)
24-41: Unusedexpected_resultparameter inrun_testfunction.The third parameter
expected_resultis defined but never used. Either remove it or implement result validation logic.♻️ Remove unused parameter
run_test() { local test_name=$1 local test_command=$2 - local expected_result=$3 echo -n "Testing: $test_name... "🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/scripts/test_cast_tts_gateway_pr_fixes.sh` around lines 24 - 41, The run_test function declares a third parameter expected_result but never uses it; either remove that parameter from run_test and all its callers (so run_test(test_name, test_command) only) or implement result validation by capturing the command output into a variable (e.g., cmd_output=$(eval "$test_command" 2>&1) or redirecting stdout), compare cmd_output to expected_result (string equality or pattern match as required) and set PASS/FAIL based on that comparison (still update TESTS_PASSED/FAILED and return codes accordingly); update all call sites of run_test to match the chosen signature.pmoves/services/cast-tts-gateway/device_manager.py (1)
120-145: Consider validating audio file exists before casting.The method passes
audio_pathdirectly tocattwithout checking if the file exists. This would give a clearer error message than relying oncatt's error output.♻️ Optional validation
+import os + async def cast_audio( self, audio_path: str, device: Optional[str] = None, ) -> dict: + if not os.path.exists(audio_path): + return { + "success": False, + "device": device, + "error": f"Audio file not found: {audio_path}", + } + try: cmd = ["catt", "cast", audio_path]🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/device_manager.py` around lines 120 - 145, The cast_audio function currently passes audio_path directly to catt; first validate the file exists (using Path(audio_path).exists() or os.path.exists) before building cmd and invoking asyncio.create_subprocess_exec, and if missing return a clear error dict (or raise) indicating the file was not found; update references in the method (audio_path, cmd, device, proc) to only proceed with subprocess execution when the existence check passes.pmoves/services/cast-tts-gateway/README.md (2)
15-19: Add language specifier to fenced code block.The architecture diagram code block is missing a language specifier, which triggers markdownlint MD040. Use
textorfor plain-text diagrams.📝 Suggested fix
-``` +```text Voice Agent → Flute-Gateway → Cast TTS Gateway → Google Cast Devices ↓ ↓ ↓ ↓ NATS Events Prosodic TTS Device Manager Nest Speakers/TV</details> <details> <summary>🤖 Prompt for AI Agents</summary>Verify each finding against the current code and only fix it if needed.
In
@pmoves/services/cast-tts-gateway/README.mdaround lines 15 - 19, The fenced
code block in README.md containing the ASCII architecture diagram lacks a
language specifier (triggering markdownlint MD040); update the opening fence for
that block (the triple backticks surrounding the diagram) to include a
plain-text specifier such as ```text so the diagram is recognized as plain
text—locate the diagram block in README.md and replace the opening fence
accordingly.</details> --- `103-132`: **Add Python and JavaScript API examples.** The README provides curl examples but is missing Python and JavaScript examples. As per coding guidelines, README documentation should include API documentation for programmatic access including JavaScript, Python, and Curl. <details> <summary>📝 Suggested Python example</summary> ```python import httpx # Discover devices response = httpx.post("http://localhost:8060/cast/discover") devices = response.json() # Cast speech response = httpx.post( "http://localhost:8060/cast/speech", json={ "text": "Hello from PMOVES voice agent", "device": "Brysons Speakers speaker" } ) ``` </details> <details> <summary>📝 Suggested JavaScript example</summary> ```javascript // Discover devices const devices = await fetch("http://localhost:8060/cast/discover", { method: "POST" }).then(r => r.json()); // Cast speech const result = await fetch("http://localhost:8060/cast/speech", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ text: "Hello from PMOVES voice agent", device: "Brysons Speakers speaker" }) }).then(r => r.json()); ``` </details> <details> <summary>🤖 Prompt for AI Agents</summary> ``` Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/README.md` around lines 103 - 132, The README currently shows only curl examples; add concise programmatic JavaScript and Python snippets demonstrating POST requests to the Cast TTS endpoints (/cast/discover, /cast/speech, /cast/stop) and the MCP API usage of the "cast_speech" tool; include examples that (1) show discovering devices via POST to /cast/discover, (2) casting speech via POST to /cast/speech with JSON body {"text": "...", "device": "..."}, and (3) stopping playback via POST to /cast/stop, using fetch/await for JavaScript and httpx (or requests) for Python, and ensure Content-Type: application/json is shown where appropriate. ``` </details> </blockquote></details> <details> <summary>pmoves/docs/voice/QUICKSTART_CAST.md (1)</summary><blockquote> `151-155`: **Add language specifier to fenced code block.** The architecture diagram is missing a language specifier. <details> <summary>📝 Suggested fix</summary> ```diff -``` +```text Voice Agent → Agent Zero → Cast MCP Tools → Cast TTS Gateway → Flute-Gateway → Nest Speakers ↓ ↓ ↓ ↓ ↓ ↓ User Input Orchestration Device Discovery TTS Synthesis Prosodic TTS Audio Output ``` ``` </details> <details> <summary>🤖 Prompt for AI Agents</summary>Verify each finding against the current code and only fix it if needed.
In
@pmoves/docs/voice/QUICKSTART_CAST.mdaround lines 151 - 155, The fenced
architecture diagram block in QUICKSTART_CAST.md lacks a language specifier;
update the fenced code block (the triple-backtick block containing "Voice Agent
→ Agent Zero → Cast MCP Tools → Cast TTS Gateway → Flute-Gateway → Nest
Speakers" and the following arrow row) to include a language token (e.g., add
"text" after the opening), so it becomestext to ensure correct syntax
highlighting/rendering.</details> </blockquote></details> <details> <summary>pmoves/services/cast-tts-gateway/service.py (1)</summary><blockquote> `120-141`: **Service uses aiohttp instead of FastAPI + uvicorn.** Per coding guidelines, API services should use FastAPI + uvicorn and include the `pmoves_health` router for `/healthz` and `/metrics` endpoints. This service uses aiohttp with custom health/metrics implementations. Consider migrating to FastAPI for consistency with other PMOVES services, or document why aiohttp was chosen for this specific use case. <details> <summary>🤖 Prompt for AI Agents</summary> ``` Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/service.py` around lines 120 - 141, The service currently constructs an aiohttp Application in CastTTSGateway.__init__ (self.app = web.Application(...)) and sets routes via _setup_routes; replace this with a FastAPI app using FastAPI(...) and include the existing auth_middleware as a dependency or middleware compatible with FastAPI, update _setup_routes to register routes on the FastAPI instance, and mount the pmoves_health router so /healthz and /metrics are provided by pmoves_health; ensure startup/shutdown hooks and any nats_client, scheduler, and background managers (concurrent_caster, priority_queue, etc.) are wired into FastAPI's lifespan events and that uvicorn is used to run the app instead of aiohttp's runner. ``` </details> </blockquote></details> <details> <summary>pmoves/services/cast-tts-gateway/flute_client.py (2)</summary><blockquote> `52-63`: **Consider reusing httpx client for connection pooling.** Creating a new `AsyncClient` per request prevents HTTP connection reuse. For a TTS service that may be called frequently, consider using a shared client instance with connection pooling. <details> <summary>📝 Suggested refactor</summary> ```diff class FluteTTSProvider: """Flute-Gateway TTS synthesis provider.""" def __init__(self, base_url: str = DEFAULT_FLUTE_URL): if not HAS_HTTPX: raise ImportError("httpx required for Flute-Gateway client") self.base_url = base_url.rstrip("/") self.api_base = f"{self.base_url}/v1/voice" + self._client: Optional[httpx.AsyncClient] = None + + async def _get_client(self) -> httpx.AsyncClient: + if self._client is None or self._client.is_closed: + self._client = httpx.AsyncClient(timeout=120.0) + return self._client async def synthesize_prosodic( self, text: str, voice: str = "default", timeout: float = 120.0, ) -> Optional[bytes]: try: - async with httpx.AsyncClient(timeout=timeout) as client: - response = await client.post( + client = await self._get_client() + response = await client.post( f"{self.api_base}/synthesize/prosodic", json={"text": text, "voice": voice}, ) - response.raise_for_status() - return response.content + response.raise_for_status() + return response.content ``` </details> <details> <summary>🤖 Prompt for AI Agents</summary> ``` Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/flute_client.py` around lines 52 - 63, The current synthesize call creates a new httpx.AsyncClient per request which prevents connection pooling; refactor FluteClient (or the class containing synthesize) to create and reuse a single AsyncClient instance (e.g., self._client) with a configured timeout and transport in the class __init__, update synthesize/prosodic call to use self._client.post(...) instead of creating a new AsyncClient, and add an async close() or __aenter__/__aexit__ to properly .aclose() the shared client when the service is shut down to avoid resource leaks. ``` </details> --- `61-63`: **Consider structured logging instead of print.** Using `print()` for error logging doesn't integrate with observability systems. Consider using Python's `logging` module for better production visibility. <details> <summary>🤖 Prompt for AI Agents</summary> ``` Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/flute_client.py` around lines 61 - 63, Replace the use of print() in the exception handler with structured logging: import and configure Python's logging and get a module logger (e.g., logger = logging.getLogger(__name__)); inside the except block in flute_client.py (the block that currently does `except Exception as e: print(f"Flute-Gateway TTS failed: {e}") return None`), call logger.exception or logger.error with exc_info=True and a clear message like "Flute-Gateway TTS failed" so the stack trace and structured fields are captured, then return None as before. ``` </details> </blockquote></details> <details> <summary>pmoves/services/cast-tts-gateway/fallback.py (1)</summary><blockquote> `219-221`: **Silent exception swallowing hides failures.** The `try-except-continue` pattern silently swallows exceptions without logging, making debugging difficult. Log the exception before continuing to the next provider/device. <details> <summary>🔧 Suggested fix</summary> ```diff except Exception as e: - # Try next provider + print(f"Provider {provider.provider_name()} failed: {e}") continue ``` </details> Also applies to: 318-320 <details> <summary>🤖 Prompt for AI Agents</summary> ``` Verify each finding against the current code and only fix it if needed. In `@pmoves/services/cast-tts-gateway/fallback.py` around lines 219 - 221, The two except Exception as e blocks that currently just "continue" should log the error (including traceback) before continuing; replace the silent swallow with a logger.exception(...) or logger.error(..., exc_info=True) call, including contextual identifiers such as the provider and device variables used in that loop (e.g., provider.name, device.id) to aid debugging, then keep the continue to try the next provider/device. ``` </details> </blockquote></details> <details> <summary>pmoves/docs/voice/cast-integration.md (1)</summary><blockquote> `41-67`: **Add language specifiers to ASCII diagram code blocks.** Multiple architecture diagram code blocks throughout this file lack language specifiers (lines 41, 173, 222, 281, 487). Use `text` for plain-text diagrams or appropriate identifiers. Example fix for line 41: <details> <summary>📝 Suggested fix</summary> ```diff -``` +```text ┌─────────────────────────────────────────────────────────────────┐ │ PMOVES Voice Agent Stack │ ``` </details> Similar fixes needed at lines 173, 222, 281, and 487. <details> <summary>🤖 Prompt for AI Agents</summary>Verify each finding against the current code and only fix it if needed.
In
@pmoves/docs/voice/cast-integration.mdaround lines 41 - 67, The ASCII
diagram code blocks (for example the "PMOVES Voice Agent Stack" diagram) lack
language specifiers; update each diagram code fence to include a plain-text
specifier by changing the openingtotext for every ASCII-art block
(e.g., the "PMOVES Voice Agent Stack" block and the other diagram blocks
mentioned in the review) so markdown renderers treat them as plain text and
preserve spacing/formatting.</details> </blockquote></details> </blockquote></details> --- <details> <summary>ℹ️ Review info</summary> <details> <summary>⚙️ Run configuration</summary> **Configuration used**: Path: .coderabbit.yaml **Review profile**: CHILL **Plan**: Pro **Run ID**: `116690ce-65b3-4105-b1e9-7bd26fc17cf1` </details> <details> <summary>📥 Commits</summary> Reviewing files that changed from the base of the PR and between 7f7e87cb50d3fc421435ecea90942b143547224f and f9a650982fec83066b390d6410eb4cd648a38356. </details> <details> <summary>📒 Files selected for processing (24)</summary> * `pmoves/docs/voice/QUICKSTART_CAST.md` * `pmoves/docs/voice/cast-integration.md` * `pmoves/scripts/cast_tts.py` * `pmoves/scripts/test_cast_tts_gateway_pr_fixes.sh` * `pmoves/services/cast-tts-gateway/Dockerfile` * `pmoves/services/cast-tts-gateway/README.md` * `pmoves/services/cast-tts-gateway/audio_queue.py` * `pmoves/services/cast-tts-gateway/auth.py` * `pmoves/services/cast-tts-gateway/concurrent.py` * `pmoves/services/cast-tts-gateway/device_manager.py` * `pmoves/services/cast-tts-gateway/docker-compose.yml` * `pmoves/services/cast-tts-gateway/fallback.py` * `pmoves/services/cast-tts-gateway/flute_client.py` * `pmoves/services/cast-tts-gateway/groups.py` * `pmoves/services/cast-tts-gateway/health.py` * `pmoves/services/cast-tts-gateway/optimize.py` * `pmoves/services/cast-tts-gateway/queue.py` * `pmoves/services/cast-tts-gateway/recovery.py` * `pmoves/services/cast-tts-gateway/requirements.txt` * `pmoves/services/cast-tts-gateway/scheduler.py` * `pmoves/services/cast-tts-gateway/security.py` * `pmoves/services/cast-tts-gateway/service.py` * `pmoves/services/cast-tts-gateway/types.py` * `pmoves/services/cast-tts-gateway/voices.py` </details> </details> <!-- This is an auto-generated comment by CodeRabbit for review status -->
| async def resume_processing(self) -> dict: | ||
| """ | ||
| Resume queue processing. | ||
|
|
||
| Returns: | ||
| Result dict | ||
| """ | ||
| async with self._lock: | ||
| if not self.session or self.session.state != QueueSessionState.PAUSED: | ||
| return { | ||
| "success": False, | ||
| "error": "No active paused session", | ||
| } | ||
|
|
||
| self.session.state = QueueSessionState.PLAYING | ||
| self.session.resumed_at = time.time() | ||
|
|
||
| return { | ||
| "success": True, | ||
| "session": self.session.to_dict(), | ||
| "message": "Resumed queue processing", | ||
| } |
There was a problem hiding this comment.
The auto-pause path cannot actually be resumed.
After 10 failures, _processing_loop() sets the session to PAUSED and breaks, which completes _processing_task. resume_processing() only flips the state back to PLAYING; it never recreates the worker, so processing stays stopped.
Also applies to: 458-471
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/audio_queue.py` around lines 141 - 162,
resume_processing currently only flips session.state to PLAYING but doesn't
restart the worker task that was stopped by _processing_loop() after
auto-pausing; update resume_processing (and the identical logic at the other
location) to check whether the background worker (_processing_task) is missing
or done and, if so, recreate and start it (e.g., self._processing_task =
asyncio.create_task(self._processing_loop()) or by calling an existing helper
like _start_processing_task()), then set session.state =
QueueSessionState.PLAYING and session.resumed_at before returning the session
dict so processing actually resumes.
| # Get next announcement | ||
| if self._queue_ref: | ||
| announcement = await self._queue_ref.dequeue() | ||
|
|
||
| if announcement: | ||
| # Reset failure counter on success | ||
| if self.session: | ||
| self.session.failed_checks = 0 | ||
| self.session.current_announcement_id = announcement.id | ||
|
|
||
| # Process announcement (this would trigger actual cast) | ||
| # For now, we'll just mark as processed | ||
| await asyncio.sleep(0.1) | ||
|
|
||
| if self.session: | ||
| self.session.processed_count += 1 | ||
| self.session.current_announcement_id = None | ||
| else: |
There was a problem hiding this comment.
Processing currently deletes announcements without ever casting them.
dequeue() removes the announcement from CastPriorityQueue, then the worker just sleeps and increments processed_count. There is no playback/cast callback here, so starting this worker will silently drain the queue without delivering audio.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/audio_queue.py` around lines 426 - 443, The
worker is currently removing announcements via self._queue_ref.dequeue() and
treating them as processed after a sleep, which silently drains the queue;
instead, change the flow in the worker (around _queue_ref.dequeue(),
session.processed_count and session.current_announcement_id) to only
remove/acknowledge the item after a successful cast/playback: use a
non-destructive peek or requeue pattern (or call a
cast_announcement(announcement) / playback API and await its result) and on
success call the queue's acknowledge/remove, increment session.processed_count
and clear current_announcement_id; on failure increment session.failed_checks
and re-enqueue or leave the item so it isn’t lost. Ensure the casting/callback
is awaited and errors are logged/handled before mutating the queue or session
state.
| from aiohttp import web | ||
| import jose.jwt | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -n -C2 --type=py 'from aiohttp import web|@web\.middleware|FastAPI|uvicorn|pmoves_health' pmoves/services/cast-tts-gatewayRepository: POWERFULMOVES/PMOVES.AI
Length of output: 1049
🏁 Script executed:
# Check the full structure of the cast-tts-gateway service
cat -n pmoves/services/cast-tts-gateway/service.py | head -100Repository: POWERFULMOVES/PMOVES.AI
Length of output: 3537
🏁 Script executed:
# Search for FastAPI usage in other services to confirm the pattern
rg -n 'from fastapi import|FastAPI|pmoves_health' pmoves/services --type=py | head -20Repository: POWERFULMOVES/PMOVES.AI
Length of output: 1786
🏁 Script executed:
# Check if there are any other middleware or app setup patterns in cast-tts-gateway
fd . pmoves/services/cast-tts-gateway --type f -name "*.py" | xargs wc -lRepository: POWERFULMOVES/PMOVES.AI
Length of output: 316
🏁 Script executed:
# Check pmoves_health usage in services
rg -n 'pmoves_health' pmoves/services --type=py | head -20Repository: POWERFULMOVES/PMOVES.AI
Length of output: 302
🏁 Script executed:
# Look for health endpoints in cast-tts-gateway
rg -n -A5 'def.*health|/health|/healthz' pmoves/services/cast-tts-gateway --type=pyRepository: POWERFULMOVES/PMOVES.AI
Length of output: 8125
🏁 Script executed:
# Check if cast-tts-gateway has uvicorn references
rg -n 'uvicorn' pmoves/services/cast-tts-gateway --type=pyRepository: POWERFULMOVES/PMOVES.AI
Length of output: 49
🏁 Script executed:
# Look at the complete auth.py to see the full middleware implementation (lines 83-122)
sed -n '83,122p' pmoves/services/cast-tts-gateway/auth.pyRepository: POWERFULMOVES/PMOVES.AI
Length of output: 1384
Refactor cast-tts-gateway to use FastAPI + uvicorn with pmoves_health integration.
The service uses aiohttp throughout (service.py line 16, auth.py line 10) while all other services in pmoves/services/ follow the FastAPI + uvicorn pattern (botz-gateway, supaserch, showtime-api, session-context-worker, presign, retrieval-eval). The auth middleware (lines 83-122) uses aiohttp-specific @web.middleware and web.Request/web.Response types, requiring a complete rewrite. Additionally, while /healthz and /metrics endpoints exist, they lack integration with the pmoves_health router as required by coding guidelines.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/auth.py` around lines 10 - 12, The file
imports and middleware are still aiohttp-specific (import aiohttp.web,
`@web.middleware`, web.Request/web.Response and the auth middleware block around
lines 83-122); rewrite auth.py to use FastAPI/uvicorn patterns: replace aiohttp
import with FastAPI, use FastAPI's Middleware or dependency injection for the
auth logic (convert the existing auth middleware into a FastAPI dependency
function or Starlette-compatible middleware), update request/response types to
FastAPI/Starlette equivalents, and register the pmoves_health router (instead of
the current /healthz and /metrics handlers) so health and metrics integrate with
pmoves_health; ensure you update any entrypoint/create_app function or router
registration to use FastAPI app startup/shutdown and uvicorn launch semantics.
| jwt_secret = os.getenv("SUPABASE_JWT_SECRET") | ||
|
|
||
| if not jwt_secret: | ||
| # In development mode, allow unauthenticated requests | ||
| return { | ||
| "user_id": "dev_user", | ||
| "role": "admin", | ||
| "email": "dev@pmoves.ai" | ||
| } |
There was a problem hiding this comment.
A missing JWT secret currently turns any Bearer header into admin access.
Once the request passes the Bearer prefix check, get_user_context() returns the hard-coded dev_user admin profile whenever SUPABASE_JWT_SECRET is unset. That makes auth fail open on a bad deployment/bootstrap instead of rejecting the request.
🔒 Minimal fail-closed change
- jwt_secret = os.getenv("SUPABASE_JWT_SECRET")
-
- if not jwt_secret:
- # In development mode, allow unauthenticated requests
- return {
- "user_id": "dev_user",
- "role": "admin",
- "email": "dev@pmoves.ai"
- }
+ jwt_secret = os.getenv("SUPABASE_JWT_SECRET")
+ if not jwt_secret:
+ raise web.HTTPUnauthorized(
+ reason="Authentication is enabled but no Supabase JWT secret is configured"
+ )As per coding guidelines, pmoves/services/**: Focus on PMOVES secret hardening conventions: - Prefer central env helpers and *_FILE secret loading paths. - Flag direct critical-secret reads and plaintext fallbacks. - Prioritize auth/bootstrap regressions affecting first-run onboarding.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/auth.py` around lines 49 - 57,
get_user_context currently treats a missing SUPABASE_JWT_SECRET as a signal to
return a hard-coded admin dev profile, which fails open; change get_user_context
(in auth.py) to load the secret using the project's central secret loader
(support SUPABASE_JWT_SECRET and SUPABASE_JWT_SECRET_FILE conventions) and, if
no secret is present, do NOT return the dev_user profile—return an explicit
authentication failure (raise an AuthError/return None/HTTP 401) so requests
with Bearer headers are rejected when the secret is absent; remove the
unconditional dev fallback and instead gate any dev-only behavior behind an
explicit, documented dev flag if needed.
| async def synthesize_with_fallback( | ||
| self, | ||
| text: str, | ||
| voice: str = "default", | ||
| ) -> FallbackResult: | ||
| """ | ||
| Synthesize with fallback chain. | ||
|
|
||
| Args: | ||
| text: Text to synthesize | ||
| voice: Voice identifier | ||
|
|
||
| Returns: | ||
| FallbackResult with audio data or error | ||
| """ | ||
| import time | ||
|
|
||
| start_time = time.time() | ||
|
|
||
| for provider in self.providers: | ||
| try: | ||
| audio_data = await provider.synthesize(text, voice) | ||
|
|
||
| if audio_data: | ||
| duration_ms = (time.time() - start_time) * 1000 | ||
|
|
||
| return FallbackResult( | ||
| success=True, | ||
| provider_used=provider.provider_name(), | ||
| attempts=self.providers.index(provider) + 1, | ||
| duration_ms=duration_ms, | ||
| ) | ||
|
|
||
| except Exception as e: | ||
| # Try next provider | ||
| continue | ||
|
|
||
| # All providers failed | ||
| duration_ms = (time.time() - start_time) * 1000 | ||
|
|
||
| return FallbackResult( | ||
| success=False, | ||
| attempts=len(self.providers), | ||
| error="All TTS providers failed", | ||
| duration_ms=duration_ms, | ||
| ) |
There was a problem hiding this comment.
Synthesized audio data is discarded and never returned.
synthesize_with_fallback calls provider.synthesize() and gets audio_data, but FallbackResult doesn't store the audio bytes. The returned result only indicates success/failure. This makes the fallback system unusable for its intended purpose.
🐛 Proposed fix: Add audio_data field to FallbackResult
`@dataclass`
class FallbackResult:
"""Result from fallback chain execution."""
success: bool
provider_used: Optional[str] = None
device_used: Optional[str] = None
attempts: int = 0
error: Optional[str] = None
duration_ms: float = 0.0
+ audio_data: Optional[bytes] = None
def to_dict(self) -> dict:
"""Convert to dictionary."""
return {
"success": self.success,
"provider_used": self.provider_used,
"device_used": self.device_used,
"attempts": self.attempts,
"error": self.error,
"duration_ms": self.duration_ms,
+ "audio_size": len(self.audio_data) if self.audio_data else 0,
}Then update synthesize_with_fallback:
if audio_data:
duration_ms = (time.time() - start_time) * 1000
return FallbackResult(
success=True,
provider_used=provider.provider_name(),
attempts=self.providers.index(provider) + 1,
duration_ms=duration_ms,
+ audio_data=audio_data,
)🧰 Tools
🪛 Ruff (0.15.5)
[error] 219-221: try-except-continue detected, consider logging the exception
(S112)
[warning] 219-219: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/fallback.py` around lines 186 - 231, The
FallbackResult currently doesn't carry the synthesized audio bytes so
synthesize_with_fallback discards audio_data from provider.synthesize; update
the FallbackResult type/constructor to include an optional audio_data: bytes (or
None) field, update its type hint wherever defined, and then in
synthesize_with_fallback set audio_data=audio_data on the successful
FallbackResult return (keep provider_used, attempts, duration_ms), leaving
audio_data=None on the failure FallbackResult; ensure the function
signature/annotation for synthesize_with_fallback and any callers/types reflect
the added field and that provider.synthesize returns/awaits bytes as expected
(use provider.provider_name() to locate the provider metadata).
| async with self._lock: | ||
| for announcement in list(self.scheduled.values()): | ||
| # Check if announcement is due | ||
| if ( | ||
| announcement.enabled | ||
| and announcement.next_run | ||
| and announcement.next_run <= now | ||
| ): | ||
| # Execute cast | ||
| try: | ||
| await self.cast_fn( | ||
| text=announcement.text, | ||
| device=announcement.device, | ||
| group=announcement.group, | ||
| voice=announcement.voice, | ||
| ) |
There was a problem hiding this comment.
Don't hold the scheduler mutex across the cast I/O.
The injected cast_fn in pmoves/services/cast-tts-gateway/service.py (Lines 1912-1962) performs TTS synthesis and device casting. Awaiting it inside async with self._lock blocks schedule(), cancel(), and other due jobs behind one long-running cast.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/scheduler.py` around lines 363 - 378, You're
holding self._lock while awaiting the long-running cast_fn which blocks other
scheduler operations (schedule(), cancel(), and other due jobs); instead, inside
the async with self._lock gather the list of due announcements (from
self.scheduled where announcement.enabled and announcement.next_run <= now) and
any metadata needed to reschedule (e.g., announcement.id or next_run delta),
then release the lock and iterate the collected announcements calling await
self.cast_fn(...) outside the lock; after each cast, reacquire self._lock only
to update the announcement state in self.scheduled (e.g., set next_run or
enabled) and then release it, so casting I/O is not done while holding
self._lock.
| # Execute cast | ||
| try: | ||
| await self.cast_fn( | ||
| text=announcement.text, | ||
| device=announcement.device, | ||
| group=announcement.group, | ||
| voice=announcement.voice, | ||
| ) | ||
|
|
||
| announcement.last_run = now | ||
| announcement.run_count += 1 | ||
|
|
||
| # Update next run for recurring | ||
| if announcement.schedule_type == ScheduleType.RECURRING: | ||
| next_run = self._parse_cron_next(announcement.cron) | ||
| if next_run: | ||
| announcement.next_run = next_run.timestamp() | ||
| else: | ||
| # Disable if cron parsing fails | ||
| announcement.enabled = False | ||
| else: | ||
| # Remove one-shot after execution | ||
| del self.scheduled[announcement.id] |
There was a problem hiding this comment.
Treat a failed cast result as a failed schedule execution.
The callback supplied by pmoves/services/cast-tts-gateway/service.py (Lines 1912-1962) returns {"success": False, ...} on synthesis/device errors. This loop ignores that result and always updates last_run / run_count; one-shot jobs are even deleted after a failed cast.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/scheduler.py` around lines 371 - 393, The
code currently ignores the return value of await self.cast_fn(...) and always
marks the announcement as executed; change this to capture the result (e.g.,
result = await self.cast_fn(...)) and only update announcement.last_run,
announcement.run_count, and reschedule/remove one-shot entries when
result.get("success") is truthy; if success is false, do not increment run_count
or delete the one-shot from self.scheduled (so the schedule can retry), and keep
announcement.enabled unchanged (unless the cast result explicitly signals a
terminal failure you want to disable). Ensure you update the logic around
cast_fn, announcement, ScheduleType.RECURRING and _parse_cron_next accordingly
to only perform reschedule/delete on successful casts.
| @dataclass | ||
| class AuditLogEntry: | ||
| """Audit log entry.""" | ||
|
|
||
| timestamp: float = field(default_factory=time.time) | ||
| action: str = "" | ||
| user: Optional[str] = None | ||
| device: Optional[str] = None | ||
| result: str = "success" # success, failure, error | ||
| error: Optional[str] = None | ||
| metadata: dict = field(default_factory=dict) | ||
| ip_address: Optional[str] = None | ||
| user_agent: Optional[str] = None | ||
|
|
||
| def to_dict(self) -> dict: | ||
| """Convert to dictionary.""" | ||
| return { | ||
| "timestamp": self.timestamp, | ||
| "timestamp_iso": datetime.fromtimestamp(self.timestamp).isoformat() + "Z", | ||
| "action": self.action, | ||
| "user": self.user, | ||
| "device": self.device, | ||
| "result": self.result, | ||
| "error": self.error, | ||
| "metadata": self.metadata, | ||
| "ip_address": self.ip_address, | ||
| "user_agent": self.user_agent, | ||
| } |
There was a problem hiding this comment.
Audit retention is manual-only right now.
retention_hours is never enforced during log() or SecurityManager.start(). These audit entries are kept until somebody calls cleanup_old_entries() or the deque evicts them at max_entries, so the configured retention window is not actually enforced.
Also applies to: 197-324, 457-463
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/security.py` around lines 16 - 43,
Retention_hours is never enforced so audit entries persist until manual cleanup
or max_entries eviction; update SecurityManager.start() and
SecurityManager.log() to call cleanup_old_entries() (or apply its logic) to
purge entries older than retention_hours before returning/after appending,
ensuring retention window is honored; use the existing cleanup_old_entries()
helper or replicate its timestamp comparison against AuditLogEntry.timestamp and
retention_hours, and keep existing max_entries deque behavior as a secondary
safeguard.
| from types import ( | ||
| SuccessResponse, | ||
| ErrorResponse, | ||
| CastSpeechResponse, | ||
| QueueAnnouncementResponse, | ||
| QueueStatusResponse, | ||
| VoiceProfileResponse, | ||
| VoiceProfilesListResponse, | ||
| GroupResponse, | ||
| GroupsListResponse, | ||
| HealthCheckResponse, | ||
| ScheduleResponse, | ||
| SchedulesListResponse, | ||
| DeviceDiscoveryResponse, | ||
| DevicesListResponse, | ||
| CastStatusResponse, | ||
| ) |
There was a problem hiding this comment.
Import shadows Python's built-in types module.
from types import ... shadows the built-in types module. Rename the local module or use a different import approach.
🐛 Proposed fix
-from types import (
+from types import ( # noqa: A004 # or rename the moduleOr rename pmoves/services/cast-tts-gateway/types.py to schemas.py or models.py:
-from types import (
+from schemas import (
SuccessResponse,
ErrorResponse,
# ...
)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/service.py` around lines 40 - 56, The
current line "from types import SuccessResponse, ErrorResponse, ..." imports
from a local module named types and shadows Python's built-in types module;
update the import to avoid the conflict by either renaming the local module
(e.g., to schemas.py or models.py) and updating imports to "from schemas import
SuccessResponse, ..." or keep the module name but import it with a
non-conflicting alias (e.g., "from . import types as local_types" and reference
local_types.SuccessResponse, ErrorResponse, CastSpeechResponse, etc.), then
update any references to those symbols accordingly.
| with CAST_LATENCY.time(): | ||
| # Synthesize TTS | ||
| audio_data = None | ||
| audio_path = None | ||
|
|
||
| # Try Flute-Gateway first | ||
| if use_flute: | ||
| audio_data = await self.flute_provider.synthesize_prosodic( | ||
| text=text, | ||
| voice=voice, | ||
| ) | ||
|
|
||
| # Fallback to Ultimate-TTS via API | ||
| if not audio_data: | ||
| try: | ||
| import httpx | ||
|
|
||
| # Use voice profile parameters (speed, pitch) with defaults | ||
| voice_model = "Kokoro" if voice == "default" else voice | ||
| speed = max(0.5, min(2.0, voice_speed)) | ||
| pitch = max(0.5, min(2.0, voice_pitch)) | ||
|
|
||
| async with httpx.AsyncClient(timeout=120.0) as client: | ||
| response = await client.post( | ||
| f"{ULTIMATE_TTS_URL}/api/predict", | ||
| json={"data": [text, voice_model, speed, pitch, speed]}, | ||
| ) | ||
| response.raise_for_status() | ||
| result_data = response.json() | ||
|
|
||
| if "data" in result_data and len(result_data["data"]) > 0: | ||
| audio_path = result_data["data"][0] | ||
|
|
||
| except Exception as e: | ||
| CAST_REQUESTS.labels(method="speech", status="error").inc() | ||
| return web.json_response( | ||
| {"error": f"Failed to synthesize TTS: {str(e)}"}, | ||
| status=500 | ||
| ) | ||
|
|
||
| if not audio_data and not audio_path: | ||
| CAST_REQUESTS.labels(method="speech", status="error").inc() | ||
| return web.json_response( | ||
| {"error": "Failed to synthesize TTS"}, | ||
| status=500 | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify if fallback_manager or optimization_manager are actually used for TTS synthesis
echo "=== Checking usage of fallback_manager.synthesize_tts ==="
rg -n "fallback_manager\.synthesize" pmoves/services/cast-tts-gateway/
echo ""
echo "=== Checking usage of optimization_manager.get_cached ==="
rg -n "optimization_manager\.(get_cached|cache_audio)" pmoves/services/cast-tts-gateway/
echo ""
echo "=== Checking if AudioCache is used anywhere ==="
rg -n "AudioCache|get_cached_audio" pmoves/services/cast-tts-gateway/Repository: POWERFULMOVES/PMOVES.AI
Length of output: 528
🏁 Script executed:
#!/bin/bash
# Check if fallback_manager and optimization_manager are initialized in the service
echo "=== Checking for fallback_manager initialization ==="
rg -n "fallback_manager" pmoves/services/cast-tts-gateway/service.py
echo ""
echo "=== Checking for optimization_manager initialization ==="
rg -n "optimization_manager" pmoves/services/cast-tts-gateway/service.py
echo ""
echo "=== Looking at handle_cast_speech method context ==="
sed -n '460,520p' pmoves/services/cast-tts-gateway/service.py | head -70Repository: POWERFULMOVES/PMOVES.AI
Length of output: 3679
🏁 Script executed:
#!/bin/bash
# Check FallbackManager implementation
echo "=== FallbackManager class and methods ==="
rg -A 5 "class FallbackManager" pmoves/services/cast-tts-gateway/
echo ""
echo "=== Checking for synthesize_tts method in FallbackManager ==="
rg -n "def synthesize_tts" pmoves/services/cast-tts-gateway/
echo ""
echo "=== OptimizationManager class ==="
rg -A 5 "class OptimizationManager" pmoves/services/cast-tts-gateway/
echo ""
echo "=== Looking for cache_audio method in OptimizationManager ==="
rg -n "def cache_audio|async def cache_audio" pmoves/services/cast-tts-gateway/Repository: POWERFULMOVES/PMOVES.AI
Length of output: 1413
🏁 Script executed:
#!/bin/bash
# Check FallbackManager.synthesize_tts implementation
echo "=== FallbackManager.synthesize_tts method ==="
sed -n '424,450p' pmoves/services/cast-tts-gateway/fallback.py
echo ""
echo "=== OptimizationManager.get_cached_audio method ==="
sed -n '428,445p' pmoves/services/cast-tts-gateway/optimize.py
echo ""
echo "=== OptimizationManager.cache_audio method ==="
sed -n '432,460p' pmoves/services/cast-tts-gateway/optimize.pyRepository: POWERFULMOVES/PMOVES.AI
Length of output: 2808
Use configured fallback and optimization managers for TTS synthesis.
The handle_cast_speech method directly calls self.flute_provider.synthesize_prosodic() and falls back to Ultimate-TTS via inline httpx code, bypassing:
self.fallback_manager.synthesize_tts()(fallback.py:424) — designed to apply configured TTS fallback chains, though currently incomplete (returnsNonefor audio bytes)self.optimization_manager.get_cached_audio()andcache_audio()(optimize.py:428, 432) — audio caching is never checked or populated, losing optimization opportunity
Route TTS synthesis through the fallback and optimization managers to ensure consistency with the configured infrastructure and enable audio caching for repeated synthesis requests.
🧰 Tools
🪛 Ruff (0.15.5)
[warning] 503-503: Do not catch blind exception: Exception
(BLE001)
[warning] 506-506: Use explicit conversion flag
Replace with conversion flag
(RUF010)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@pmoves/services/cast-tts-gateway/service.py` around lines 470 - 515, The
handler currently calls self.flute_provider.synthesize_prosodic() and inlines an
httpx fallback, bypassing the configured managers; change handle_cast_speech to
first consult self.optimization_manager.get_cached_audio(text, voice,
voice_speed, voice_pitch) and return cached bytes if present, otherwise call
self.fallback_manager.synthesize_tts(text, voice, voice_speed, voice_pitch)
(which should encapsulate trying flute then Ultimate-TTS) to obtain audio bytes
or a path, then if audio bytes were produced call
self.optimization_manager.cache_audio(...) to persist them before responding;
remove the direct httpx POST and the direct call to
self.flute_provider.synthesize_prosodic() so all synthesis flows through
fallback_manager.synthesize_tts and caching flows through
optimization_manager.get_cached_audio and optimization_manager.cache_audio.
Downgrade dev bypass role from "admin" to "dev" to limit privilege escalation in development mode. Add logger.warning() when auth is bypassed so operators can detect misconfiguration in production logs. Addresses: Z890 gap analysis Issue #4 (PR #926) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* fix(security): remove hardcoded CHIT passphrase from consciousness-service
Remove `pmoves-chit-default` default from Dockerfile ENV and main.py fallback.
Docker-compose enforces runtime injection via ${CHIT_PROD_PASSPHRASE:?...},
but the Dockerfile default was a security smell if the image ran standalone.
Addresses: Z890 gap analysis Issue #3 (PR #905)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* fix(security): add dev-mode warning and restrict role in cast-tts auth
Downgrade dev bypass role from "admin" to "dev" to limit privilege escalation
in development mode. Add logger.warning() when auth is bypassed so operators
can detect misconfiguration in production logs.
Addresses: Z890 gap analysis Issue #4 (PR #926)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* fix(security): warn on unauthenticated NATS fallback in flute-gateway
_build_nats_url() silently fell back to unauthenticated nats:// when
NATS_URL and NATS_USER/NATS_PASSWORD were all unset. Add logger.warning()
so operators can detect missing NATS credentials in logs.
Addresses: Z890 gap analysis Issue #5 (PR #927)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* fix(security): remove docsRoot path leak from audit summary API
The /api/audit/summary response included `docsRoot` — an absolute server
filesystem path — in the JSON body. This leaks internal directory structure
to unauthenticated clients. Remove it from the response payload.
Addresses: Z890 gap analysis Issue #7 (PR #922)
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 participant
Summary
pmoves/services/cast-tts-gateway/Files Changed (24 files, ~8.6k lines)
pmoves/services/cast-tts-gateway/— 18 service files (service.py is 2066 lines)pmoves/docs/voice/cast-integration.md— 773-line integration guidepmoves/docs/voice/QUICKSTART_CAST.md— quick start guidepmoves/scripts/cast_tts.py— CLI helperpmoves/scripts/test_cast_tts_gateway_pr_fixes.sh— test scriptTest plan
python -c "import pmoves.services.cast_tts_gateway.service"succeedsdocker build pmoves/services/cast-tts-gateway/builds without errorbash pmoves/scripts/test_cast_tts_gateway_pr_fixes.shpassescast-tts-gateway/,docs/voice/, orscripts/present🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Documentation