diff --git a/src/api/routers/community.py b/src/api/routers/community.py
index 6b9f3ff..1022cc3 100644
--- a/src/api/routers/community.py
+++ b/src/api/routers/community.py
@@ -764,10 +764,11 @@ def create_community_assistant(
if config.get("callbacks"):
langfuse_config = config
langfuse_trace_id = trace_id
- except Exception:
+ except (AttributeError, ValueError, RuntimeError, OSError, ImportError) as e:
logger.warning(
- "LangFuse tracing setup failed for %s, continuing without it",
+ "LangFuse tracing setup failed for %s: %s, continuing without it",
community_id,
+ e,
exc_info=True,
)
@@ -1183,8 +1184,13 @@ async def get_community_config() -> CommunityConfigResponse:
if info.community_config:
try:
health_status = compute_community_health(info.community_config)["status"]
- except Exception:
- logger.exception("Failed to compute health for community %s", info.id)
+ except (AttributeError, KeyError, TypeError) as e:
+ logger.error(
+ "Failed to compute health for community %s: %s",
+ info.id,
+ e,
+ exc_info=True,
+ )
return CommunityConfigResponse(
id=info.id,
@@ -1357,8 +1363,13 @@ async def community_metrics_public() -> dict[str, Any]:
"documents": health["documents"],
"warnings": public_warnings,
}
- except Exception:
- logger.exception("Failed to compute health for community %s", community_id)
+ except (AttributeError, KeyError, TypeError) as e:
+ logger.error(
+ "Failed to compute health for community %s: %s",
+ community_id,
+ e,
+ exc_info=True,
+ )
result["config_health"] = fallback_health
else:
result["config_health"] = fallback_health
diff --git a/src/assistants/__init__.py b/src/assistants/__init__.py
index 5d415cb..172dcc1 100644
--- a/src/assistants/__init__.py
+++ b/src/assistants/__init__.py
@@ -88,6 +88,10 @@ def discover_assistants() -> list[str]:
"Community '%s': env var '%s' not set, will use platform key",
config.id,
config.openrouter_api_key_env_var,
+ extra={
+ "community_id": config.id,
+ "env_var": config.openrouter_api_key_env_var,
+ },
)
except KeyboardInterrupt:
# Never catch user interruption
diff --git a/src/cli/client.py b/src/cli/client.py
index 33426bb..4701be1 100644
--- a/src/cli/client.py
+++ b/src/cli/client.py
@@ -162,7 +162,7 @@ def _stream_request(
event_type = data.get("event", "unknown")
yield (event_type, data)
except json.JSONDecodeError:
- logger.debug("Malformed SSE data, skipping: %s", line[:200])
+ logger.warning("Malformed SSE data, skipping: %s", line[:200])
continue
def ask_stream(
diff --git a/src/cli/config.py b/src/cli/config.py
index f4f17ac..fe125a5 100644
--- a/src/cli/config.py
+++ b/src/cli/config.py
@@ -7,6 +7,7 @@
import contextlib
import json
+import logging
import os
import uuid
from pathlib import Path
@@ -16,6 +17,8 @@
from platformdirs import user_config_dir, user_data_dir
from pydantic import BaseModel, Field, ValidationError
+logger = logging.getLogger(__name__)
+
# Paths
CONFIG_DIR = Path(user_config_dir("osa", appauthor=False, ensure_exists=True))
CONFIG_FILE = CONFIG_DIR / "config.yaml"
@@ -79,7 +82,8 @@ def load_config() -> CLIConfig:
try:
data = yaml.safe_load(CONFIG_FILE.read_text()) or {}
return CLIConfig(**data)
- except (yaml.YAMLError, OSError, TypeError, ValidationError):
+ except (yaml.YAMLError, OSError, TypeError, ValidationError) as e:
+ logger.warning("Failed to load config from %s, using defaults: %s", CONFIG_FILE, e)
return CLIConfig()
@@ -98,7 +102,12 @@ def load_credentials() -> CredentialsConfig:
try:
data = yaml.safe_load(CREDENTIALS_FILE.read_text()) or {}
return CredentialsConfig(**data)
- except (yaml.YAMLError, OSError, TypeError, ValidationError):
+ except (yaml.YAMLError, OSError, TypeError, ValidationError) as e:
+ logger.warning(
+ "Failed to load credentials from %s, no API keys available: %s",
+ CREDENTIALS_FILE,
+ e,
+ )
return CredentialsConfig()
@@ -115,11 +124,23 @@ def save_credentials(creds: CredentialsConfig) -> None:
os.write(fd, content.encode())
finally:
os.close(fd)
- except OSError:
+ except OSError as e:
# Fallback for platforms that don't support os.open mode (e.g., Windows)
+ logger.warning(
+ "Secure file write failed (%s), falling back to standard write for %s",
+ e,
+ CREDENTIALS_FILE,
+ )
CREDENTIALS_FILE.write_text(content)
- with contextlib.suppress(OSError):
+ try:
os.chmod(CREDENTIALS_FILE, 0o600)
+ except OSError as chmod_err:
+ logger.warning(
+ "Could not restrict permissions on %s: %s. "
+ "Credentials file may be readable by other users.",
+ CREDENTIALS_FILE,
+ chmod_err,
+ )
def get_effective_config(
@@ -154,7 +175,8 @@ def _migrate_legacy_config() -> CLIConfig:
try:
with LEGACY_CONFIG_FILE.open() as f:
data = json.load(f)
- except (json.JSONDecodeError, OSError):
+ except (json.JSONDecodeError, OSError) as e:
+ logger.warning("Failed to migrate legacy config from %s: %s", LEGACY_CONFIG_FILE, e)
return CLIConfig()
# Build new config from legacy fields
diff --git a/tests/test_api/test_logo.py b/tests/test_api/test_logo.py
new file mode 100644
index 0000000..14e598b
--- /dev/null
+++ b/tests/test_api/test_logo.py
@@ -0,0 +1,220 @@
+"""Tests for community logo serving.
+
+Tests cover:
+- find_logo_file convention-based detection
+- convention_logo_url helper
+- GET /{community_id}/logo endpoint (404, SVG CSP header)
+- Logo URL in /communities and /{community_id} config responses
+"""
+
+from pathlib import Path
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.routers.community import (
+ _LOGO_MEDIA_TYPES,
+ convention_logo_url,
+ create_community_router,
+ find_logo_file,
+)
+from src.assistants import discover_assistants, registry
+from src.core.config.community import WidgetConfig
+
+# Discover assistants to populate registry
+discover_assistants()
+
+
+class TestFindLogoFile:
+ """Tests for find_logo_file function."""
+
+ def test_returns_none_for_nonexistent_community(self) -> None:
+ """Should return None for a community directory that doesn't exist."""
+ result = find_logo_file("nonexistent-community-xyz")
+ assert result is None
+
+ def test_returns_none_when_no_logo_exists(self) -> None:
+ """Should return None for real communities without logo files."""
+ # Check all registered communities; unless someone has added a logo
+ # file, they should all return None
+ for info in registry.list_available():
+ result = find_logo_file(info.id)
+ if result is not None:
+ # A logo file exists; that's fine, just verify it's a valid path
+ assert result.is_file()
+ assert result.suffix in _LOGO_MEDIA_TYPES
+
+ def test_finds_logo_in_temp_dir(self, tmp_path: Path) -> None:
+ """Should find a logo file when one exists in the community folder."""
+ from src.api.routers import community as community_module
+
+ original_dir = community_module._ASSISTANTS_DIR
+ try:
+ # Create a fake community directory with a logo
+ community_dir = tmp_path / "test-community"
+ community_dir.mkdir()
+ logo_file = community_dir / "logo.png"
+ logo_file.write_bytes(b"\x89PNG\r\n\x1a\n") # PNG magic bytes
+
+ community_module._ASSISTANTS_DIR = tmp_path
+ result = find_logo_file("test-community")
+ assert result is not None
+ assert result.name == "logo.png"
+ finally:
+ community_module._ASSISTANTS_DIR = original_dir
+
+ def test_prefers_svg_over_png(self, tmp_path: Path) -> None:
+ """Should prefer SVG over PNG when both exist."""
+ from src.api.routers import community as community_module
+
+ original_dir = community_module._ASSISTANTS_DIR
+ try:
+ community_dir = tmp_path / "test-community"
+ community_dir.mkdir()
+ (community_dir / "logo.svg").write_text("")
+ (community_dir / "logo.png").write_bytes(b"\x89PNG\r\n\x1a\n")
+
+ community_module._ASSISTANTS_DIR = tmp_path
+ result = find_logo_file("test-community")
+ assert result is not None
+ assert result.suffix == ".svg"
+ finally:
+ community_module._ASSISTANTS_DIR = original_dir
+
+
+class TestConventionLogoUrl:
+ """Tests for convention_logo_url helper."""
+
+ def test_returns_none_when_explicit_logo_url_set(self) -> None:
+ """Should return None when widget already has an explicit logo_url."""
+ widget = WidgetConfig(logo_url="https://example.com/logo.png")
+ result = convention_logo_url("hed", widget)
+ assert result is None
+
+ def test_returns_none_when_no_logo_file(self) -> None:
+ """Should return None for communities without logo files."""
+ widget = WidgetConfig()
+ # Use a non-existent community to ensure no file is found
+ result = convention_logo_url("nonexistent-community-xyz", widget)
+ assert result is None
+
+ def test_returns_url_when_logo_file_exists(self, tmp_path: Path) -> None:
+ """Should return convention URL when logo file exists."""
+ from src.api.routers import community as community_module
+
+ original_dir = community_module._ASSISTANTS_DIR
+ try:
+ community_dir = tmp_path / "test-community"
+ community_dir.mkdir()
+ (community_dir / "logo.png").write_bytes(b"\x89PNG\r\n\x1a\n")
+
+ community_module._ASSISTANTS_DIR = tmp_path
+ widget = WidgetConfig()
+ result = convention_logo_url("test-community", widget)
+ assert result == "/test-community/logo"
+ finally:
+ community_module._ASSISTANTS_DIR = original_dir
+
+
+class TestLogoEndpoint:
+ """Tests for GET /{community_id}/logo endpoint."""
+
+ def test_returns_404_when_no_logo(self) -> None:
+ """Should return 404 for communities without logo files."""
+ # Use a real community that doesn't have a logo file
+ for info in registry.list_available():
+ if find_logo_file(info.id) is None:
+ app = FastAPI()
+ app.include_router(create_community_router(info.id))
+ client = TestClient(app)
+ response = client.get(f"/{info.id}/logo")
+ assert response.status_code == 404
+ return
+ pytest.skip("All communities have logo files")
+
+ def test_serves_logo_with_correct_content_type(self, tmp_path: Path) -> None:
+ """Should serve logo with correct media type and cache headers."""
+ from src.api.routers import community as community_module
+
+ original_dir = community_module._ASSISTANTS_DIR
+ try:
+ # Create a fake community with a logo file
+ community_dir = tmp_path / "hed"
+ community_dir.mkdir()
+ png_content = b"\x89PNG\r\n\x1a\n" + b"\x00" * 100
+ (community_dir / "logo.png").write_bytes(png_content)
+
+ community_module._ASSISTANTS_DIR = tmp_path
+
+ app = FastAPI()
+ app.include_router(create_community_router("hed"))
+ client = TestClient(app)
+ response = client.get("/hed/logo")
+ assert response.status_code == 200
+ assert response.headers["content-type"] == "image/png"
+ assert "max-age=86400" in response.headers["cache-control"]
+ finally:
+ community_module._ASSISTANTS_DIR = original_dir
+
+ def test_svg_gets_csp_header(self, tmp_path: Path) -> None:
+ """SVG logos should include Content-Security-Policy to prevent XSS."""
+ from src.api.routers import community as community_module
+
+ original_dir = community_module._ASSISTANTS_DIR
+ try:
+ community_dir = tmp_path / "hed"
+ community_dir.mkdir()
+ (community_dir / "logo.svg").write_text(
+ ''
+ )
+
+ community_module._ASSISTANTS_DIR = tmp_path
+
+ app = FastAPI()
+ app.include_router(create_community_router("hed"))
+ client = TestClient(app)
+ response = client.get("/hed/logo")
+ assert response.status_code == 200
+ assert "image/svg+xml" in response.headers["content-type"]
+ assert "default-src 'none'" in response.headers["content-security-policy"]
+ finally:
+ community_module._ASSISTANTS_DIR = original_dir
+
+ def test_png_does_not_get_csp_header(self, tmp_path: Path) -> None:
+ """Non-SVG logos should not get CSP header."""
+ from src.api.routers import community as community_module
+
+ original_dir = community_module._ASSISTANTS_DIR
+ try:
+ community_dir = tmp_path / "hed"
+ community_dir.mkdir()
+ (community_dir / "logo.png").write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
+
+ community_module._ASSISTANTS_DIR = tmp_path
+
+ app = FastAPI()
+ app.include_router(create_community_router("hed"))
+ client = TestClient(app)
+ response = client.get("/hed/logo")
+ assert response.status_code == 200
+ assert "content-security-policy" not in response.headers
+ finally:
+ community_module._ASSISTANTS_DIR = original_dir
+
+
+class TestLogoInCommunityConfig:
+ """Tests that logo_url appears in community config responses."""
+
+ def test_communities_endpoint_includes_logo_url(self) -> None:
+ """GET /communities should include logo_url in widget config."""
+ from src.api.routers.communities import router
+
+ app = FastAPI()
+ app.include_router(router)
+ client = TestClient(app)
+ response = client.get("/communities")
+ assert response.status_code == 200
+ data = response.json()
+ for community in data:
+ assert "logo_url" in community["widget"]
diff --git a/tests/test_core/test_config/test_community.py b/tests/test_core/test_config/test_community.py
index 1f3aa66..1e4ae5c 100644
--- a/tests/test_core/test_config/test_community.py
+++ b/tests/test_core/test_config/test_community.py
@@ -531,6 +531,73 @@ def test_resolve_with_values(self) -> None:
assert result["placeholder"] == "Custom placeholder"
+class TestWidgetConfigLogoUrl:
+ """Tests for WidgetConfig.logo_url field and validation."""
+
+ def test_logo_url_accepts_https(self) -> None:
+ """Should accept HTTPS URLs."""
+ widget = WidgetConfig(logo_url="https://example.com/logo.png")
+ assert widget.logo_url == "https://example.com/logo.png"
+
+ def test_logo_url_accepts_http(self) -> None:
+ """Should accept HTTP URLs."""
+ widget = WidgetConfig(logo_url="http://example.com/logo.png")
+ assert widget.logo_url == "http://example.com/logo.png"
+
+ def test_logo_url_accepts_relative_path(self) -> None:
+ """Should accept paths starting with /."""
+ widget = WidgetConfig(logo_url="/hed/logo")
+ assert widget.logo_url == "/hed/logo"
+
+ def test_logo_url_rejects_javascript(self) -> None:
+ """Should reject javascript: URLs."""
+ with pytest.raises(ValidationError, match="logo_url must use"):
+ WidgetConfig(logo_url="javascript:alert(1)")
+
+ def test_logo_url_rejects_data_uri(self) -> None:
+ """Should reject data: URIs."""
+ with pytest.raises(ValidationError, match="logo_url must use"):
+ WidgetConfig(logo_url="data:text/html,")
+
+ def test_logo_url_rejects_ftp(self) -> None:
+ """Should reject ftp: URLs."""
+ with pytest.raises(ValidationError, match="logo_url must use"):
+ WidgetConfig(logo_url="ftp://example.com/logo.png")
+
+ def test_logo_url_none_by_default(self) -> None:
+ """Should default to None."""
+ widget = WidgetConfig()
+ assert widget.logo_url is None
+
+ def test_logo_url_empty_string_normalized(self) -> None:
+ """Empty or whitespace-only string should become None."""
+ widget = WidgetConfig(logo_url=" ")
+ assert widget.logo_url is None
+
+ def test_logo_url_strips_whitespace(self) -> None:
+ """Should strip whitespace from logo_url."""
+ widget = WidgetConfig(logo_url=" https://example.com/logo.png ")
+ assert widget.logo_url == "https://example.com/logo.png"
+
+ def test_resolve_with_logo_url_fallback(self) -> None:
+ """resolve() should use fallback logo_url when self.logo_url is None."""
+ widget = WidgetConfig()
+ result = widget.resolve("Test", logo_url="/test/logo")
+ assert result["logo_url"] == "/test/logo"
+
+ def test_resolve_explicit_logo_url_takes_precedence(self) -> None:
+ """resolve() should prefer explicit logo_url over fallback."""
+ widget = WidgetConfig(logo_url="https://example.com/explicit.png")
+ result = widget.resolve("Test", logo_url="/test/logo")
+ assert result["logo_url"] == "https://example.com/explicit.png"
+
+ def test_resolve_no_logo_url_returns_none(self) -> None:
+ """resolve() should return None when no logo_url is set anywhere."""
+ widget = WidgetConfig()
+ result = widget.resolve("Test")
+ assert result["logo_url"] is None
+
+
class TestCommunityConfigWidget:
"""Tests for CommunityConfig.widget field."""