Release v0.3.136

fix: make URL-token auth opt-in so env token no longer forces 401

docs/features/knowledge-embeddings.md

@@ -0,0 +1,69 @@
+# Knowledge Embedding Model Configuration
+
+Knowledge / RAG search uses vector **embeddings**. The default embedding model is
+`text-embedding-3-small`. Many OpenAI keys or projects do not have access to that
+model (organization tier, model allowlist, or regional restrictions), which caused
+knowledge search to silently degrade to local substring matching.
+
+PraisonAIUI now makes this failure mode configurable and observable.
+
+## Configure the embedding model
+
+Set an accessible model via environment variable:
+
+```bash
+export AIUI_EMBEDDING_MODEL=text-embedding-ada-002
+```
+
+`OPENAI_EMBEDDING_MODEL` is also honoured as a fallback if `AIUI_EMBEDDING_MODEL`
+is not set.
+
+## Fallback chain
+
+The OpenAI-compatible `/v1/embeddings` endpoint tries the requested (or default)
+model first, then each model in the fallback chain if the primary returns a
+`model_not_found` / 403 access error:
+
+```bash
+export AIUI_EMBEDDING_FALLBACK_MODELS="text-embedding-ada-002,text-embedding-3-small"
+```
+
+If every candidate is inaccessible, the endpoint returns `503` with an actionable
+`model_not_found` error instead of a generic `500`.
+
+## Observing degraded search
+
+`GET /api/knowledge/status` reports embedding availability:
+
+```json
+{
+  "total": 12,
+  "backend": "SDKKnowledgeManager",
+  "status": "degraded",
+  "embedding": {"available": false, "error": "Project does not have access to model `text-embedding-3-small`"},
+  "warnings": ["Embedding model unavailable; knowledge search degraded to local text matching. Set AIUI_EMBEDDING_MODEL to an accessible model."]
+}
+```
+
+`POST /api/knowledge/search` indicates whether results came from vector search or a
+degraded local fallback:
+
+```json
+{
+  "results": [],
+  "count": 0,
+  "search_mode": "fallback",
+  "warning": "Vector search unavailable: Project does not have access to model text-embedding-3-small. Set AIUI_EMBEDDING_MODEL=text-embedding-ada-002 or enable embedding models in your provider project."
+}
+```
+
+A dashboard can show a warning banner whenever `search_mode == "fallback"` or
+`embedding.available == false`.
+
+## Model reference
+
+| Model | Notes |
+|-------|-------|
+| `text-embedding-3-small` | Default — best cost/quality when available |
+| `text-embedding-3-large` | Higher quality — may also be restricted |
+| `text-embedding-ada-002` | Legacy — widely available on older projects |

examples/python/16-gateway-integration/app.py

@@ -44,15 +44,18 @@ import sys
 # ── Import praisonaiui ──────────────────────────────────────
 import praisonaiui as aiui
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check gateway availability ──────────────────────────────
 try:
     from praisonaiui.integration import AIUIGateway
     from praisonaiagents import Agent
     GATEWAY_OK = True
 except ImportError as e:
-    print(f"⚠️  Gateway dependencies missing: {e}")
-    print("   Install with: pip install praisonai praisonaiagents")
-    print("   Falling back to standalone mode.")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Gateway dependencies missing: {e}")
+    safe_print("   Install with: pip install praisonai praisonaiagents")
+    safe_print("   Falling back to standalone mode.")
     GATEWAY_OK = False
 
 # ── Set dashboard style ─────────────────────────────────────
@@ -149,15 +152,15 @@ def _register_agents_in_dashboard():
             "model": agent_def["model"],
             "icon": agent_def["icon"],
         })
-        print(f"   ✓ Dashboard: {agent_def['icon']} {agent_def['name']}")
+        safe_print(f"   {icon('✓', '[OK]')} Dashboard: {agent_def['name']}")
 
 
 # ── Main ────────────────────────────────────────────────────
 async def main():
     host = os.getenv("HOST", "127.0.0.1")
     if GATEWAY_OK:
-        print("🚀 Starting with Gateway mode (full execution + streaming)")
-        print()
+        safe_print(f"{icon('🚀', '[START]')} Starting with Gateway mode (full execution + streaming)")
+        safe_print()
 
         # Create the gateway
         gateway = AIUIGateway(
@@ -174,22 +177,22 @@ async def main():
                 self_reflect=False,
             )
             gateway.register_agent(agent, agent_id=agent_def["agent_id"])
-            print(f"   ✓ Gateway:   {agent_def['icon']} {agent_def['name']} ({agent_def['model']})")
+            safe_print(f"   {icon('✓', '[OK]')} Gateway:   {agent_def['name']} ({agent_def['model']})")
 
         # ── Register agents in dashboard CRUD storage ───────
         _register_agents_in_dashboard()
 
-        print()
-        print(f"   Dashboard:  http://localhost:8083")
-        print(f"   WebSocket:  ws://localhost:8083/ws")
-        print(f"   Inspector:  http://localhost:8083 → 🔍 Feature Inspector")
-        print()
+        safe_print()
+        safe_print("   Dashboard:  http://localhost:8083")
+        safe_print("   WebSocket:  ws://localhost:8083/ws")
+        safe_print(f"   Inspector:  http://localhost:8083 {icon('→', '->')} {icon('🔍', '')} Feature Inspector")
+        safe_print()
 
         # Start the gateway — this runs the full server
         await gateway.start()
     else:
         # Fallback to standalone mode (same as example 15)
-        print("📦 Starting in standalone mode (CRUD only, no agent execution)")
+        safe_print(f"{icon('📦', '[INFO]')} Starting in standalone mode (CRUD only, no agent execution)")
         from praisonaiui.server import create_app
         import uvicorn
 

examples/python/17-three-column-demo/app.py

@@ -26,15 +26,18 @@ import sys
 
 import praisonaiui as aiui
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check gateway availability ──────────────────────────────
 try:
     from praisonaiui.integration import AIUIGateway
     from praisonaiagents import Agent
     GATEWAY_OK = True
 except ImportError as e:
-    print(f"⚠️  Gateway dependencies not found: {e}")
-    print("   Install with: pip install praisonai praisonaiagents")
-    print("   Falling back to standalone mode (REST APIs only).\n")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Gateway dependencies not found: {e}")
+    safe_print("   Install with: pip install praisonai praisonaiagents")
+    safe_print("   Falling back to standalone mode (REST APIs only).\n")
     GATEWAY_OK = False
 
 aiui.set_style("dashboard")
@@ -115,7 +118,7 @@ def _register_agents_in_dashboard():
             "model": agent_def["model"],
             "icon": agent_def["icon"],
         })
-        print(f"   ✓ Agent: {agent_def['icon']} {agent_def['name']}")
+        safe_print(f"   {icon('✓', '[OK]')} Agent: {agent_def['name']}")
 
 
 # ── Main ────────────────────────────────────────────────────
@@ -125,8 +128,8 @@ async def main():
     host = os.getenv("HOST", "127.0.0.1")
 
     if GATEWAY_OK:
-        print("🚀 Starting with Gateway mode (full execution + streaming)")
-        print()
+        safe_print(f"{icon('🚀', '[START]')} Starting with Gateway mode (full execution + streaming)")
+        safe_print()
 
         gateway = AIUIGateway(host=host, port=port)
 
@@ -139,32 +142,32 @@ async def main():
                 reflection=False,
             )
             gateway.register_agent(agent, agent_id=agent_def["agent_id"])
-            print(f"   ✓ Gateway:  {agent_def['icon']} {agent_def['name']} ({agent_def['model']})")
+            safe_print(f"   {icon('✓', '[OK]')} Gateway:  {agent_def['name']} ({agent_def['model']})")
 
         # Also register in dashboard CRUD
         _register_agents_in_dashboard()
 
-        print()
-        print(f"   Dashboard:  http://localhost:{port}")
-        print(f"   Explorer:   http://localhost:{port} → 🔬 Feature Explorer")
-        print(f"   WebSocket:  ws://localhost:{port}/ws")
-        print()
+        safe_print()
+        safe_print(f"   Dashboard:  http://localhost:{port}")
+        safe_print(f"   Explorer:   http://localhost:{port} {icon('→', '->')} {icon('🔬', '')} Feature Explorer")
+        safe_print(f"   WebSocket:  ws://localhost:{port}/ws")
+        safe_print()
 
         await gateway.start()
     else:
         # Standalone: REST APIs only, no agent execution
-        print("📦 Starting in standalone mode (REST APIs only, no agent execution)")
+        safe_print(f"{icon('📦', '[INFO]')} Starting in standalone mode (REST APIs only, no agent execution)")
         from praisonaiui.server import create_app
         import uvicorn
 
         _register_agents_in_dashboard()
 
         app = create_app()
-        print()
-        print(f"   Dashboard:  http://localhost:{port}")
-        print(f"   Explorer:   http://localhost:{port} → 🔬 Feature Explorer")
-        print(f"   Note:       POST endpoints may fail without gateway")
-        print()
+        safe_print()
+        safe_print(f"   Dashboard:  http://localhost:{port}")
+        safe_print(f"   Explorer:   http://localhost:{port} {icon('→', '->')} {icon('🔬', '')} Feature Explorer")
+        safe_print("   Note:       POST endpoints may fail without gateway")
+        safe_print()
 
         config = uvicorn.Config(app, host=host, port=port, log_level="info")
         server = uvicorn.Server(config)

examples/python/18-full-chat/app.py

@@ -25,9 +25,13 @@ Usage:
 
 import asyncio
 import os
+import sys
 
 import praisonaiui as aiui
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check gateway availability ──────────────────────────────
 try:
     from praisonaiagents import Agent
@@ -35,9 +39,9 @@ try:
     from praisonaiui.integration import AIUIGateway
     GATEWAY_OK = True
 except ImportError as e:
-    print(f"⚠️  Gateway dependencies not found: {e}")
-    print("   Install with: pip install praisonai praisonaiagents")
-    print("   Falling back to standalone mode (REST APIs only).\n")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Gateway dependencies not found: {e}")
+    safe_print("   Install with: pip install praisonai praisonaiagents")
+    safe_print("   Falling back to standalone mode (REST APIs only).\n")
     GATEWAY_OK = False
 
 aiui.set_style("dashboard")
@@ -127,7 +131,7 @@ def _register_agents_in_dashboard():
             "model": agent_def["model"],
             "icon": agent_def["icon"],
         })
-        print(f"   ✓ Agent: {agent_def['icon']} {agent_def['name']}")
+        safe_print(f"   {icon('✓', '[OK]')} Agent: {agent_def['name']}")
 
 
 # ── Main ────────────────────────────────────────────────────
@@ -137,8 +141,8 @@ async def main():
     host = os.getenv("HOST", "127.0.0.1")
 
     if GATEWAY_OK:
-        print("🚀 Starting with Gateway mode (full execution + streaming)")
-        print()
+        safe_print(f"{icon('🚀', '[START]')} Starting with Gateway mode (full execution + streaming)")
+        safe_print()
 
         gateway = AIUIGateway(host=host, port=port)
 
@@ -151,21 +155,21 @@ async def main():
                 reflection=False,
             )
             gateway.register_agent(agent, agent_id=agent_def["agent_id"])
-            print(f"   ✓ Gateway:  {agent_def['icon']} {agent_def['name']} ({agent_def['model']})")
+            safe_print(f"   {icon('✓', '[OK]')} Gateway:  {agent_def['name']} ({agent_def['model']})")
 
         # Also register in dashboard CRUD
         _register_agents_in_dashboard()
 
-        print()
-        print(f"   Dashboard:  http://localhost:{port}")
-        print(f"   Explorer:   http://localhost:{port} → 🔬 Feature Explorer")
-        print(f"   WebSocket:  ws://localhost:{port}/ws")
-        print()
+        safe_print()
+        safe_print(f"   Dashboard:  http://localhost:{port}")
+        safe_print(f"   Explorer:   http://localhost:{port} {icon('→', '->')} {icon('🔬', '')} Feature Explorer")
+        safe_print(f"   WebSocket:  ws://localhost:{port}/ws")
+        safe_print()
 
         await gateway.start()
     else:
         # Standalone: REST APIs only, no agent execution
-        print("📦 Starting in standalone mode (REST APIs only, no agent execution)")
+        safe_print(f"{icon('📦', '[INFO]')} Starting in standalone mode (REST APIs only, no agent execution)")
         import uvicorn
 
         from praisonaiui.server import create_app
@@ -173,11 +177,11 @@ async def main():
         _register_agents_in_dashboard()
 
         app = create_app()
-        print()
-        print(f"   Dashboard:  http://localhost:{port}")
-        print(f"   Explorer:   http://localhost:{port} → 🔬 Feature Explorer")
-        print("   Note:       POST endpoints may fail without gateway")
-        print()
+        safe_print()
+        safe_print(f"   Dashboard:  http://localhost:{port}")
+        safe_print(f"   Explorer:   http://localhost:{port} {icon('→', '->')} {icon('🔬', '')} Feature Explorer")
+        safe_print("   Note:       POST endpoints may fail without gateway")
+        safe_print()
 
         config = uvicorn.Config(app, host=host, port=port, log_level="info")
         server = uvicorn.Server(config)

examples/python/20-email-bot/app.py

@@ -32,6 +32,9 @@ import asyncio
 import os
 import sys
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check prerequisites ─────────────────────────────────────────────
 
 try:
@@ -40,8 +43,8 @@ try:
     SDK_OK = True
 except ImportError:
     SDK_OK = False
-    print("⚠️  Missing dependencies. Install with:")
-    print("   pip install praisonai[email] praisonaiagents")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Missing dependencies. Install with:")
+    safe_print("   pip install praisonai[email] praisonaiagents")
     sys.exit(1)
 
 
@@ -68,8 +71,8 @@ agent = Agent(
 async def run_simple():
     """Start email bot with automatic env var resolution."""
     bot = Bot("email", agent=agent)
-    print(f"📧 Starting email bot for: {os.getenv('EMAIL_ADDRESS', '(not set)')}")
-    print("   Listening for new emails... (Ctrl+C to stop)")
+    safe_print(f"{icon('📧', '[EMAIL]')} Starting email bot for: {os.getenv('EMAIL_ADDRESS', '(not set)')}")
+    safe_print("   Listening for new emails... (Ctrl+C to stop)")
     await bot.start()
 
 
@@ -96,9 +99,9 @@ async def run_with_handlers():
     @bot.on_message
     async def log_email(message):
         subject = message.metadata.get("subject", "(no subject)")
-        print(f"📨 From: {message.sender.username}")
-        print(f"   Subject: {subject}")
-        print(f"   Preview: {message.text[:100]}...")
+        safe_print(f"{icon('📨', '[MSG]')} From: {message.sender.username}")
+        safe_print(f"   Subject: {subject}")
+        safe_print(f"   Preview: {message.text[:100]}...")
 
     # Custom command: emails with subject starting with "/status"
     @bot.on_command("status")
@@ -118,11 +121,11 @@ async def run_with_handlers():
             reply_to=message.message_id,
         )
 
-    print(f"📧 Starting email bot with custom handlers")
-    print(f"   Email: {bot._email_address}")
-    print(f"   Poll interval: {config.polling_interval}s")
-    print(f"   Allowed senders: {'all' if not config.allowed_users else config.allowed_users}")
-    print("   Listening... (Ctrl+C to stop)")
+    safe_print(f"{icon('📧', '[EMAIL]')} Starting email bot with custom handlers")
+    safe_print(f"   Email: {bot._email_address}")
+    safe_print(f"   Poll interval: {config.polling_interval}s")
+    safe_print(f"   Allowed senders: {'all' if not config.allowed_users else config.allowed_users}")
+    safe_print("   Listening... (Ctrl+C to stop)")
     await bot.start()
 
 
@@ -131,17 +134,17 @@ async def run_with_handlers():
 async def run_probe():
     """Test email server connectivity."""
     bot = Bot("email", agent=agent)
-    print("🔍 Testing email server connectivity...")
+    safe_print(f"{icon('🔍', '[PROBE]')} Testing email server connectivity...")
     result = await bot.probe()
 
-    print(f"\n   Connection: {'✅ OK' if result.ok else '❌ FAILED'}")
-    print(f"   Platform:   {result.platform}")
+    safe_print(f"\n   Connection: {icon('✅', '[OK]') + ' OK' if result.ok else icon('❌', '[FAIL]') + ' FAILED'}")
+    safe_print(f"   Platform:   {result.platform}")
     if result.bot_username:
-        print(f"   Email:      {result.bot_username}")
+        safe_print(f"   Email:      {result.bot_username}")
     if result.elapsed_ms:
-        print(f"   Latency:    {result.elapsed_ms:.0f}ms")
+        safe_print(f"   Latency:    {result.elapsed_ms:.0f}ms")
     if result.error:
-        print(f"   Error:      {result.error}")
+        safe_print(f"   Error:      {result.error}")
 
 
 # ── Main ────────────────────────────────────────────────────────────

examples/python/21-email-channel/app.py

@@ -42,7 +42,9 @@ import sys
 # ── Imports ─────────────────────────────────────────────────────────
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "src"))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
+from _shared.console import icon, safe_print
 from praisonaiui.server import create_app
 import uvicorn
 
@@ -91,11 +93,11 @@ def seed_email_channel():
         })
         if r.status_code in (200, 201):
             channel_id = r.json().get("id", "unknown")
-            print(f"   ✅ Email channel registered: {channel_id}")
+            safe_print(f"   {icon('✅', '[OK]')} Email channel registered: {channel_id}")
         else:
-            print(f"   ⚠️  Channel registration: {r.status_code} — {r.json()}")
+            safe_print(f"   {icon('⚠️', '[WARN]')}  Channel registration: {r.status_code} - {r.json()}")
     else:
-        print("   ✓ Email channel already exists")
+        safe_print(f"   {icon('✓', '[OK]')} Email channel already exists")
 
     # Also seed other channels for comparison in dashboard
     for name, platform in [("Discord #general", "discord"), ("Telegram Support", "telegram")]:
@@ -106,7 +108,7 @@ def seed_email_channel():
                 "platform": platform,
                 "config": {},
             })
-            print(f"   ✓ Seeded {platform} channel")
+            safe_print(f"   {icon('✓', '[OK]')} Seeded {platform} channel")
 
 
 # ── Verify email platform support ──────────────────────────────────
@@ -122,8 +124,8 @@ def verify_email_support():
     if r.status_code == 200:
         platforms = r.json().get("platforms", [])
         email_supported = "email" in platforms
-        print(f"   Supported platforms: {platforms}")
-        print(f"   Email supported: {'✅ Yes' if email_supported else '❌ No — add email to SUPPORTED_PLATFORMS'}")
+        safe_print(f"   Supported platforms: {platforms}")
+        safe_print(f"   Email supported: {icon('✅', '[OK]') + ' Yes' if email_supported else icon('❌', '[X]') + ' No - add email to SUPPORTED_PLATFORMS'}")
         return email_supported
     return False
 
@@ -137,51 +139,51 @@ def test_channel_api():
     app = create_app()
     client = TestClient(app)
 
-    print("\n📋 Channel API Test")
-    print("=" * 40)
+    safe_print(f"\n{icon('📋', '[TEST]')} Channel API Test")
+    safe_print("=" * 40)
 
     # List channels
     r = client.get("/api/channels")
     channels = r.json().get("channels", [])
-    print(f"   GET /api/channels: {len(channels)} channels")
+    safe_print(f"   GET /api/channels: {len(channels)} channels")
     for ch in channels:
-        icon = "📧" if ch.get("platform") == "email" else "💬"
-        print(f"      {icon} {ch.get('name', '?')} ({ch.get('platform', '?')})")
+        ch_icon = icon("📧", "[email]") if ch.get("platform") == "email" else icon("💬", "[chat]")
+        safe_print(f"      {ch_icon} {ch.get('name', '?')} ({ch.get('platform', '?')})")
 
     # Get email channel status
     email_ch = next((ch for ch in channels if ch.get("platform") == "email"), None)
     if email_ch:
         ch_id = email_ch.get("id")
         r = client.get(f"/api/channels/{ch_id}/status")
-        print(f"   GET /api/channels/{ch_id}/status: {r.json()}")
+        safe_print(f"   GET /api/channels/{ch_id}/status: {r.json()}")
 
-    print()
+    safe_print()
 
 
 # ── Main ────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
-    print("📧 PraisonAIUI — Email Channel Integration")
-    print("=" * 50)
-    print()
+    safe_print(f"{icon('📧', '[EMAIL]')} PraisonAIUI - Email Channel Integration")
+    safe_print("=" * 50)
+    safe_print()
 
     if "--test" in sys.argv:
         # Test mode: verify APIs + seed data
-        print("🔍 Verifying email platform support...")
+        safe_print(f"{icon('🔍', '[VERIFY]')} Verifying email platform support...")
         verify_email_support()
-        print()
-        print("🌱 Seeding email channel...")
+        safe_print()
+        safe_print(f"{icon('🌱', '[SEED]')} Seeding email channel...")
         seed_email_channel()
         test_channel_api()
     else:
         # Run the dashboard
-        print("🌱 Seeding channels...")
+        safe_print(f"{icon('🌱', '[SEED]')} Seeding channels...")
         seed_email_channel()
-        print()
+        safe_print()
 
         app = create_app()
-        print(f"✅ Dashboard at http://localhost:8084")
-        print(f"   Channels: http://localhost:8084/api/channels")
-        print(f"   Platforms: http://localhost:8084/api/channels/platforms")
+        safe_print(f"{icon('✅', '[OK]')} Dashboard at http://localhost:8084")
+        safe_print("   Channels: http://localhost:8084/api/channels")
+        safe_print("   Platforms: http://localhost:8084/api/channels/platforms")
         host = os.getenv("HOST", "127.0.0.1")
         uvicorn.run(app, host=host, port=8084, log_level="info")

examples/python/22-agentmail-bot/app.py

@@ -33,6 +33,9 @@ import asyncio
 import os
 import sys
 
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+from _shared.console import icon, safe_print
+
 # ── Check prerequisites ─────────────────────────────────────────────
 
 try:
@@ -41,8 +44,8 @@ try:
     SDK_OK = True
 except ImportError:
     SDK_OK = False
-    print("⚠️  Missing dependencies. Install with:")
-    print("   pip install praisonai agentmail praisonaiagents")
+    safe_print(f"{icon('⚠️', '[WARN]')}  Missing dependencies. Install with:")
+    safe_print("   pip install praisonai agentmail praisonaiagents")
     sys.exit(1)
 
 
@@ -69,9 +72,9 @@ agent = Agent(
 async def run_simple():
     """Start agentmail bot with automatic env var resolution."""
     bot = Bot("agentmail", agent=agent)
-    print(f"📩 Starting AgentMail bot")
-    print(f"   API key: {'✅ set' if os.getenv('AGENTMAIL_API_KEY') else '❌ not set'}")
-    print("   Listening for new emails... (Ctrl+C to stop)")
+    safe_print(f"{icon('📩', '[MAIL]')} Starting AgentMail bot")
+    safe_print(f"   API key: {icon('✅', '[OK]') + ' set' if os.getenv('AGENTMAIL_API_KEY') else icon('❌', '[X]') + ' not set'}")
+    safe_print("   Listening for new emails... (Ctrl+C to stop)")
     await bot.start()
 
 
@@ -99,9 +102,9 @@ async def run_with_handlers():
     @bot.on_message
     async def log_email(message):
         subject = message.metadata.get("subject", "(no subject)")
-        print(f"📨 From: {message.sender.username}")
-        print(f"   Subject: {subject}")
-        print(f"   Preview: {message.text[:100]}...")
+        safe_print(f"{icon('📨', '[MSG]')} From: {message.sender.username}")
+        safe_print(f"   Subject: {subject}")
+        safe_print(f"   Preview: {message.text[:100]}...")
 
     # Custom command: emails with subject starting with "/status"
     @bot.on_command("status")
@@ -121,10 +124,10 @@ async def run_with_handlers():
             reply_to=message.message_id,
         )
 
-    print(f"📩 Starting AgentMail bot with custom handlers")
-    print(f"   Email: {bot.email_address or '(will be assigned on start)'}")
-    print(f"   Poll interval: {config.polling_interval}s")
-    print("   Listening... (Ctrl+C to stop)")
+    safe_print(f"{icon('📩', '[MAIL]')} Starting AgentMail bot with custom handlers")
+    safe_print(f"   Email: {bot.email_address or '(will be assigned on start)'}")
+    safe_print(f"   Poll interval: {config.polling_interval}s")
+    safe_print("   Listening... (Ctrl+C to stop)")
     await bot.start()
 
 
@@ -136,30 +139,30 @@ async def run_inbox_demo():
         token=os.environ.get("AGENTMAIL_API_KEY", ""),
     )
 
-    print("📩 AgentMail Inbox Lifecycle Demo")
-    print("=" * 50)
+    safe_print(f"{icon('📩', '[MAIL]')} AgentMail Inbox Lifecycle Demo")
+    safe_print("=" * 50)
 
     # Create inbox
-    print("\n1. Creating inbox...")
+    safe_print("\n1. Creating inbox...")
     inbox = await bot.create_inbox(domain=os.getenv("AGENTMAIL_DOMAIN"))
-    print(f"   ✅ Created: {inbox.get('email_address', 'unknown')}")
-    print(f"   ID: {inbox.get('id', 'unknown')}")
+    safe_print(f"   {icon('✅', '[OK]')} Created: {inbox.get('email_address', 'unknown')}")
+    safe_print(f"   ID: {inbox.get('id', 'unknown')}")
 
     # List inboxes
-    print("\n2. Listing inboxes...")
+    safe_print("\n2. Listing inboxes...")
     inboxes = await bot.list_inboxes()
-    print(f"   Found: {len(inboxes)} inbox(es)")
+    safe_print(f"   Found: {len(inboxes)} inbox(es)")
     for ib in inboxes:
-        print(f"      📧 {ib.get('email_address', '?')} (id: {ib.get('id', '?')})")
+        safe_print(f"      {icon('📧', '[EMAIL]')} {ib.get('email_address', '?')} (id: {ib.get('id', '?')})")
 
     # Delete inbox
     inbox_id = inbox.get("id", "")
     if inbox_id:
-        print(f"\n3. Deleting inbox {inbox_id}...")
+        safe_print(f"\n3. Deleting inbox {inbox_id}...")
         deleted = await bot.delete_inbox(inbox_id)
-        print(f"   {'✅ Deleted' if deleted else '❌ Failed to delete'}")
+        safe_print(f"   {icon('✅', '[OK]') + ' Deleted' if deleted else icon('❌', '[X]') + ' Failed to delete'}")
 
-    print("\n✅ Inbox lifecycle complete")
+    safe_print(f"\n{icon('✅', '[OK]')} Inbox lifecycle complete")
 
 
 # ── Probe: Test connectivity ────────────────────────────────────────
@@ -167,17 +170,17 @@ async def run_inbox_demo():
 async def run_probe():
     """Test AgentMail API connectivity."""
     bot = Bot("agentmail", agent=agent)
-    print("🔍 Testing AgentMail API connectivity...")
+    safe_print(f"{icon('🔍', '[PROBE]')} Testing AgentMail API connectivity...")
     result = await bot.probe()
 
-    print(f"\n   Connection: {'✅ OK' if result.ok else '❌ FAILED'}")
-    print(f"   Platform:   {result.platform}")
+    safe_print(f"\n   Connection: {icon('✅', '[OK]') + ' OK' if result.ok else icon('❌', '[FAIL]') + ' FAILED'}")
+    safe_print(f"   Platform:   {result.platform}")
     if result.bot_username:
-        print(f"   Email:      {result.bot_username}")
+        safe_print(f"   Email:      {result.bot_username}")
     if result.elapsed_ms:
-        print(f"   Latency:    {result.elapsed_ms:.0f}ms")
+        safe_print(f"   Latency:    {result.elapsed_ms:.0f}ms")
     if result.error:
-        print(f"   Error:      {result.error}")
+        safe_print(f"   Error:      {result.error}")
 
 
 # ── Main ────────────────────────────────────────────────────────────

examples/python/23-agentmail-channel/app.py

@@ -40,7 +40,9 @@ import sys
 # ── Imports ─────────────────────────────────────────────────────────
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "src"))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
+from _shared.console import icon, safe_print
 from praisonaiui.server import create_app
 import uvicorn
 
@@ -88,11 +90,11 @@ def seed_agentmail_channel():
         })
         if r.status_code in (200, 201):
             channel_id = r.json().get("id", "unknown")
-            print(f"   ✅ AgentMail channel registered: {channel_id}")
+            safe_print(f"   {icon('✅', '[OK]')} AgentMail channel registered: {channel_id}")
         else:
-            print(f"   ⚠️  Channel registration: {r.status_code} — {r.json()}")
+            safe_print(f"   {icon('⚠️', '[WARN]')}  Channel registration: {r.status_code} - {r.json()}")
     else:
-        print("   ✓ AgentMail channel already exists")
+        safe_print(f"   {icon('✓', '[OK]')} AgentMail channel already exists")
 
     # Also seed email + telegram for comparison in dashboard
     for name, platform in [("Support Email", "email"), ("Telegram Support", "telegram")]:
@@ -103,7 +105,7 @@ def seed_agentmail_channel():
                 "platform": platform,
                 "config": {},
             })
-            print(f"   ✓ Seeded {platform} channel")
+            safe_print(f"   {icon('✓', '[OK]')} Seeded {platform} channel")
 
 
 # ── Verify agentmail platform support ──────────────────────────────
@@ -120,9 +122,9 @@ def verify_agentmail_support():
         platforms = r.json().get("platforms", [])
         am_supported = "agentmail" in platforms
         email_supported = "email" in platforms
-        print(f"   Supported platforms: {platforms}")
-        print(f"   email supported: {'✅ Yes' if email_supported else '❌ No'}")
-        print(f"   agentmail supported: {'✅ Yes' if am_supported else '❌ No'}")
+        safe_print(f"   Supported platforms: {platforms}")
+        safe_print(f"   email supported: {icon('✅', '[OK]') + ' Yes' if email_supported else icon('❌', '[X]') + ' No'}")
+        safe_print(f"   agentmail supported: {icon('✅', '[OK]') + ' Yes' if am_supported else icon('❌', '[X]') + ' No'}")
         return am_supported
     return False
 
@@ -136,51 +138,51 @@ def test_channel_api():
     app = create_app()
     client = TestClient(app)
 
-    print("\n📋 Channel API Test")
-    print("=" * 40)
+    safe_print(f"\n{icon('📋', '[TEST]')} Channel API Test")
+    safe_print("=" * 40)
 
     # List channels
     r = client.get("/api/channels")
     channels = r.json().get("channels", [])
-    print(f"   GET /api/channels: {len(channels)} channels")
+    safe_print(f"   GET /api/channels: {len(channels)} channels")
     for ch in channels:
-        icon = {"email": "📧", "agentmail": "📩"}.get(ch.get("platform", ""), "💬")
-        print(f"      {icon} {ch.get('name', '?')} ({ch.get('platform', '?')})")
+        ch_icon = {"email": icon("📧", "[email]"), "agentmail": icon("📩", "[mail]")}.get(ch.get("platform", ""), icon("💬", "[chat]"))
+        safe_print(f"      {ch_icon} {ch.get('name', '?')} ({ch.get('platform', '?')})")
 
     # Get agentmail channel status
     am_ch = next((ch for ch in channels if ch.get("platform") == "agentmail"), None)
     if am_ch:
         ch_id = am_ch.get("id")
         r = client.get(f"/api/channels/{ch_id}/status")
-        print(f"   GET /api/channels/{ch_id}/status: {r.json()}")
+        safe_print(f"   GET /api/channels/{ch_id}/status: {r.json()}")
 
-    print()
+    safe_print()
 
 
 # ── Main ────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
-    print("📩 PraisonAIUI — AgentMail Channel Integration")
-    print("=" * 50)
-    print()
+    safe_print(f"{icon('📩', '[MAIL]')} PraisonAIUI - AgentMail Channel Integration")
+    safe_print("=" * 50)
+    safe_print()
 
     if "--test" in sys.argv:
         # Test mode: verify APIs + seed data
-        print("🔍 Verifying agentmail platform support...")
+        safe_print(f"{icon('🔍', '[VERIFY]')} Verifying agentmail platform support...")
         verify_agentmail_support()
-        print()
-        print("🌱 Seeding agentmail channel...")
+        safe_print()
+        safe_print(f"{icon('🌱', '[SEED]')} Seeding agentmail channel...")
         seed_agentmail_channel()
         test_channel_api()
     else:
         # Run the dashboard
-        print("🌱 Seeding channels...")
+        safe_print(f"{icon('🌱', '[SEED]')} Seeding channels...")
         seed_agentmail_channel()
-        print()
+        safe_print()
 
         app = create_app()
-        print(f"✅ Dashboard at http://localhost:8084")
-        print(f"   Channels: http://localhost:8084/api/channels")
-        print(f"   Platforms: http://localhost:8084/api/channels/platforms")
+        safe_print(f"{icon('✅', '[OK]')} Dashboard at http://localhost:8084")
+        safe_print("   Channels: http://localhost:8084/api/channels")
+        safe_print("   Platforms: http://localhost:8084/api/channels/platforms")
         host = os.getenv("HOST", "127.0.0.1")
         uvicorn.run(app, host=host, port=8084, log_level="info")

examples/python/_shared/console.py

@@ -0,0 +1,40 @@
+"""Cross-platform console output for examples.
+
+Legacy Windows consoles default stdout to a non-UTF code page (cp1252/cp437)
+that cannot encode emoji such as U+1F680. Calling ``print`` with those
+characters raises ``UnicodeEncodeError`` and crashes the example before the
+server starts. These helpers mirror the ``_supports_unicode``/``_icon`` pattern
+in ``praisonaiui.cli`` so examples degrade to ASCII instead of crashing while
+preserving emoji on UTF-aware terminals.
+
+Usage:
+    from _shared.console import safe_print, icon
+
+    safe_print(f"{icon('🚀', '[START]')} Starting gateway...")
+"""
+
+from __future__ import annotations
+
+import sys
+from typing import Any
+
+
+def supports_unicode() -> bool:
+    try:
+        encoding = sys.stdout.encoding or ""
+    except Exception:
+        return False
+    return "utf" in encoding.lower()
+
+
+def icon(symbol: str, fallback: str) -> str:
+    return symbol if supports_unicode() else fallback
+
+
+def safe_print(*args: Any, **kwargs: Any) -> None:
+    sep = kwargs.pop("sep", " ")
+    text = sep.join(str(a) for a in args)
+    try:
+        print(text, **kwargs)
+    except UnicodeEncodeError:
+        print(text.encode("ascii", errors="replace").decode("ascii"), **kwargs)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "aiui"
-version = "0.3.134"
+version = "0.3.136"
 description = "YAML-driven website generator - CLI and compiler"
 readme = "README.md"
 license = {text = "MIT"}

src/praisonaiui/__version__.py

@@ -1,3 +1,3 @@
 """Version information."""
 
-__version__ = "0.3.134"
+__version__ = "0.3.136"

src/praisonaiui/features/knowledge.py

@@ -27,6 +27,17 @@ from ._base import BaseFeatureProtocol
 logger = logging.getLogger(__name__)
 
 
+def _is_embedding_unavailable(err: BaseException) -> bool:
+    """Detect an embedding model access / not-found failure."""
+    msg = str(err).lower()
+    return (
+        "does not have access" in msg
+        or "model_not_found" in msg
+        or "does not exist" in msg
+        or "notfounderror" in msg
+    )
+
+
 # ── Knowledge Protocol ───────────────────────────────────────────────
 
 
@@ -95,6 +106,10 @@ class KnowledgeProtocol(ABC):
         """Health check for this backend."""
         return {"status": "ok", "provider": self.__class__.__name__}
 
+    def last_search_meta(self) -> Dict[str, Any]:
+        """Metadata about the most recent search (mode, error)."""
+        return {"mode": "vector", "error": None}
+
 
 # ── Simple Knowledge Manager (Default, no deps) ─────────────────────
 
@@ -171,8 +186,12 @@ class SimpleKnowledgeManager(KnowledgeProtocol):
             "status": "ok",
             "provider": "SimpleKnowledgeManager",
             "total_entries": len(self._entries),
+            "embedding": {"available": False, "reason": "no_vector_backend"},
         }
 
+    def last_search_meta(self) -> Dict[str, Any]:
+        return {"mode": "local", "error": None}
+
 
 # ── SDK Knowledge Manager (wraps praisonaiagents.knowledge) ──────────
 
@@ -191,6 +210,9 @@ class SDKKnowledgeManager(KnowledgeProtocol):
         self._sdk_knowledge = None
         self._sdk_probed = False  # True once we've verified SDK actually works
         self._local_index: Dict[str, Dict[str, Any]] = {}
+        self._embedding_unavailable = False
+        self._embedding_error: Optional[str] = None
+        self._last_search_meta: Dict[str, Any] = {"mode": "vector", "error": None}
 
     def _get_sdk_knowledge(self) -> Any:
         """Lazy-init SDK Knowledge with health probe.
@@ -242,6 +264,10 @@ class SDKKnowledgeManager(KnowledgeProtocol):
                         "using local in-memory index"
                     )
                     self._sdk_knowledge = None
+                elif _is_embedding_unavailable(e):
+                    self._embedding_unavailable = True
+                    self._embedding_error = err_msg
+                    logger.warning("Embedding model unavailable: %s", e)
                 else:
                     # Other errors (e.g. empty DB) are fine — SDK works
                     pass
@@ -294,6 +320,7 @@ class SDKKnowledgeManager(KnowledgeProtocol):
             try:
                 results = sdk.search(query, limit=limit, user_id="praisonaiui")
                 if isinstance(results, list) and results:
+                    self._last_search_meta = {"mode": "vector", "error": None}
                     return [
                         {
                             "id": str(r.get("id", i)),
@@ -305,6 +332,9 @@ class SDKKnowledgeManager(KnowledgeProtocol):
                     ]
             except BaseException as e:
                 logger.warning("SDK knowledge search failed: %s; falling back to local", e)
+                if _is_embedding_unavailable(e):
+                    self._embedding_unavailable = True
+                    self._embedding_error = str(e)
 
         # Fallback to local text search
         query_lower = query.lower()
@@ -314,8 +344,26 @@ class SDKKnowledgeManager(KnowledgeProtocol):
                 results.append(e)
             if len(results) >= limit:
                 break
+
+        if self._embedding_unavailable:
+            self._last_search_meta = {
+                "mode": "fallback",
+                "error": (
+                    "Vector search unavailable: "
+                    f"{self._embedding_error or 'embedding model not accessible'}. "
+                    "Set AIUI_EMBEDDING_MODEL=text-embedding-ada-002 or enable embedding "
+                    "models in your provider project."
+                ),
+            }
+        elif sdk is not None:
+            self._last_search_meta = {"mode": "vector", "error": None}
+        else:
+            self._last_search_meta = {"mode": "local", "error": None}
         return results
 
+    def last_search_meta(self) -> Dict[str, Any]:
+        return dict(self._last_search_meta)
+
     def add_file(
         self,
         file_path: str,
@@ -453,11 +501,25 @@ class SDKKnowledgeManager(KnowledgeProtocol):
 
     def health(self) -> Dict[str, Any]:
         sdk = self._get_sdk_knowledge()
+        embedding_available = sdk is not None and not self._embedding_unavailable
+        warnings: List[str] = []
+        embedding: Dict[str, Any] = {"available": embedding_available}
+        if self._embedding_unavailable:
+            embedding["error"] = self._embedding_error
+            warnings.append(
+                "Embedding model unavailable; knowledge search degraded to local text "
+                "matching. Set AIUI_EMBEDDING_MODEL to an accessible model."
+            )
+        status = "ok"
+        if sdk is None or self._embedding_unavailable:
+            status = "degraded"
         return {
-            "status": "ok" if sdk is not None else "degraded",
+            "status": status,
             "provider": "SDKKnowledgeManager",
             "sdk_available": sdk is not None,
             "total_indexed": len(self._local_index),
+            "embedding": embedding,
+            "warnings": warnings,
         }
 
 
@@ -626,7 +688,15 @@ class KnowledgeFeature(BaseFeatureProtocol):
             query=query,
             limit=body.get("limit", 10),
         )
-        return JSONResponse({"results": results, "count": len(results)})
+        meta = mgr.last_search_meta()
+        return JSONResponse(
+            {
+                "results": results,
+                "count": len(results),
+                "search_mode": meta.get("mode", "unknown"),
+                "warning": meta.get("error"),
+            }
+        )
 
     async def _add_file(self, request: Request) -> JSONResponse:
         """POST /api/knowledge/add-file — ingest a file."""
@@ -675,6 +745,8 @@ class KnowledgeFeature(BaseFeatureProtocol):
                 "files": stats.get("files", 0),
                 "backend": h.get("provider", "unknown"),
                 "status": h.get("status", "ok"),
+                "embedding": h.get("embedding", {"available": True}),
+                "warnings": h.get("warnings", []),
             }
         )
 

src/praisonaiui/features/openai_api.py

@@ -6,6 +6,7 @@ Wraps praisonai.capabilities functions as HTTP endpoints.
 
 from __future__ import annotations
 
+import os
 import time
 import uuid
 from dataclasses import asdict
@@ -17,6 +18,35 @@ from starlette.routing import Route
 
 from ._base import BaseFeatureProtocol
 
+
+def _default_embedding_model() -> str:
+    """Resolve the default embedding model from environment."""
+    return os.environ.get(
+        "AIUI_EMBEDDING_MODEL",
+        os.environ.get("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small"),
+    )
+
+
+def _embedding_fallback_models() -> list[str]:
+    """Resolve the embedding fallback chain from environment."""
+    raw = os.environ.get(
+        "AIUI_EMBEDDING_FALLBACK_MODELS",
+        "text-embedding-ada-002,text-embedding-3-small",
+    )
+    return [m.strip() for m in raw.split(",") if m.strip()]
+
+
+def _is_model_not_found(err: BaseException) -> bool:
+    """Heuristically detect embedding model access / not-found errors."""
+    msg = str(err).lower()
+    return (
+        "does not have access" in msg
+        or "model_not_found" in msg
+        or "does not exist" in msg
+        or "notfounderror" in msg
+    )
+
+
 # Lazy import capabilities
 _capabilities = None
 
@@ -280,29 +310,57 @@ class OpenAIAPIFeature(BaseFeatureProtocol):
 
         body = await request.json()
         input_text = body.get("input", "")
-        model = body.get("model", "text-embedding-3-small")
+        model = body.get("model") or _default_embedding_model()
 
-        try:
-            result = await caps.aembed(
-                input=input_text,
-                model=model,
-            )
+        candidates = [model]
+        for fallback in _embedding_fallback_models():
+            if fallback not in candidates:
+                candidates.append(fallback)
 
-            data = _result_to_dict(result)
+        last_err: BaseException | None = None
+        for candidate in candidates:
+            try:
+                result = await caps.aembed(
+                    input=input_text,
+                    model=candidate,
+                )
+                data = _result_to_dict(result)
+                return JSONResponse(
+                    {
+                        "object": "list",
+                        "data": [
+                            {
+                                "object": "embedding",
+                                "index": 0,
+                                "embedding": data.get("embedding", []),
+                            }
+                        ],
+                        "model": candidate,
+                        "usage": data.get("usage", {"prompt_tokens": 0, "total_tokens": 0}),
+                    }
+                )
+            except Exception as e:
+                last_err = e
+                if not _is_model_not_found(e):
+                    break
+
+        if last_err is not None and _is_model_not_found(last_err):
             return JSONResponse(
                 {
-                    "object": "list",
-                    "data": [
-                        {"object": "embedding", "index": 0, "embedding": data.get("embedding", [])}
-                    ],
-                    "model": model,
-                    "usage": data.get("usage", {"prompt_tokens": 0, "total_tokens": 0}),
-                }
-            )
-        except Exception as e:
-            return JSONResponse(
-                {"error": {"message": str(e), "type": "api_error"}}, status_code=500
+                    "error": {
+                        "message": (
+                            f"Embedding model unavailable: {last_err}. "
+                            "Set AIUI_EMBEDDING_MODEL to a model your project can access "
+                            "(e.g. text-embedding-ada-002)."
+                        ),
+                        "type": "model_not_found",
+                    }
+                },
+                status_code=503,
             )
+        return JSONResponse(
+            {"error": {"message": str(last_err), "type": "api_error"}}, status_code=500
+        )
 
     async def _image_generate(self, request: Request) -> JSONResponse:
         """POST /v1/images/generations — Generate images."""

src/praisonaiui/server.py

@@ -2860,7 +2860,14 @@ def create_app(
     require_auth: bool = False,
     config_path: Optional[Path] = None,
 ) -> Starlette:
-    """Create the Starlette application."""
+    """Create the Starlette application.
+
+    URL-token auth (``TokenQueryMiddleware``) is opt-in: it activates only
+    when ``AIUI_REQUIRE_TOKEN`` is truthy, YAML ``auth.requireToken`` is set,
+    or ``require_auth=True``. A bare ``GATEWAY_AUTH_TOKEN`` /
+    ``AIUI_URL_TOKEN`` env var (e.g. leaked into a developer shell after a
+    gateway run) no longer forces 401 on standalone ``/api/*`` (#238, #251).
+    """
     from praisonaiui.auth import (
         AuthMiddleware,
         login_handler,
@@ -2874,10 +2881,13 @@ def create_app(
         config = load_config_from_yaml(config_path)
 
     # Extract auth settings from config
+    _config_requires_token = False
     if config:
         auth_config = config.get("auth", {})
         if auth_config.get("requireAuth") or auth_config.get("require_auth"):
             require_auth = True
+        if auth_config.get("requireToken") or auth_config.get("require_token"):
+            _config_requires_token = True
 
     middleware = [
         Middleware(
@@ -2891,10 +2901,20 @@ def create_app(
     # Add optional AUTH_ENFORCE middleware (env var driven)
     middleware.append(Middleware(AuthEnforcementMiddleware))
 
-    # URL-token auth — opt-in via env. Enables query-param / cookie auth so
-    # dashboards can be shared by URL (Jupyter model).
+    # URL-token auth — opt-in. Enables query-param / cookie auth so dashboards
+    # can be shared by URL (Jupyter model). The mere presence of a token env
+    # var (e.g. GATEWAY_AUTH_TOKEN leaking into a developer shell after a
+    # gateway run) must NOT force auth on standalone create_app() — that broke
+    # ad-hoc TestClient scripts (#238, #251). Activation now requires an
+    # explicit opt-in: AIUI_REQUIRE_TOKEN=1, YAML auth.requireToken, or the
+    # require_auth flag.
     _tok = os.environ.get("AIUI_URL_TOKEN") or os.environ.get("GATEWAY_AUTH_TOKEN")
-    if _tok:
+    _token_auth_opt_in = (
+        os.environ.get("AIUI_REQUIRE_TOKEN", "").lower() in ("true", "1", "yes")
+        or _config_requires_token
+        or require_auth
+    )
+    if _tok and _token_auth_opt_in:
         from praisonaiui.auth import TokenQueryMiddleware
         middleware.append(
             Middleware(TokenQueryMiddleware, expected_token=_tok)

tests/integration/test_cli.py

@@ -49,13 +49,16 @@ templates:
 routes:
   - match: "/docs/**"
     template: docs
-"""
+""",
+        encoding="utf-8",
     )
 
     docs_dir = tmp_path / "docs"
     docs_dir.mkdir()
-    (docs_dir / "index.md").write_text("# Welcome\n\nHello from integration test.")
-    (docs_dir / "guide.md").write_text("# Guide\n\nA test guide.")
+    (docs_dir / "index.md").write_text(
+        "# Welcome\n\nHello from integration test.", encoding="utf-8"
+    )
+    (docs_dir / "guide.md").write_text("# Guide\n\nA test guide.", encoding="utf-8")
 
     return tmp_path
 
@@ -94,7 +97,7 @@ class TestBuildCommand:
             ["build", "--config", str(project_dir / "aiui.template.yaml"), "--output", str(output)],
         )
 
-        ui_config = json.loads((output / "ui-config.json").read_text())
+        ui_config = json.loads((output / "ui-config.json").read_text(encoding="utf-8"))
         assert ui_config["site"]["title"] == "Integration Test Site"
         assert "header_main" in ui_config["components"]
         assert "footer_main" in ui_config["components"]
@@ -111,12 +114,12 @@ class TestBuildCommand:
             ],
         )
 
-        content = (output / "ui-config.json").read_text()
+        content = (output / "ui-config.json").read_text(encoding="utf-8")
         assert "\n" not in content  # Minified = no newlines
 
     def test_build_invalid_config(self, tmp_path: Path):
         config = tmp_path / "bad.yaml"
-        config.write_text("not: valid: yaml: config")
+        config.write_text("not: valid: yaml: config", encoding="utf-8")
         result = runner.invoke(
             app,
             ["build", "--config", str(config), "--output", str(tmp_path / "out")],
@@ -141,7 +144,8 @@ templates:
 routes:
   - match: "/docs/**"
     template: docs
-"""
+""",
+            encoding="utf-8",
         )
         result = runner.invoke(
             app,
@@ -215,8 +219,8 @@ class TestFullPipeline:
         assert (output / "index.html").exists()
 
         # Verify content
-        ui_config = json.loads((output / "ui-config.json").read_text())
+        ui_config = json.loads((output / "ui-config.json").read_text(encoding="utf-8"))
         assert ui_config["site"]["title"] == "Integration Test Site"
 
-        routes = json.loads((output / "route-manifest.json").read_text())
+        routes = json.loads((output / "route-manifest.json").read_text(encoding="utf-8"))
         assert len(routes["routes"]) >= 1

tests/unit/test_channels_view_crud.py

@@ -22,7 +22,7 @@ EXPLORER_VIEW = _FRONTEND / "views" / "explorer.js"
 
 
 def _view() -> str:
-    return CHANNELS_VIEW.read_text()
+    return CHANNELS_VIEW.read_text(encoding="utf-8")
 
 
 class TestAddChannelControls:
@@ -91,6 +91,6 @@ class TestEmptyState:
 
 class TestExplorerPreset:
     def test_explorer_has_add_channel_post(self):
-        src = EXPLORER_VIEW.read_text()
+        src = EXPLORER_VIEW.read_text(encoding="utf-8")
         assert "Add Channel" in src
         assert "env:TELEGRAM_BOT_TOKEN" in src

tests/unit/test_chat_offline_error.py

@@ -25,13 +25,13 @@ CHAT_JS = FRONTEND / "views" / "chat.js"
 
 class TestSelectPageShellError:
     def test_shell_error_helper_defined(self):
-        src = DASHBOARD_JS.read_text()
+        src = DASHBOARD_JS.read_text(encoding="utf-8")
         assert "function showDashboardInitError(" in src
         assert "'alert'" in src
         assert "db-error" in src
 
     def test_select_page_calls_shell_error_on_missing_main(self):
-        src = DASHBOARD_JS.read_text()
+        src = DASHBOARD_JS.read_text(encoding="utf-8")
         idx = src.find("const main = document.getElementById('db-main-content')")
         assert idx != -1
         region = src[idx:idx + 400]
@@ -41,16 +41,16 @@ class TestSelectPageShellError:
 
 class TestChatOfflineMessaging:
     def test_toast_imported(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert "import { showToast } from '../toast.js';" in src
 
     def test_offline_hint_in_fallback(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert "navigator.onLine === false" in src
         assert "You are offline. " in src
 
     def test_fallback_catch_shows_toast(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         idx = src.find("fetch('/api/chat/send'")
         assert idx != -1
         region = src[idx:idx + 900]
@@ -62,7 +62,7 @@ class TestChatOfflineMessaging:
 
 class TestSystemErrorRoleAlert:
     def test_append_message_sets_role_alert(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         idx = src.find("function appendMessage(")
         assert idx != -1
         region = src[idx:idx + 700]
@@ -72,7 +72,7 @@ class TestSystemErrorRoleAlert:
 
 class TestWebSocketSendGuard:
     def test_ws_send_wrapped_in_try_catch(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         idx = src.find("ws.send(JSON.stringify(payload))")
         assert idx != -1
         before = src[max(0, idx - 120):idx]

tests/unit/test_code_studio.py

@@ -50,28 +50,28 @@ class TestCodeStudioFrontend:
         assert (VIEWS / "code-studio.js").is_file()
 
     def test_view_exports_render(self):
-        src = (VIEWS / "code-studio.js").read_text()
+        src = (VIEWS / "code-studio.js").read_text(encoding="utf-8")
         assert "export async function render" in src
         assert "export function cleanup" in src
 
     def test_view_calls_execute_api(self):
-        src = (VIEWS / "code-studio.js").read_text()
+        src = (VIEWS / "code-studio.js").read_text(encoding="utf-8")
         assert "/api/code/execute" in src
         assert "/api/code/languages" in src
 
     def test_no_browser_eval(self):
-        src = (VIEWS / "code-studio.js").read_text()
+        src = (VIEWS / "code-studio.js").read_text(encoding="utf-8")
         assert "eval(" not in src
 
     def test_registered_in_dashboard(self):
-        src = (PLUGINS / "dashboard.js").read_text()
+        src = (PLUGINS / "dashboard.js").read_text(encoding="utf-8")
         assert "'code-studio'" in src
         assert "code-studio.js" in src
 
     def test_send_to_agent_event(self):
-        src = (VIEWS / "code-studio.js").read_text()
+        src = (VIEWS / "code-studio.js").read_text(encoding="utf-8")
         assert "aiui:prefill-composer" in src
 
     def test_chat_has_prefill_listener(self):
-        src = (VIEWS / "chat.js").read_text()
+        src = (VIEWS / "chat.js").read_text(encoding="utf-8")
         assert "aiui:prefill-composer" in src

tests/unit/test_command_palette.py

@@ -22,7 +22,7 @@ DASHBOARD_JS = FRONTEND / "dashboard.js"
 
 class TestCommandPalette:
     def _src(self) -> str:
-        return DASHBOARD_JS.read_text()
+        return DASHBOARD_JS.read_text(encoding="utf-8")
 
     def test_open_command_palette_exposed_globally(self):
         src = self._src()

tests/unit/test_compiler.py

@@ -86,7 +86,7 @@ class TestCompilerUiConfig:
         result = compiler.compile(tmp_path / "output")
         assert result.success is True
 
-        ui_config = json.loads((tmp_path / "output" / "ui-config.json").read_text())
+        ui_config = json.loads((tmp_path / "output" / "ui-config.json").read_text(encoding="utf-8"))
         assert ui_config["site"]["title"] == "Test Site"
         assert "templates" in ui_config
         assert "docs" in ui_config["templates"]
@@ -96,7 +96,7 @@ class TestCompilerUiConfig:
         result = compiler.compile(tmp_path / "output")
         assert result.success is True
 
-        ui_config = json.loads((tmp_path / "output" / "ui-config.json").read_text())
+        ui_config = json.loads((tmp_path / "output" / "ui-config.json").read_text(encoding="utf-8"))
         assert "header_main" in ui_config["components"]
         assert ui_config["components"]["header_main"]["props"]["logoText"] == "MySite"
         assert "footer_main" in ui_config["components"]
@@ -107,7 +107,7 @@ class TestCompilerUiConfig:
         result = compiler.compile(tmp_path / "output")
         assert result.success is True
 
-        ui_config = json.loads((tmp_path / "output" / "ui-config.json").read_text())
+        ui_config = json.loads((tmp_path / "output" / "ui-config.json").read_text(encoding="utf-8"))
         docs_template = ui_config["templates"]["docs"]
         assert docs_template["slots"]["header"]["ref"] == "header_main"
         assert docs_template["slots"]["footer"]["ref"] == "footer_main"
@@ -123,7 +123,7 @@ class TestCompilerRouteManifest:
         assert "route-manifest.json" in result.files
 
         manifest = json.loads(
-            (tmp_path / "output" / "route-manifest.json").read_text()
+            (tmp_path / "output" / "route-manifest.json").read_text(encoding="utf-8")
         )
         assert len(manifest["routes"]) == 1
         assert manifest["routes"][0]["pattern"] == "/docs/**"
@@ -147,7 +147,7 @@ class TestCompilerRouteManifest:
         assert result.success is True
 
         manifest = json.loads(
-            (tmp_path / "output" / "route-manifest.json").read_text()
+            (tmp_path / "output" / "route-manifest.json").read_text(encoding="utf-8")
         )
         # First route in list gets highest priority
         assert manifest["routes"][0]["priority"] > manifest["routes"][1]["priority"]
@@ -200,7 +200,7 @@ class TestCompilerOutputFiles:
         compiler = Compiler(minimal_config, base_path=tmp_path)
         compiler.compile(tmp_path / "output", minify=True)
 
-        content = (tmp_path / "output" / "ui-config.json").read_text()
+        content = (tmp_path / "output" / "ui-config.json").read_text(encoding="utf-8")
         # Minified JSON should not have newlines
         assert "\n" not in content
 
@@ -208,7 +208,7 @@ class TestCompilerOutputFiles:
         compiler = Compiler(minimal_config, base_path=tmp_path)
         compiler.compile(tmp_path / "output", minify=False)
 
-        content = (tmp_path / "output" / "ui-config.json").read_text()
+        content = (tmp_path / "output" / "ui-config.json").read_text(encoding="utf-8")
         # Non-minified JSON should have newlines
         assert "\n" in content
 
@@ -228,7 +228,7 @@ class TestCompilerDocsContent:
         result = compiler.compile(tmp_path / "output")
         assert "docs-nav.json" in result.files
 
-        nav = json.loads((tmp_path / "output" / "docs-nav.json").read_text())
+        nav = json.loads((tmp_path / "output" / "docs-nav.json").read_text(encoding="utf-8"))
         assert "items" in nav
 
     def test_no_docs_nav_without_content(self, minimal_config: Config, tmp_path: Path):
@@ -291,7 +291,7 @@ class TestCompilerThemeCss:
         assert result.success is True
         theme_css = tmp_path / "output" / "assets" / "theme.css"
         assert theme_css.exists(), "assets/theme.css must be generated"
-        content = theme_css.read_text()
+        content = theme_css.read_text(encoding="utf-8")
         assert ":root {" in content
         assert "--radius:" in content
 
@@ -340,7 +340,7 @@ class TestInjectThemeCss:
         inject_theme_css(tmp_path, preset="zinc", dark_mode=True, radius="0.5rem")
         theme_file = tmp_path / "assets" / "theme.css"
         assert theme_file.exists()
-        content = theme_file.read_text()
+        content = theme_file.read_text(encoding="utf-8")
         assert ":root {" in content
 
     def test_creates_assets_dir(self, tmp_path: Path):
@@ -367,7 +367,7 @@ class TestThemePipelineIntegration:
         assert result.success is True
         assert "assets/theme.css" in result.files
 
-        theme_css = (tmp_path / "output" / "assets" / "theme.css").read_text()
+        theme_css = (tmp_path / "output" / "assets" / "theme.css").read_text(encoding="utf-8")
         assert ":root {" in theme_css
         assert ".dark {" in theme_css
         assert "--radius: 1rem" in theme_css  # xl = 1rem
@@ -386,7 +386,7 @@ class TestThemePipelineIntegration:
         result = compiler.compile(tmp_path / "output")
         assert result.success is True
 
-        theme_css = (tmp_path / "output" / "assets" / "theme.css").read_text()
+        theme_css = (tmp_path / "output" / "assets" / "theme.css").read_text(encoding="utf-8")
         assert ":root {" in theme_css
         assert ".dark" not in theme_css
 
@@ -405,7 +405,7 @@ class TestThemePipelineIntegration:
         assert result.success is True
 
         ui_config = json.loads(
-            (tmp_path / "output" / "ui-config.json").read_text()
+            (tmp_path / "output" / "ui-config.json").read_text(encoding="utf-8")
         )
         theme = ui_config["site"]["theme"]
         assert theme["preset"] == "blue"
@@ -426,7 +426,7 @@ class TestThemePipelineIntegration:
 
         theme_css_path = tmp_path / "output" / "assets" / "theme.css"
         assert theme_css_path.exists(), "theme.css must always be generated"
-        content = theme_css_path.read_text()
+        content = theme_css_path.read_text(encoding="utf-8")
         assert ":root {" in content
         assert "--radius:" in content
 
@@ -444,7 +444,7 @@ class TestThemePipelineIntegration:
         result = compiler.compile(tmp_path / "output")
         assert result.success is True
 
-        index_html = (tmp_path / "output" / "index.html").read_text()
+        index_html = (tmp_path / "output" / "index.html").read_text(encoding="utf-8")
         assert "classList.add('dark')" not in index_html
 
     def test_dark_mode_antiflicker_has_dark_class(self, tmp_path: Path):
@@ -461,7 +461,7 @@ class TestThemePipelineIntegration:
         result = compiler.compile(tmp_path / "output")
         assert result.success is True
 
-        index_html = (tmp_path / "output" / "index.html").read_text()
+        index_html = (tmp_path / "output" / "index.html").read_text(encoding="utf-8")
         assert "classList.add('dark')" in index_html
 
 
@@ -514,7 +514,7 @@ class TestCompositionResolver:
 
         # Load ui-config.json and verify auto-bridging
         ui_config_path = tmp_path / "output" / "ui-config.json"
-        ui_config = json.loads(ui_config_path.read_text())
+        ui_config = json.loads(ui_config_path.read_text(encoding="utf-8"))
 
         docs_template = ui_config["templates"]["docs"]
         assert "zones" in docs_template
@@ -556,7 +556,7 @@ class TestCompositionResolver:
 
         # Load ui-config.json and verify no auto-bridging occurred
         ui_config_path = tmp_path / "output" / "ui-config.json"
-        ui_config = json.loads(ui_config_path.read_text())
+        ui_config = json.loads(ui_config_path.read_text(encoding="utf-8"))
 
         docs_template = ui_config["templates"]["docs"]
         # Should not have zones for non-FlexibleLayout
@@ -589,7 +589,7 @@ class TestCompositionResolver:
 
         # Load ui-config.json and verify no duplicate components
         ui_config_path = tmp_path / "output" / "ui-config.json"
-        ui_config = json.loads(ui_config_path.read_text())
+        ui_config = json.loads(ui_config_path.read_text(encoding="utf-8"))
 
         docs_template = ui_config["templates"]["docs"]
         zones = docs_template["zones"]

tests/unit/test_compiler_chat.py

@@ -30,7 +30,7 @@ class TestCompilerChatOutput:
         result = compiler.compile(tmp_path)
 
         assert result.success
-        ui_config = json.loads((tmp_path / "ui-config.json").read_text())
+        ui_config = json.loads((tmp_path / "ui-config.json").read_text(encoding="utf-8"))
         assert ui_config["style"] == "chat"
 
     def test_layout_in_ui_config(self, tmp_path: Path):
@@ -43,7 +43,7 @@ class TestCompilerChatOutput:
         result = compiler.compile(tmp_path)
 
         assert result.success
-        ui_config = json.loads((tmp_path / "ui-config.json").read_text())
+        ui_config = json.loads((tmp_path / "ui-config.json").read_text(encoding="utf-8"))
         assert ui_config["layout"]["mode"] == "bottom-right"
         assert ui_config["layout"]["width"] == "400px"
         assert ui_config["layout"]["height"] == "600px"
@@ -73,7 +73,7 @@ class TestCompilerChatOutput:
         result = compiler.compile(tmp_path)
 
         assert result.success
-        ui_config = json.loads((tmp_path / "ui-config.json").read_text())
+        ui_config = json.loads((tmp_path / "ui-config.json").read_text(encoding="utf-8"))
         assert ui_config["chat"]["enabled"] is True
         assert ui_config["chat"]["name"] == "AI Assistant"
         assert len(ui_config["chat"]["starters"]) == 1
@@ -96,7 +96,7 @@ class TestCompilerChatOutput:
         result = compiler.compile(tmp_path)
 
         assert result.success
-        ui_config = json.loads((tmp_path / "ui-config.json").read_text())
+        ui_config = json.loads((tmp_path / "ui-config.json").read_text(encoding="utf-8"))
         assert ui_config["auth"]["enabled"] is True
         assert "password" in ui_config["auth"]["providers"]
         assert "google" in ui_config["auth"]["providers"]
@@ -128,7 +128,7 @@ class TestCompilerChatOutput:
         result = compiler.compile(tmp_path)
 
         assert result.success
-        ui_config = json.loads((tmp_path / "ui-config.json").read_text())
+        ui_config = json.loads((tmp_path / "ui-config.json").read_text(encoding="utf-8"))
         assert len(ui_config["widgets"]) == 2
         assert ui_config["widgets"][0]["type"] == "slider"
         assert ui_config["widgets"][0]["name"] == "temperature"

tests/unit/test_component_registry.py

@@ -24,7 +24,7 @@ DASHBOARD_JS = (
 @pytest.fixture
 def dashboard_source():
     """Read dashboard.js source."""
-    return DASHBOARD_JS.read_text()
+    return DASHBOARD_JS.read_text(encoding="utf-8")
 
 
 class TestComponentRegistry:
@@ -68,7 +68,7 @@ class TestPythonJSComponentParity:
             Path(__file__).resolve().parents[2]
             / "src" / "praisonaiui" / "ui.py"
         )
-        source = ui_py.read_text()
+        source = ui_py.read_text(encoding="utf-8")
 
         # Extract all "type": "xxx" patterns from ui.py
         py_types = set(re.findall(r'"type":\s*"(\w+)"', source))

tests/unit/test_component_schemas.py

@@ -111,7 +111,7 @@ class TestBuiltinSchemaCoverage:
         ui_src = (
             Path(__file__).resolve().parents[2]
             / "src" / "praisonaiui" / "ui.py"
-        ).read_text()
+        ).read_text(encoding="utf-8")
         py_types = set(re.findall(r'"type":\s*"(\w+)"', ui_src))
         schemas = srv.get_component_schemas()
         missing = py_types - set(schemas)

tests/unit/test_context_ring.py

@@ -136,47 +136,47 @@ class TestRingThresholds:
 
 class TestChatJsComponents:
     def test_ring_dom_present(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert 'id="chat-ring-wrap"' in src
         assert 'id="chat-ring-arc"' in src
         assert 'id="chat-ring-tooltip"' in src
 
     def test_send_button_inside_ring_preserved_id(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert 'class="chat-ring-send" id="chat-send-btn"' in src
         assert 'aria-valuenow' in src
         assert 'aria-valuemax' in src
 
     def test_compaction_banner_present(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert 'id="chat-compaction-banner"' in src
         assert "function updateCompactionBanner(" in src
         assert "function dismissCompactionBanner(" in src
 
     def test_memory_chip_present(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert 'id="chat-memory-chip"' in src
         assert "function refreshMemoryChip(" in src
 
     def test_ring_state_machine_thresholds(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert "function ringStateFor(" in src
         assert "pct >= 95" in src
         assert "pct >= 85" in src
         assert "pct >= 60" in src
 
     def test_ws_context_events_handled(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert "case 'context_update':" in src
         assert "case 'context_compacted':" in src
 
     def test_poll_fallback_and_debounce(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert "function startContextPoll(" in src
         assert "function scheduleContextRefresh(" in src
 
     def test_existing_memory_panel_and_metrics_preserved(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         # Regression: do not remove existing shipped elements
         assert 'id="chat-memory-panel"' in src
         assert 'id="chat-memory-btn"' in src

tests/unit/test_dashboard_dropdown_contrast.py

@@ -27,7 +27,7 @@ DASHBOARD_JS = (
 
 
 def _style() -> str:
-    src = DASHBOARD_JS.read_text()
+    src = DASHBOARD_JS.read_text(encoding="utf-8")
     m = re.search(r"const DASHBOARD_STYLE = `(.*?)`;", src, re.DOTALL)
     assert m, "DASHBOARD_STYLE template literal not found"
     return m.group(1)

tests/unit/test_dashboard_select_page_guard.py

@@ -21,7 +21,7 @@ DASHBOARD_JS = FRONTEND / "dashboard.js"
 
 class TestSelectPageGuard:
     def _select_page_body(self) -> str:
-        src = DASHBOARD_JS.read_text()
+        src = DASHBOARD_JS.read_text(encoding="utf-8")
         m = re.search(
             r"async function selectPage\(pageId\)\s*\{(.*?)^}",
             src, re.DOTALL | re.MULTILINE,
@@ -56,7 +56,7 @@ class TestSelectPageGuard:
         assert "console.error" in m.group(1)
 
     def test_show_dashboard_init_error_helper_present(self):
-        src = DASHBOARD_JS.read_text()
+        src = DASHBOARD_JS.read_text(encoding="utf-8")
         m = re.search(
             r"function showDashboardInitError\(message\)\s*\{(.*?)^}",
             src, re.DOTALL | re.MULTILINE,

tests/unit/test_dashboard_shell_order.py

@@ -23,7 +23,7 @@ GENERATOR = ROOT / "scripts" / "patch_dashboard_modular.py"
 
 class TestDashboardShellOrder:
     def _build_body(self) -> str:
-        src = DASHBOARD_JS.read_text()
+        src = DASHBOARD_JS.read_text(encoding="utf-8")
         m = re.search(
             r"function buildDashboard\([^)]*\)\s*\{(.*?)^}",
             src, re.DOTALL | re.MULTILINE,
@@ -51,7 +51,7 @@ class TestDashboardShellOrder:
         assert m, "shellHeader insert must be guarded by children.length"
 
     def test_generator_emits_main_before_insert_before(self):
-        src = GENERATOR.read_text()
+        src = GENERATOR.read_text(encoding="utf-8")
         append_idx = src.find("root.appendChild(main);\n  const shellHeader")
         insert_idx = src.find("root.insertBefore(shellHeader, main)")
         assert append_idx != -1, (

tests/unit/test_examples_console.py

@@ -0,0 +1,70 @@
+"""Tests for the shared examples console helper.
+
+These verify that examples degrade emoji to ASCII on legacy Windows code pages
+(cp1252) instead of raising ``UnicodeEncodeError`` while preserving emoji on
+UTF-aware terminals, mirroring the CLI ``_icon``/``_supports_unicode`` pattern.
+"""
+
+import io
+import os
+import sys
+from unittest import mock
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "examples", "python", "_shared"))
+from console import icon, safe_print, supports_unicode
+
+
+class TestSupportsUnicode:
+    def test_utf8_encoding_is_supported(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = "utf-8"
+        with mock.patch("sys.stdout", mock_stdout):
+            assert supports_unicode() is True
+
+    def test_cp1252_encoding_is_not_supported(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = "cp1252"
+        with mock.patch("sys.stdout", mock_stdout):
+            assert supports_unicode() is False
+
+    def test_missing_encoding_falls_back_to_ascii(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = None
+        with mock.patch("sys.stdout", mock_stdout):
+            assert supports_unicode() is False
+
+
+class TestIcon:
+    def test_returns_symbol_on_utf8(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = "utf-8"
+        with mock.patch("sys.stdout", mock_stdout):
+            assert icon("🚀", "[START]") == "🚀"
+
+    def test_returns_fallback_on_cp1252(self):
+        mock_stdout = mock.MagicMock()
+        mock_stdout.encoding = "cp1252"
+        with mock.patch("sys.stdout", mock_stdout):
+            assert icon("🚀", "[START]") == "[START]"
+
+
+class TestSafePrint:
+    def test_never_raises_on_cp1252(self):
+        buffer = io.BytesIO()
+        stream = io.TextIOWrapper(buffer, encoding="cp1252", errors="strict")
+        with mock.patch("sys.stdout", stream):
+            safe_print("🚀 Starting")
+            stream.flush()
+        assert buffer.getvalue()
+
+    def test_preserves_unicode_on_utf8(self):
+        buffer = io.StringIO()
+        with mock.patch("sys.stdout", buffer):
+            safe_print("🚀 Starting")
+        assert "🚀" in buffer.getvalue()
+
+    def test_respects_sep(self):
+        buffer = io.StringIO()
+        with mock.patch("sys.stdout", buffer):
+            safe_print("a", "b", sep="-")
+        assert buffer.getvalue().strip() == "a-b"

tests/unit/test_finops_strip.py

@@ -154,13 +154,13 @@ class TestFinOpsValidation:
 
 class TestHelpersJs:
     def test_donut_and_strip_helpers_present(self):
-        src = HELPERS_JS.read_text()
+        src = HELPERS_JS.read_text(encoding="utf-8")
         assert "export function MiniDonutSVG(" in src
         assert "export function costStripHTML(" in src
         assert "export function budgetBannerHTML(" in src
 
     def test_math_helpers_present(self):
-        src = HELPERS_JS.read_text()
+        src = HELPERS_JS.read_text(encoding="utf-8")
         assert "export function computeBudgetPct(" in src
         assert "export function budgetLevelFor(" in src
         assert "export function sumTodayTokens(" in src
@@ -168,53 +168,53 @@ class TestHelpersJs:
         assert "export function formatCost(" in src
 
     def test_donut_empty_state_accessible(self):
-        src = HELPERS_JS.read_text()
+        src = HELPERS_JS.read_text(encoding="utf-8")
         assert 'aria-label="No model data"' in src
 
     def test_strip_navigates_and_is_region(self):
-        src = HELPERS_JS.read_text()
+        src = HELPERS_JS.read_text(encoding="utf-8")
         assert 'data-nav="usage"' in src
         assert 'role="region"' in src
 
 
 class TestOverviewJs:
     def test_finops_imports_and_render(self):
-        src = OVERVIEW_JS.read_text()
+        src = OVERVIEW_JS.read_text(encoding="utf-8")
         assert "costStripHTML" in src
         assert "budgetBannerHTML" in src
         assert "function renderFinOps(" in src
         assert 'id="ov-finops"' in src
 
     def test_fetches_models_and_config(self):
-        src = OVERVIEW_JS.read_text()
+        src = OVERVIEW_JS.read_text(encoding="utf-8")
         assert "/api/usage/models" in src
         assert "/api/config/runtime" in src
 
     def test_graceful_degrade_state_machine(self):
-        src = OVERVIEW_JS.read_text()
+        src = OVERVIEW_JS.read_text(encoding="utf-8")
         assert "function finopsState(" in src
         assert "finops.enabled === false" in src
 
 
 class TestChatJs:
     def test_session_cost_chip_present(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert 'id="chat-session-cost-chip"' in src
         assert "function updateSessionCostChip(" in src
 
     def test_finops_banner_present(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert 'id="chat-finops-banner"' in src
         assert "function updateFinopsBanner(" in src
         assert "function loadFinopsConfig(" in src
 
     def test_chip_updates_on_run_completed(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert "updateFinopsBanner()" in src
         assert "updateSessionCostChip(stats)" in src
 
     def test_context_ring_regression_preserved(self):
-        src = CHAT_JS.read_text()
+        src = CHAT_JS.read_text(encoding="utf-8")
         assert 'id="chat-ring-wrap"' in src
         assert "function ringStateFor(" in src
         assert 'class="chat-ring-send" id="chat-send-btn"' in src

tests/unit/test_form_action.py

@@ -108,7 +108,7 @@ class TestFormInputDataLabel:
             / "src" / "praisonaiui" / "templates" / "frontend"
             / "plugins" / "dashboard.js"
         )
-        return p.read_text()
+        return p.read_text(encoding="utf-8")
 
     @pytest.mark.parametrize("renderer", [
         "renderTextInput",

tests/unit/test_form_input_names.py

@@ -84,7 +84,7 @@ class TestRenderersEmitName:
     via `el.name` (which has priority over `el.dataset.label`)."""
 
     def _src(self):
-        return DASHBOARD_JS.read_text()
+        return DASHBOARD_JS.read_text(encoding="utf-8")
 
     @pytest.mark.parametrize("renderer", [
         "renderTextInput",
@@ -117,7 +117,7 @@ class TestSubmitCollectorPriority:
     """renderFormAction submit handler must prefer `name` over `data-label`."""
 
     def test_name_has_priority_over_label(self):
-        source = DASHBOARD_JS.read_text()
+        source = DASHBOARD_JS.read_text(encoding="utf-8")
         m = re.search(
             r"function renderFormAction\(comp\)\s*\{(.*?)^}",
             source, re.DOTALL | re.MULTILINE,

tests/unit/test_frontend_navigation.py

@@ -16,7 +16,7 @@ class TestFrontendNavigation:
         index_path = Path("src/praisonaiui/templates/frontend/index.html")
         assert index_path.exists(), "index.html should exist"
 
-        content = index_path.read_text()
+        content = index_path.read_text(encoding="utf-8")
         assert "/assets/index.js" in content, "Should reference index.js"
         assert "/assets/index.css" in content, "Should reference index.css"
 
@@ -26,7 +26,7 @@ class TestFrontendNavigation:
         if bundle_path.exists():
             # Check if the bundle contains popstate handling code
             # Note: The bundle is minified, so we check for the event name
-            content = bundle_path.read_text()
+            content = bundle_path.read_text(encoding="utf-8")
             assert "popstate" in content, "Bundle should contain popstate event handler"
 
     def test_title_template_handling(self):

tests/unit/test_inbox_view.py

@@ -23,7 +23,7 @@ SERVER = _ROOT / "src" / "praisonaiui" / "server.py"
 
 
 def _inbox() -> str:
-    return INBOX_VIEW.read_text()
+    return INBOX_VIEW.read_text(encoding="utf-8")
 
 
 class TestThreePaneShell:
@@ -109,21 +109,21 @@ class TestEmptyState:
 
 class TestRegistration:
     def test_route_registered_in_dashboard(self):
-        src = DASHBOARD.read_text()
+        src = DASHBOARD.read_text(encoding="utf-8")
         assert "inbox:" in src
         assert "/plugins/views/inbox.js" in src
 
     def test_page_registered_in_server(self):
-        src = SERVER.read_text()
+        src = SERVER.read_text(encoding="utf-8")
         assert '"id": "inbox"' in src
         assert "Channel Inbox" in src
 
     def test_channels_links_to_inbox(self):
-        assert 'href="/inbox"' in CHANNELS_VIEW.read_text()
+        assert 'href="/inbox"' in CHANNELS_VIEW.read_text(encoding="utf-8")
 
 
 class TestSessionsPlatformField:
     def test_sessions_list_surfaces_platform(self):
-        src = SESSIONS_EXT.read_text()
+        src = SESSIONS_EXT.read_text(encoding="utf-8")
         assert '"platform"' in src
         assert '"channel_id"' in src

tests/unit/test_knowledge_embeddings.py

@@ -0,0 +1,194 @@
+"""Unit tests for knowledge embedding fallback + degraded-mode reporting."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from starlette.testclient import TestClient
+
+from praisonaiui.features import _features, register_feature
+from praisonaiui.features import knowledge as knowledge_mod
+from praisonaiui.features.knowledge import (
+    KnowledgeFeature,
+    SDKKnowledgeManager,
+    SimpleKnowledgeManager,
+    _is_embedding_unavailable,
+)
+from praisonaiui.features.openai_api import (
+    OpenAIAPIFeature,
+    _default_embedding_model,
+    _embedding_fallback_models,
+    _is_model_not_found,
+)
+from praisonaiui.server import create_app
+
+
+class _FakeSDK:
+    def __init__(self, search_error: BaseException | None = None) -> None:
+        self._search_error = search_error
+
+    def search(self, query, limit=1, user_id=None):
+        if query == "__probe__":
+            return []
+        if self._search_error is not None:
+            raise self._search_error
+        return []
+
+
+def test_is_model_not_found_detects_403():
+    assert _is_model_not_found(Exception("Project does not have access to model x"))
+    assert _is_model_not_found(Exception("litellm.NotFoundError: nope"))
+    assert not _is_model_not_found(Exception("timeout"))
+
+
+def test_is_embedding_unavailable_detects_403():
+    assert _is_embedding_unavailable(Exception("model_not_found"))
+    assert not _is_embedding_unavailable(Exception("connection reset"))
+
+
+def test_default_embedding_model_env(monkeypatch):
+    monkeypatch.delenv("AIUI_EMBEDDING_MODEL", raising=False)
+    monkeypatch.delenv("OPENAI_EMBEDDING_MODEL", raising=False)
+    assert _default_embedding_model() == "text-embedding-3-small"
+    monkeypatch.setenv("AIUI_EMBEDDING_MODEL", "text-embedding-ada-002")
+    assert _default_embedding_model() == "text-embedding-ada-002"
+
+
+def test_fallback_models_env(monkeypatch):
+    monkeypatch.setenv("AIUI_EMBEDDING_FALLBACK_MODELS", "a, b ,,c")
+    assert _embedding_fallback_models() == ["a", "b", "c"]
+
+
+def test_simple_manager_reports_local_mode():
+    mgr = SimpleKnowledgeManager()
+    mgr.store("hello world")
+    mgr.search("hello")
+    assert mgr.last_search_meta()["mode"] == "local"
+    assert mgr.health()["embedding"]["available"] is False
+
+
+def test_sdk_search_reports_fallback_on_embedding_error():
+    mgr = SDKKnowledgeManager()
+    err = Exception("Project does not have access to model text-embedding-3-small")
+    mgr._sdk_knowledge = _FakeSDK(search_error=err)
+    mgr._sdk_probed = True
+
+    mgr.store("PraisonAIUI is a YAML-driven UI generator.")
+    mgr.search("documentation generator")
+
+    meta = mgr.last_search_meta()
+    assert meta["mode"] == "fallback"
+    assert "text-embedding" in meta["error"].lower()
+
+    health = mgr.health()
+    assert health["embedding"]["available"] is False
+    assert health["status"] == "degraded"
+    assert health["warnings"]
+
+
+def test_sdk_search_vector_mode_when_healthy():
+    class _OkSDK:
+        def search(self, query, limit=1, user_id=None):
+            if query == "__probe__":
+                return []
+            return [{"id": "1", "memory": "hit", "score": 0.9}]
+
+    mgr = SDKKnowledgeManager()
+    mgr._sdk_knowledge = _OkSDK()
+    mgr._sdk_probed = True
+    results = mgr.search("anything")
+    assert results
+    assert mgr.last_search_meta()["mode"] == "vector"
+
+
+def test_probe_flags_embedding_unavailable():
+    mgr = SDKKnowledgeManager()
+
+    class _Kn:
+        def search(self, query, limit=1):
+            raise Exception("model_not_found: no access")
+
+    with patch("praisonaiagents.knowledge.Knowledge", return_value=_Kn(), create=True):
+        mgr._get_sdk_knowledge()
+    assert mgr._embedding_unavailable is True
+
+
+@pytest.fixture
+def client():
+    saved = dict(_features)
+    saved_mgr = knowledge_mod._knowledge_manager
+    _features.clear()
+    register_feature(KnowledgeFeature())
+    register_feature(OpenAIAPIFeature())
+    yield TestClient(create_app())
+    _features.clear()
+    _features.update(saved)
+    knowledge_mod._knowledge_manager = saved_mgr
+
+
+def test_knowledge_search_reports_fallback(client):
+    mgr = SDKKnowledgeManager()
+    err = Exception("Project does not have access to model text-embedding-3-small")
+    mgr._sdk_knowledge = _FakeSDK(search_error=err)
+    mgr._sdk_probed = True
+    mgr.store("PraisonAIUI is a YAML-driven documentation site generator.")
+    knowledge_mod._knowledge_manager = mgr
+
+    r = client.post("/api/knowledge/search", json={"query": "documentation generator YAML"})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["search_mode"] == "fallback"
+    assert body["warning"]
+    assert "embedding" in body["warning"].lower()
+
+
+def test_knowledge_status_reports_embedding(client):
+    mgr = SDKKnowledgeManager()
+    mgr._sdk_knowledge = _FakeSDK()
+    mgr._sdk_probed = True
+    mgr._embedding_unavailable = True
+    mgr._embedding_error = "no access"
+    knowledge_mod._knowledge_manager = mgr
+
+    r = client.get("/api/knowledge/status")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["embedding"]["available"] is False
+    assert body["warnings"]
+
+
+def test_embeddings_endpoint_fallback_models(client):
+    call_models: list[str] = []
+
+    async def _aembed(input, model):
+        call_models.append(model)
+        if model == "text-embedding-3-small":
+            raise Exception("Project does not have access to model text-embedding-3-small")
+        return {"embedding": [0.1, 0.2], "usage": {"prompt_tokens": 1, "total_tokens": 1}}
+
+    fake_caps = type("Caps", (), {"aembed": staticmethod(AsyncMock(side_effect=_aembed))})
+    with patch(
+        "praisonaiui.features.openai_api._get_capabilities", return_value=fake_caps
+    ):
+        r = client.post(
+            "/v1/embeddings",
+            json={"input": "test", "model": "text-embedding-3-small"},
+        )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["model"] == "text-embedding-ada-002"
+    assert call_models[0] == "text-embedding-3-small"
+
+
+def test_embeddings_endpoint_all_fail_returns_503(client):
+    async def _aembed(input, model):
+        raise Exception("model_not_found: no access to any embedding model")
+
+    fake_caps = type("Caps", (), {"aembed": staticmethod(AsyncMock(side_effect=_aembed))})
+    with patch(
+        "praisonaiui.features.openai_api._get_capabilities", return_value=fake_caps
+    ):
+        r = client.post("/v1/embeddings", json={"input": "test"})
+    assert r.status_code == 503
+    assert r.json()["error"]["type"] == "model_not_found"

tests/unit/test_session_search_dashboard.py

@@ -22,7 +22,7 @@ CHAT_JS = FRONTEND / "views" / "chat.js"
 
 class TestDashboardSessionSearch:
     def _src(self) -> str:
-        return DASHBOARD_JS.read_text()
+        return DASHBOARD_JS.read_text(encoding="utf-8")
 
     def test_ctrl_k_keydown_handler_present(self):
         # The Ctrl+K binding is owned by the command palette, which exposes a
@@ -107,7 +107,7 @@ class TestDashboardSessionSearch:
 
 class TestChatViewSessionSelectListener:
     def _src(self) -> str:
-        return CHAT_JS.read_text()
+        return CHAT_JS.read_text(encoding="utf-8")
 
     def test_chat_listens_for_session_select(self):
         src = self._src()

tests/unit/test_skills_studio.py

@@ -48,41 +48,41 @@ class TestI18nKeys:
 
 class TestHelpersJs:
     def test_helpers_export_functions(self):
-        src = HELPERS_JS.read_text()
+        src = HELPERS_JS.read_text(encoding="utf-8")
         assert "export function diffLines(" in src
         assert "export function renderMarkdownPreview(" in src
         assert "export function isSkillApproval(" in src
 
     def test_diff_normalises_crlf(self):
-        src = HELPERS_JS.read_text()
+        src = HELPERS_JS.read_text(encoding="utf-8")
         assert "replace(/\\r\\n?/g, '\\n')" in src
 
     def test_preview_escapes_first(self):
-        src = HELPERS_JS.read_text()
+        src = HELPERS_JS.read_text(encoding="utf-8")
         assert "esc(s)" in src
 
 
 class TestSkillsJs:
     def test_exports_render_and_cleanup(self):
-        src = SKILLS_JS.read_text()
+        src = SKILLS_JS.read_text(encoding="utf-8")
         assert "export async function render" in src
         assert "export function cleanup" in src
 
     def test_tab_shell_present(self):
-        src = SKILLS_JS.read_text()
+        src = SKILLS_JS.read_text(encoding="utf-8")
         assert "class=\"ss-tab\"" in src
         assert "tabBtn('pending'" in src
         assert "tabBtn('installed'" in src
         assert "tabBtn('history'" in src
 
     def test_url_state_and_deep_link(self):
-        src = SKILLS_JS.read_text()
+        src = SKILLS_JS.read_text(encoding="utf-8")
         assert "approval_id" in src
         assert "currentApprovalId(" in src
         assert "openDiff(" in src
 
     def test_uses_existing_apis(self):
-        src = SKILLS_JS.read_text()
+        src = SKILLS_JS.read_text(encoding="utf-8")
         assert "/api/approvals/pending" in src
         assert "/api/approvals/history" in src
         assert "/api/skills" in src
@@ -90,30 +90,30 @@ class TestSkillsJs:
         assert "/api/approvals/${id}/${verb}" in src
 
     def test_imports_skill_helpers(self):
-        src = SKILLS_JS.read_text()
+        src = SKILLS_JS.read_text(encoding="utf-8")
         assert "isSkillApproval" in src
         assert "diffLines" in src
         assert "renderMarkdownPreview" in src
 
     def test_no_browser_eval(self):
-        src = SKILLS_JS.read_text()
+        src = SKILLS_JS.read_text(encoding="utf-8")
         assert "eval(" not in src
 
     def test_proposed_content_escaped_in_diff(self):
-        src = SKILLS_JS.read_text()
+        src = SKILLS_JS.read_text(encoding="utf-8")
         # Diff rows must escape line text; no raw interpolation of r.text.
         assert "esc(r.text)" in src
         assert "${r.text}" not in src
 
     def test_empty_state_and_no_existing_labels(self):
-        src = SKILLS_JS.read_text()
+        src = SKILLS_JS.read_text(encoding="utf-8")
         assert "No pending skill writes" in src
         assert "No existing skill" in src
 
 
 class TestApprovalsCrossLink:
     def test_skill_row_links_to_studio(self):
-        src = APPROVALS_JS.read_text()
+        src = APPROVALS_JS.read_text(encoding="utf-8")
         assert "isSkillApproval" in src
         assert "/skills?tab=pending&approval_id=" in src
 

tests/unit/test_token_auth_optin.py

@@ -0,0 +1,67 @@
+"""Regression tests for opt-in token auth activation (#238, #251).
+
+The mere presence of GATEWAY_AUTH_TOKEN / AIUI_URL_TOKEN in the environment
+(e.g. leaking into a developer shell after a gateway run) must not force 401
+on standalone create_app() + TestClient. Token auth activates only when
+explicitly opted in via AIUI_REQUIRE_TOKEN, YAML auth.requireToken, or
+require_auth=True.
+"""
+
+import pytest
+from starlette.testclient import TestClient
+
+from praisonaiui import server
+from praisonaiui.server import create_app
+
+
+@pytest.fixture(autouse=True)
+def _reset_state():
+    server.reset_state()
+    yield
+    server.reset_state()
+
+
+def test_env_token_alone_does_not_force_401(monkeypatch):
+    monkeypatch.setenv("GATEWAY_AUTH_TOKEN", "test-secret-token")
+    monkeypatch.delenv("AIUI_REQUIRE_TOKEN", raising=False)
+    client = TestClient(create_app())
+    assert client.get("/api/agents").status_code == 200
+
+
+def test_aiui_url_token_alone_does_not_force_401(monkeypatch):
+    monkeypatch.setenv("AIUI_URL_TOKEN", "test-secret-token")
+    monkeypatch.delenv("AIUI_REQUIRE_TOKEN", raising=False)
+    client = TestClient(create_app())
+    assert client.get("/api/agents").status_code == 200
+
+
+def test_require_token_env_opt_in_enforces_401(monkeypatch):
+    monkeypatch.setenv("GATEWAY_AUTH_TOKEN", "test-secret-token")
+    monkeypatch.setenv("AIUI_REQUIRE_TOKEN", "1")
+    client = TestClient(create_app())
+    assert client.get("/api/agents").status_code == 401
+
+
+def test_require_token_env_opt_in_allows_bearer(monkeypatch):
+    monkeypatch.setenv("GATEWAY_AUTH_TOKEN", "test-secret-token")
+    monkeypatch.setenv("AIUI_REQUIRE_TOKEN", "1")
+    client = TestClient(create_app())
+    resp = client.get(
+        "/api/agents",
+        headers={"Authorization": "Bearer test-secret-token"},
+    )
+    assert resp.status_code == 200
+
+
+def test_config_require_token_opt_in_enforces_401(monkeypatch):
+    monkeypatch.setenv("AIUI_URL_TOKEN", "test-secret-token")
+    monkeypatch.delenv("AIUI_REQUIRE_TOKEN", raising=False)
+    client = TestClient(create_app(config={"auth": {"requireToken": True}}))
+    assert client.get("/api/agents").status_code == 401
+
+
+def test_require_auth_opt_in_enforces_401(monkeypatch):
+    monkeypatch.setenv("AIUI_URL_TOKEN", "test-secret-token")
+    monkeypatch.delenv("AIUI_REQUIRE_TOKEN", raising=False)
+    client = TestClient(create_app(require_auth=True))
+    assert client.get("/api/agents").status_code == 401

uv.lock

@@ -207,7 +207,7 @@ wheels = [
 
 [[package]]
 name = "aiui"
-version = "0.3.134"
+version = "0.3.136"
 source = { editable = "." }
 dependencies = [
     { name = "pydantic" },