Upload 52 files

.dockerignore

@@ -0,0 +1,31 @@
+# Exclude local dev environment
+.venv
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+
+# Exclude git history
+.git
+.gitignore
+
+# Exclude test cache and coverage reports
+.pytest_cache
+.coverage
+htmlcov
+tests/
+
+# Exclude logs and data (unless specific data is needed)
+logs/
+data/
+
+# Exclude local env files (secrets should be passed as env vars)
+.env
+.env.example
+
+# Exclude IDE settings
+.vscode
+.idea
+
+# Exclude poetry cache (if local)
+.cache

.env.example

@@ -0,0 +1,13 @@
+# Google Gemini API Key
+GEMINI_API_KEY=
+
+# Application Settings
+UPDATE_INTERVAL_MINUTES=
+CACHE_EXPIRY_MINUTES=
+LOG_LEVEL=
+AUTO_UPDATE_ENABLED=
+REDDIT_CLIENT_ID=
+REDDIT_CLIENT_SECRET=
+REDDIT_USER_AGENT=
+NEWSAPI_KEY=
+WESTERNFRONT_API_KEY=

Dockerfile

@@ -1,27 +1,42 @@
-FROM python:3.11-slim
-
-WORKDIR /app
-
-# Install Poetry
-RUN pip install --no-cache-dir poetry==1.8.0
-
-# Copy dependency files
-COPY pyproject.toml poetry.lock ./
-
-# Install dependencies
-RUN poetry config virtualenvs.create false \
-    && poetry install --only main --no-interaction --no-ansi
-
-# Copy application code
-COPY src/ ./src/
-
-# Create data directory
-RUN mkdir -p /app/data /app/logs
-
-ENV PYTHONDONTWRITEBYTECODE=1
-ENV PYTHONUNBUFFERED=1
-ENV PYTHONPATH=/app/src
-
-EXPOSE 7860
-
+FROM python:3.11-slim AS builder
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    curl \
+    && rm -rf /var/lib/apt/lists/* \
+    && pip install --no-cache-dir poetry==1.8.0
+
+ENV POETRY_NO_INTERACTION=1 \
+    POETRY_VIRTUALENVS_IN_PROJECT=1 \
+    POETRY_VIRTUALENVS_CREATE=1 \
+    POETRY_CACHE_DIR=/tmp/poetry_cache
+
+COPY pyproject.toml poetry.lock ./
+
+RUN poetry install --only main --no-root && rm -rf $POETRY_CACHE_DIR
+
+FROM python:3.11-slim AS runtime
+
+WORKDIR /app
+
+RUN groupadd -g 1000 appuser && \
+    useradd -u 1000 -g appuser -s /bin/bash -m appuser
+
+RUN mkdir -p /app/data /app/logs && \
+    chown -R appuser:appuser /app
+
+COPY --from=builder /app/.venv /app/.venv
+COPY --chown=appuser:appuser src/ ./src/
+
+ENV PATH="/app/.venv/bin:$PATH" \
+    PYTHONPATH="/app/src" \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PORT=7860
+
+USER appuser
+
+EXPOSE 7860
+
 CMD ["uvicorn", "westernfront.main:app", "--host", "0.0.0.0", "--port", "7860"]

src/westernfront/analytics/aggregator.py

@@ -1,8 +1,6 @@
 """Analytics aggregation for graph data."""
 
 from collections import Counter
-from datetime import datetime, timedelta, timezone
-from typing import Optional
 
 from westernfront.core.enums import TensionTrend
 from westernfront.repositories.analysis import AnalysisRepository
@@ -12,24 +10,10 @@ class AnalyticsAggregator:
     """Aggregates analysis data for visualization."""
 
     def __init__(self, repository: AnalysisRepository) -> None:
-        """
-        Initialize the aggregator.
-
-        Args:
-            repository: Analysis repository for data access.
-        """
         self._repository = repository
 
     async def get_tension_history(self, days: int = 30) -> dict:
-        """
-        Get tension score history for graphing.
-
-        Args:
-            days: Number of days to include.
-
-        Returns:
-            Dictionary with data points and summary statistics.
-        """
+        """Get tension score history for graphing."""
         history = await self._repository.get_tension_history(days)
 
         if not history:
@@ -66,15 +50,7 @@ class AnalyticsAggregator:
         }
 
     async def get_source_breakdown(self, days: int = 7) -> dict:
-        """
-        Get breakdown of sources used in recent analyses.
-
-        Args:
-            days: Number of days to include.
-
-        Returns:
-            Dictionary with source counts.
-        """
+        """Get breakdown of sources used in recent analyses."""
         snapshots = await self._repository.get_history(days=days)
 
         return {
@@ -90,16 +66,7 @@ class AnalyticsAggregator:
         }
 
     async def get_entity_frequency(self, days: int = 30, top_n: int = 10) -> dict:
-        """
-        Get most frequently mentioned entities.
-
-        Args:
-            days: Number of days to include.
-            top_n: Number of top entities to return.
-
-        Returns:
-            Dictionary with entity frequency data.
-        """
+        """Get most frequently mentioned entities."""
         snapshots = await self._repository.get_history(days=days)
 
         all_entities: list[str] = []
@@ -118,15 +85,7 @@ class AnalyticsAggregator:
         }
 
     async def get_analysis_type_distribution(self, days: int = 30) -> dict:
-        """
-        Get distribution of analysis types.
-
-        Args:
-            days: Number of days to include.
-
-        Returns:
-            Dictionary with type distribution.
-        """
+        """Get distribution of analysis types."""
         snapshots = await self._repository.get_history(days=days)
 
         counter = Counter(s.analysis_type.value for s in snapshots)

src/westernfront/api/auth.py

@@ -1,13 +1,11 @@
 """API key authentication middleware."""
 
+import secrets
+
 from fastapi import HTTPException, Request, status
-from fastapi.security import APIKeyHeader
 
 from westernfront.config import get_settings
 
-
-API_KEY_HEADER = APIKeyHeader(name="X-API-Key", auto_error=False)
-
 PUBLIC_PATHS = frozenset([
     "/",
     "/health",
@@ -18,15 +16,7 @@ PUBLIC_PATHS = frozenset([
 
 
 async def verify_api_key(request: Request) -> None:
-    """
-    Verify the API key from request headers.
-
-    Args:
-        request: The incoming request.
-
-    Raises:
-        HTTPException: If API key is missing or invalid.
-    """
+    """Verify the API key from request headers using timing-safe comparison."""
     if request.url.path in PUBLIC_PATHS:
         return
 
@@ -39,7 +29,7 @@ async def verify_api_key(request: Request) -> None:
             detail="Missing API key. Include X-API-Key header.",
         )
 
-    if api_key != settings.api_key:
+    if not secrets.compare_digest(api_key.encode(), settings.api_key.encode()):
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Invalid API key.",

src/westernfront/api/middleware/__init__.py

@@ -0,0 +1,5 @@
+"""Middleware package exports."""
+
+from westernfront.api.middleware.rate_limit import RateLimitMiddleware
+
+__all__ = ["RateLimitMiddleware"]

src/westernfront/api/middleware/rate_limit.py

@@ -0,0 +1,89 @@
+"""Rate limiting middleware using token bucket algorithm with automatic cleanup."""
+
+import time
+
+from cachetools import TTLCache
+from fastapi import Request, Response
+from fastapi.responses import JSONResponse
+from starlette.middleware.base import BaseHTTPMiddleware
+
+
+class TokenBucket:
+    """Token bucket for rate limiting."""
+
+    __slots__ = ("capacity", "refill_rate", "tokens", "last_refill")
+
+    def __init__(self, capacity: int, refill_rate: float) -> None:
+        self.capacity = capacity
+        self.refill_rate = refill_rate
+        self.tokens = float(capacity)
+        self.last_refill = time.monotonic()
+
+    def consume(self) -> bool:
+        """Attempt to consume a token. Returns True if successful."""
+        now = time.monotonic()
+        elapsed = now - self.last_refill
+        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
+        self.last_refill = now
+
+        if self.tokens >= 1:
+            self.tokens -= 1
+            return True
+        return False
+
+    def time_until_available(self) -> float:
+        """Calculate seconds until next token is available."""
+        if self.tokens >= 1:
+            return 0.0
+        return (1 - self.tokens) / self.refill_rate
+
+
+class RateLimitMiddleware(BaseHTTPMiddleware):
+    """Rate limiting middleware using per-IP token buckets with automatic cleanup."""
+
+    def __init__(
+        self,
+        app,
+        requests_per_minute: int = 60,
+        burst_size: int = 10,
+        bucket_ttl_seconds: int = 300,
+        max_buckets: int = 10000,
+    ) -> None:
+        super().__init__(app)
+        self._requests_per_minute = requests_per_minute
+        self._burst_size = burst_size
+        self._refill_rate = requests_per_minute / 60.0
+        self._buckets: TTLCache[str, TokenBucket] = TTLCache(
+            maxsize=max_buckets,
+            ttl=bucket_ttl_seconds,
+        )
+
+    def _get_bucket(self, client_ip: str) -> TokenBucket:
+        """Get or create a token bucket for the client IP."""
+        bucket = self._buckets.get(client_ip)
+        if bucket is None:
+            bucket = TokenBucket(self._burst_size, self._refill_rate)
+            self._buckets[client_ip] = bucket
+        return bucket
+
+    async def dispatch(self, request: Request, call_next) -> Response:
+        """Process request with rate limiting."""
+        client_ip = self._get_client_ip(request)
+        bucket = self._get_bucket(client_ip)
+
+        if not bucket.consume():
+            retry_after = int(bucket.time_until_available()) + 1
+            return JSONResponse(
+                status_code=429,
+                content={"detail": "Rate limit exceeded. Please slow down."},
+                headers={"Retry-After": str(retry_after)},
+            )
+
+        return await call_next(request)
+
+    def _get_client_ip(self, request: Request) -> str:
+        """Extract client IP from request, handling proxies."""
+        forwarded = request.headers.get("X-Forwarded-For")
+        if forwarded:
+            return forwarded.split(",")[0].strip()
+        return request.client.host if request.client else "unknown"

src/westernfront/api/routes.py

@@ -1,15 +1,15 @@
 """API route definitions."""
 
 from datetime import datetime
-from typing import Optional
 
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi import APIRouter, Query, Request, status
+from fastapi.exceptions import HTTPException
 
 from westernfront import __version__
-from westernfront.analytics import AnalyticsAggregator
 from westernfront.api.schemas import (
     AnalysisHistoryResponse,
     AnalysisSnapshotResponse,
+    AnalysisTypeDistributionResponse,
     ConflictAnalysisResponse,
     EntityFrequencyResponse,
     HealthResponse,
@@ -20,25 +20,17 @@ from westernfront.api.schemas import (
     SourcesResponse,
     SubredditSourceResponse,
     TensionHistoryResponse,
-    AnalysisTypeDistributionResponse,
 )
 from westernfront.core.enums import TensionLevel
-from westernfront.dependencies import (
-    get_analysis_service,
-    get_app_state,
-    get_repository,
-)
-from westernfront.repositories import AnalysisRepository
-from westernfront.services import AnalysisService
-
+from westernfront.dependencies import get_state_from_request
 
 router = APIRouter()
 
 
 @router.get("/", response_model=RootResponse, tags=["General"])
-async def root() -> RootResponse:
+async def root(request: Request) -> RootResponse:
     """Root endpoint with API information."""
-    state = get_app_state()
+    state = get_state_from_request(request)
     return RootResponse(
         name="WesternFront API",
         description="AI-powered conflict tracker for India-Pakistan tensions",
@@ -49,9 +41,9 @@ async def root() -> RootResponse:
 
 @router.get("/health", response_model=HealthResponse, tags=["General"])
 @router.head("/health", response_model=HealthResponse, tags=["General"])
-async def health_check() -> HealthResponse:
+async def health_check(request: Request) -> HealthResponse:
     """Health check endpoint."""
-    state = get_app_state()
+    state = get_state_from_request(request)
     latest = await state.repository.get_latest()
 
     return HealthResponse(
@@ -68,16 +60,11 @@ async def health_check() -> HealthResponse:
     )
 
 
-@router.get(
-    "/analysis",
-    response_model=ConflictAnalysisResponse,
-    tags=["Analysis"],
-)
-async def get_latest_analysis(
-    service: AnalysisService = Depends(get_analysis_service),
-) -> ConflictAnalysisResponse:
+@router.get("/analysis", response_model=ConflictAnalysisResponse, tags=["Analysis"])
+async def get_latest_analysis(request: Request) -> ConflictAnalysisResponse:
     """Get the latest conflict analysis."""
-    analysis = await service.get_latest()
+    state = get_state_from_request(request)
+    analysis = await state.analysis.get_latest()
 
     if not analysis:
         raise HTTPException(
@@ -88,18 +75,15 @@ async def get_latest_analysis(
     return ConflictAnalysisResponse.model_validate(analysis.model_dump())
 
 
-@router.get(
-    "/analysis/history",
-    response_model=AnalysisHistoryResponse,
-    tags=["Analysis"],
-)
+@router.get("/analysis/history", response_model=AnalysisHistoryResponse, tags=["Analysis"])
 async def get_analysis_history(
+    request: Request,
     days: int = Query(default=30, ge=1, le=90),
     limit: int = Query(default=50, ge=1, le=100),
-    repository: AnalysisRepository = Depends(get_repository),
 ) -> AnalysisHistoryResponse:
     """Get historical analysis snapshots."""
-    snapshots = await repository.get_history(days=days, limit=limit)
+    state = get_state_from_request(request)
+    snapshots = await state.repository.get_history(days=days, limit=limit)
 
     return AnalysisHistoryResponse(
         count=len(snapshots),
@@ -110,71 +94,55 @@ async def get_analysis_history(
     )
 
 
-@router.get(
-    "/analytics/tension-history",
-    response_model=TensionHistoryResponse,
-    tags=["Analytics"],
-)
+@router.get("/analytics/tension-history", response_model=TensionHistoryResponse, tags=["Analytics"])
 async def get_tension_history(
+    request: Request,
     days: int = Query(default=30, ge=1, le=90),
-    repository: AnalysisRepository = Depends(get_repository),
 ) -> TensionHistoryResponse:
     """Get tension score history for graphing."""
-    aggregator = AnalyticsAggregator(repository)
-    result = await aggregator.get_tension_history(days)
+    state = get_state_from_request(request)
+    result = await state.analytics.get_tension_history(days)
     return TensionHistoryResponse.model_validate(result)
 
 
-@router.get(
-    "/analytics/source-breakdown",
-    response_model=SourceBreakdownResponse,
-    tags=["Analytics"],
-)
+@router.get("/analytics/source-breakdown", response_model=SourceBreakdownResponse, tags=["Analytics"])
 async def get_source_breakdown(
+    request: Request,
     days: int = Query(default=7, ge=1, le=30),
-    repository: AnalysisRepository = Depends(get_repository),
 ) -> SourceBreakdownResponse:
     """Get breakdown of sources used in analyses."""
-    aggregator = AnalyticsAggregator(repository)
-    result = await aggregator.get_source_breakdown(days)
+    state = get_state_from_request(request)
+    result = await state.analytics.get_source_breakdown(days)
     return SourceBreakdownResponse.model_validate(result)
 
 
-@router.get(
-    "/analytics/entity-frequency",
-    response_model=EntityFrequencyResponse,
-    tags=["Analytics"],
-)
+@router.get("/analytics/entity-frequency", response_model=EntityFrequencyResponse, tags=["Analytics"])
 async def get_entity_frequency(
+    request: Request,
     days: int = Query(default=30, ge=1, le=90),
     top_n: int = Query(default=10, ge=1, le=50),
-    repository: AnalysisRepository = Depends(get_repository),
 ) -> EntityFrequencyResponse:
     """Get most frequently mentioned entities."""
-    aggregator = AnalyticsAggregator(repository)
-    result = await aggregator.get_entity_frequency(days, top_n)
+    state = get_state_from_request(request)
+    result = await state.analytics.get_entity_frequency(days, top_n)
     return EntityFrequencyResponse.model_validate(result)
 
 
-@router.get(
-    "/analytics/type-distribution",
-    response_model=AnalysisTypeDistributionResponse,
-    tags=["Analytics"],
-)
+@router.get("/analytics/type-distribution", response_model=AnalysisTypeDistributionResponse, tags=["Analytics"])
 async def get_type_distribution(
+    request: Request,
     days: int = Query(default=30, ge=1, le=90),
-    repository: AnalysisRepository = Depends(get_repository),
 ) -> AnalysisTypeDistributionResponse:
     """Get distribution of analysis types."""
-    aggregator = AnalyticsAggregator(repository)
-    result = await aggregator.get_analysis_type_distribution(days)
+    state = get_state_from_request(request)
+    result = await state.analytics.get_analysis_type_distribution(days)
     return AnalysisTypeDistributionResponse.model_validate(result)
 
 
 @router.get("/sources", response_model=SourcesResponse, tags=["Configuration"])
-async def get_sources() -> SourcesResponse:
+async def get_sources(request: Request) -> SourcesResponse:
     """Get current data sources configuration."""
-    state = get_app_state()
+    state = get_state_from_request(request)
 
     return SourcesResponse(
         subreddits=[
@@ -190,13 +158,12 @@ async def get_sources() -> SourcesResponse:
 
 
 @router.get("/keywords", response_model=KeywordsResponse, tags=["Configuration"])
-async def get_keywords(
-    service: AnalysisService = Depends(get_analysis_service),
-) -> KeywordsResponse:
+async def get_keywords(request: Request) -> KeywordsResponse:
     """Get current search keywords."""
+    state = get_state_from_request(request)
     return KeywordsResponse(
-        count=len(service.keywords),
-        keywords=service.keywords,
+        count=len(state.analysis.keywords),
+        keywords=state.analysis.keywords,
     )
 
 

src/westernfront/api/schemas.py

@@ -1,7 +1,6 @@
 """API request and response schemas."""
 
 from datetime import datetime
-from typing import Optional
 
 from pydantic import BaseModel, Field
 
@@ -14,7 +13,7 @@ class HealthResponse(BaseModel):
     status: str
     version: str
     timestamp: datetime
-    last_update: Optional[datetime] = None
+    last_update: datetime | None = None
     components: dict[str, bool]
 
 
@@ -93,7 +92,7 @@ class KeyDevelopmentResponse(BaseModel):
     title: str
     description: str
     sources: list[str]
-    timestamp: Optional[datetime] = None
+    timestamp: datetime | None = None
 
 
 class ReliabilityAssessmentResponse(BaseModel):

src/westernfront/config.py

@@ -6,7 +6,6 @@ for type-safe environment variable parsing and validation.
 """
 
 from functools import lru_cache
-from typing import Optional
 
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -34,7 +33,7 @@ class Settings(BaseSettings):
     gemini_api_key: str = Field(alias="GEMINI_API_KEY")
 
     # NewsAPI (optional)
-    newsapi_key: Optional[str] = Field(default=None, alias="NEWSAPI_KEY")
+    newsapi_key: str | None = Field(default=None, alias="NEWSAPI_KEY")
 
     # Application Settings
     update_interval_minutes: int = Field(default=60, alias="UPDATE_INTERVAL_MINUTES")

src/westernfront/core/__init__.py

@@ -1,20 +1,44 @@
 """Core package exports."""
 
 from westernfront.core.enums import AnalysisType, SourceType, TensionLevel, TensionTrend
+from westernfront.core.exceptions import (
+    AnalysisError,
+    AuthenticationError,
+    DataFetchError,
+    RateLimitExceededError,
+    ServiceNotInitializedError,
+    VectorStoreError,
+    WesternFrontError,
+)
 from westernfront.core.models import (
+    AnalysisSnapshot,
     ConflictAnalysis,
     KeyDevelopment,
     NewsItem,
+    RegionalImplications,
+    ReliabilityAssessment,
+    RssFeed,
     SubredditSource,
 )
 
 __all__ = [
+    "AnalysisError",
+    "AnalysisSnapshot",
     "AnalysisType",
+    "AuthenticationError",
     "ConflictAnalysis",
+    "DataFetchError",
     "KeyDevelopment",
     "NewsItem",
+    "RateLimitExceededError",
+    "RegionalImplications",
+    "ReliabilityAssessment",
+    "RssFeed",
+    "ServiceNotInitializedError",
     "SourceType",
     "SubredditSource",
     "TensionLevel",
     "TensionTrend",
+    "VectorStoreError",
+    "WesternFrontError",
 ]

src/westernfront/core/constants.py

@@ -0,0 +1,123 @@
+"""Constants and default configurations for WesternFront."""
+
+from westernfront.core.models import RssFeed, SubredditSource
+
+RELIABLE_DOMAINS = frozenset([
+    "bbc.com",
+    "reuters.com",
+    "apnews.com",
+    "aljazeera.com",
+    "nytimes.com",
+    "wsj.com",
+    "ft.com",
+    "economist.com",
+    "thediplomat.com",
+    "foreignpolicy.com",
+    "foreignaffairs.com",
+    "dawn.com",
+    "timesofindia.indiatimes.com",
+    "ndtv.com",
+    "geo.tv",
+])
+
+DEFAULT_SUBREDDITS = [
+    SubredditSource(name="geopolitics", reliability_score=0.85),
+    SubredditSource(name="CredibleDefense", reliability_score=0.9),
+    SubredditSource(name="worldnews", reliability_score=0.8),
+    SubredditSource(name="neutralnews", reliability_score=0.8),
+    SubredditSource(name="DefenseNews", reliability_score=0.85),
+    SubredditSource(name="GeopoliticsIndia", reliability_score=0.75),
+    SubredditSource(name="SouthAsia", reliability_score=0.7),
+    SubredditSource(name="india", reliability_score=0.7),
+    SubredditSource(name="pakistan", reliability_score=0.7),
+    SubredditSource(name="Nepal", reliability_score=0.65),
+    SubredditSource(name="bangladesh", reliability_score=0.65),
+    SubredditSource(name="srilanka", reliability_score=0.65),
+    SubredditSource(name="China", reliability_score=0.6),
+]
+
+DEFAULT_RSS_FEEDS = [
+    RssFeed(name="Dawn (Pakistan)", url="https://www.dawn.com/feeds/home", reliability_score=0.85),
+    RssFeed(name="Geo News", url="https://www.geo.tv/rss/1/1", reliability_score=0.8),
+    RssFeed(name="Express Tribune", url="https://tribune.com.pk/feed/home", reliability_score=0.75),
+    RssFeed(name="Times of India", url="https://timesofindia.indiatimes.com/rssfeeds/296589292.cms", reliability_score=0.75),
+    RssFeed(name="NDTV India", url="https://feeds.feedburner.com/ndtvnews-india-news", reliability_score=0.8),
+    RssFeed(name="The Hindu", url="https://www.thehindu.com/news/national/feeder/default.rss", reliability_score=0.85),
+    RssFeed(name="Indian Express", url="https://indianexpress.com/section/india/feed/", reliability_score=0.85),
+    RssFeed(name="South China Morning Post - Asia", url="https://www.scmp.com/rss/91/feed", reliability_score=0.85),
+    RssFeed(name="Kathmandu Post", url="https://kathmandupost.com/rss", reliability_score=0.75),
+    RssFeed(name="Dhaka Tribune", url="https://www.dhakatribune.com/rss", reliability_score=0.75),
+    RssFeed(name="Daily Star Bangladesh", url="https://www.thedailystar.net/rss.xml", reliability_score=0.75),
+    RssFeed(name="Daily Mirror Sri Lanka", url="http://www.dailymirror.lk/RSS_Feeds/breaking-news", reliability_score=0.7),
+]
+
+NEWSAPI_QUERIES = [
+    "India Pakistan",
+    "Kashmir conflict",
+    "India Pakistan border",
+    "LOC firing",
+    "Indo-Pak",
+]
+
+RAG_QUERY_TOPICS = [
+    "India Pakistan military conflict border tensions ceasefire violation",
+    "Kashmir territorial dispute LOC Line of Control",
+    "India China LAC Ladakh Arunachal standoff",
+    "Nepal Bangladesh Sri Lanka India bilateral relations",
+    "South Asia terrorism cross-border insurgency",
+    "India diplomatic relations regional geopolitics",
+    "Military exercises defense buildup South Asia",
+]
+
+NEWSAPI_BASE_URL = "https://newsapi.org/v2"
+
+HTTP_TIMEOUT_SECONDS = 30
+MAX_CONCURRENT_REQUESTS = 10
+
+# Source diversity rules for retrieval
+SOURCE_DIVERSITY_RULES = {
+    "reddit": {"min_pct": 0.25, "max_pct": 0.50},
+    "rss": {"min_pct": 0.30, "max_pct": 0.55},
+    "newsapi": {"min_pct": 0.10, "max_pct": 0.30},
+}
+
+# Temporal weighting for recency boost
+RECENCY_BOOST = {
+    "hours_24": 1.5,
+    "hours_48": 1.25,
+    "days_7": 1.0,
+    "older": 0.75,
+}
+
+# Tension level criteria for validation
+TENSION_LEVEL_CRITERIA = {
+    "LOW": {
+        "score_range": (1, 3),
+        "description": "Normal diplomatic activity, routine border incidents, no escalation",
+    },
+    "MEDIUM": {
+        "score_range": (4, 5),
+        "description": "Heightened rhetoric, minor military movements, diplomatic notes exchanged",
+    },
+    "HIGH": {
+        "score_range": (6, 8),
+        "description": "Military mobilization, cross-border firing, diplomatic summoning",
+    },
+    "CRITICAL": {
+        "score_range": (9, 10),
+        "description": "Active military engagement, imminent conflict, emergency measures",
+    },
+}
+
+SEARCH_KEYWORDS = [
+    "India Pakistan",
+    "Kashmir",
+    "LOC",
+    "ceasefire",
+    "border tension",
+    "military",
+    "diplomatic",
+    "terrorist",
+    "strike",
+    "conflict",
+]

src/westernfront/core/exceptions.py

@@ -0,0 +1,29 @@
+"""Custom exceptions for WesternFront."""
+
+
+class WesternFrontError(Exception):
+    """Base exception for all WesternFront errors."""
+
+
+class AuthenticationError(WesternFrontError):
+    """Raised when API authentication fails."""
+
+
+class RateLimitExceededError(WesternFrontError):
+    """Raised when rate limit is exceeded."""
+
+
+class ServiceNotInitializedError(WesternFrontError):
+    """Raised when a service is accessed before initialization."""
+
+
+class DataFetchError(WesternFrontError):
+    """Raised when fetching data from external sources fails."""
+
+
+class AnalysisError(WesternFrontError):
+    """Raised when AI analysis fails."""
+
+
+class VectorStoreError(WesternFrontError):
+    """Raised when vector store operations fail."""

src/westernfront/core/models.py

@@ -1,7 +1,6 @@
 """Core domain models for WesternFront."""
 
 from datetime import datetime
-from typing import Optional
 
 from pydantic import BaseModel, Field
 
@@ -36,8 +35,8 @@ class NewsItem(BaseModel):
     source_type: SourceType
     published_at: datetime
     reliability_score: float = Field(default=0.5, ge=0.0, le=1.0)
-    author: Optional[str] = None
-    score: Optional[int] = None
+    author: str | None = None
+    score: int | None = None
 
 
 class KeyDevelopment(BaseModel):
@@ -46,7 +45,7 @@ class KeyDevelopment(BaseModel):
     title: str
     description: str
     sources: list[str]
-    timestamp: Optional[datetime] = None
+    timestamp: datetime | None = None
 
 
 class ReliabilityAssessment(BaseModel):

src/westernfront/dependencies.py

@@ -1,21 +1,24 @@
-"""
-Dependency injection container for WesternFront.
-
-Provides FastAPI dependencies for services with proper lifecycle management.
-"""
+"""Dependency injection container for WesternFront."""
 
+import asyncio
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
-from typing import AsyncGenerator
+from typing import TYPE_CHECKING
 
 from loguru import logger
 
+if TYPE_CHECKING:
+    from fastapi import Request
+
+from westernfront.analytics import AnalyticsAggregator
 from westernfront.config import Settings, get_settings
 from westernfront.repositories.analysis import AnalysisRepository
 from westernfront.repositories.vectors import VectorRepository
 from westernfront.services.analysis import AnalysisService
 from westernfront.services.cache import CacheService
 from westernfront.services.embeddings import EmbeddingService
+from westernfront.services.http import HttpService
 from westernfront.services.newsapi import NewsApiService
 from westernfront.services.reddit import RedditService
 from westernfront.services.rss import RssService
@@ -26,6 +29,7 @@ class AppState:
     """Container for application-scoped services."""
 
     settings: Settings
+    http: HttpService
     cache: CacheService
     reddit: RedditService
     rss: RssService
@@ -34,16 +38,28 @@ class AppState:
     embeddings: EmbeddingService
     vectors: VectorRepository
     analysis: AnalysisService
+    analytics: AnalyticsAggregator
+
 
+async def _init_sync_services(settings: Settings) -> tuple[EmbeddingService, VectorRepository]:
+    """Initialize synchronous services in a thread pool."""
+    def _init_embeddings_and_vectors() -> tuple[EmbeddingService, VectorRepository]:
+        embeddings = EmbeddingService()
+        embeddings.initialize()
+        vectors = VectorRepository(embedding_service=embeddings)
+        vectors.initialize()
+        return embeddings, vectors
 
-_app_state: AppState | None = None
+    return await asyncio.to_thread(_init_embeddings_and_vectors)
 
 
 async def init_services() -> AppState:
-    """Initialize all services for the application."""
+    """Initialize all services for the application with parallel execution."""
     settings = get_settings()
 
+    http = HttpService()
     cache = CacheService(ttl_seconds=settings.cache_expiry_minutes * 60)
+    repository = AnalysisRepository(db_path=settings.database_path)
 
     reddit = RedditService(
         client_id=settings.reddit_client_id,
@@ -51,23 +67,23 @@ async def init_services() -> AppState:
         user_agent=settings.reddit_user_agent,
         cache=cache,
     )
-    await reddit.initialize()
-
-    rss = RssService(cache=cache)
 
-    newsapi = NewsApiService(api_key=settings.newsapi_key, cache=cache)
+    rss = RssService(cache=cache, http=http)
+    newsapi = NewsApiService(api_key=settings.newsapi_key, cache=cache, http=http)
 
-    repository = AnalysisRepository(db_path=settings.database_path)
-    await repository.initialize()
-
-    embeddings = EmbeddingService()
-    embeddings.initialize()
-    logger.info("Embedding service initialized")
+    # Parallel initialization of independent services
+    http_init, reddit_init, repo_init, vectors_result = await asyncio.gather(
+        http.initialize(),
+        reddit.initialize(),
+        repository.initialize(),
+        _init_sync_services(settings),
+    )
 
-    vectors = VectorRepository(embedding_service=embeddings)
-    vectors.initialize()
+    embeddings, vectors = vectors_result
     logger.info(f"Vector repository initialized with {vectors.get_count()} items")
 
+    analytics = AnalyticsAggregator(repository)
+
     analysis = AnalysisService(
         gemini_api_key=settings.gemini_api_key,
         reddit=reddit,
@@ -81,6 +97,7 @@ async def init_services() -> AppState:
 
     return AppState(
         settings=settings,
+        http=http,
         cache=cache,
         reddit=reddit,
         rss=rss,
@@ -89,6 +106,7 @@ async def init_services() -> AppState:
         embeddings=embeddings,
         vectors=vectors,
         analysis=analysis,
+        analytics=analytics,
     )
 
 
@@ -96,43 +114,35 @@ async def shutdown_services(state: AppState) -> None:
     """Clean up all services."""
     await state.analysis.close()
     await state.reddit.close()
+    await state.http.close()
     await state.repository.close()
 
 
 @asynccontextmanager
 async def lifespan_context() -> AsyncGenerator[AppState, None]:
     """Lifespan context manager for FastAPI."""
-    global _app_state
-    _app_state = await init_services()
+    state = await init_services()
     try:
-        yield _app_state
+        yield state
     finally:
-        await shutdown_services(_app_state)
-        _app_state = None
-
-
-def get_app_state() -> AppState:
-    """Get the current application state."""
-    if _app_state is None:
-        raise RuntimeError("Application not initialized")
-    return _app_state
+        await shutdown_services(state)
 
 
-def get_analysis_service() -> AnalysisService:
-    """FastAPI dependency for AnalysisService."""
-    return get_app_state().analysis
+def get_state_from_request(request: "Request") -> AppState:
+    """Get application state from request."""
+    return request.app.state.westernfront
 
 
-def get_repository() -> AnalysisRepository:
-    """FastAPI dependency for AnalysisRepository."""
-    return get_app_state().repository
+def get_analysis_service(request: "Request") -> AnalysisService:
+    """Get AnalysisService from request."""
+    return get_state_from_request(request).analysis
 
 
-def get_vectors() -> VectorRepository:
-    """FastAPI dependency for VectorRepository."""
-    return get_app_state().vectors
+def get_repository(request: "Request") -> AnalysisRepository:
+    """Get AnalysisRepository from request."""
+    return get_state_from_request(request).repository
 
 
-def get_settings_dep() -> Settings:
-    """FastAPI dependency for Settings."""
-    return get_app_state().settings
+def get_analytics(request: "Request") -> AnalyticsAggregator:
+    """Get AnalyticsAggregator from request."""
+    return get_state_from_request(request).analytics

src/westernfront/main.py

@@ -1,15 +1,16 @@
 """FastAPI application factory and entry point."""
 
 import os
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-from typing import AsyncGenerator
 
-from fastapi import FastAPI, Request
+from fastapi import FastAPI, Request, Response
 from fastapi.middleware.cors import CORSMiddleware
 from loguru import logger
 
 from westernfront import __version__
 from westernfront.api.auth import verify_api_key
+from westernfront.api.middleware import RateLimitMiddleware
 from westernfront.api.routes import router
 from westernfront.config import get_settings
 from westernfront.dependencies import lifespan_context
@@ -43,6 +44,8 @@ def create_app() -> FastAPI:
         lifespan=lifespan,
     )
 
+    app.add_middleware(RateLimitMiddleware, requests_per_minute=120, burst_size=20)
+
     app.add_middleware(
         CORSMiddleware,
         allow_origins=settings.allowed_origins,
@@ -52,7 +55,19 @@ def create_app() -> FastAPI:
     )
 
     @app.middleware("http")
-    async def api_key_middleware(request: Request, call_next):
+    async def security_headers_middleware(request: Request, call_next) -> Response:
+        """Add security headers to all responses."""
+        response = await call_next(request)
+        response.headers["X-Content-Type-Options"] = "nosniff"
+        response.headers["X-Frame-Options"] = "DENY"
+        response.headers["X-XSS-Protection"] = "1; mode=block"
+        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
+        response.headers["Content-Security-Policy"] = "default-src 'self'; frame-ancestors 'none'"
+        return response
+
+    @app.middleware("http")
+    async def api_key_middleware(request: Request, call_next) -> Response:
+        """Verify API key for protected endpoints."""
         await verify_api_key(request)
         return await call_next(request)
 

src/westernfront/prompts/analysis.py

@@ -2,33 +2,53 @@
 
 from datetime import datetime
 
+from westernfront.core.constants import TENSION_LEVEL_CRITERIA
 
-def build_rag_prompt(retrieved_items: list[dict], total_in_memory: int = 0) -> str:
-    """
-    Build prompt for RAG-enhanced analysis using vector-retrieved items.
+RELIABILITY_THRESHOLDS = {
+    "HIGH": 0.8,
+    "MEDIUM": 0.6,
+}
+
+
+def _get_reliability_label(score: float) -> str:
+    """Convert reliability score to label."""
+    if score > RELIABILITY_THRESHOLDS["HIGH"]:
+        return "HIGH"
+    if score > RELIABILITY_THRESHOLDS["MEDIUM"]:
+        return "MEDIUM"
+    return "LOW"
+
+
+def _build_tension_criteria() -> str:
+    """Build tension level criteria section for prompt."""
+    lines = []
+    for level, criteria in TENSION_LEVEL_CRITERIA.items():
+        score_range = criteria["score_range"]
+        desc = criteria["description"]
+        lines.append(f"- **{level}** (Score {score_range[0]}-{score_range[1]}): {desc}")
+    return "\n".join(lines)
 
-    Args:
-        retrieved_items: Items retrieved from vector search with metadata.
-        total_in_memory: Total items in the vector database.
 
-    Returns:
-        The formatted prompt string.
-    """
+def build_rag_prompt(retrieved_items: list[dict], total_in_memory: int = 0) -> str:
+    """Build prompt for RAG-enhanced analysis using vector-retrieved items."""
     source_entries = []
     for i, item in enumerate(retrieved_items):
         meta = item.get("metadata", {})
         doc = item.get("document", "")
-        score = item.get("similarity_score", 0)
+        score = item.get("boosted_score", item.get("similarity_score", 0))
         reliability_val = meta.get("reliability_score", 0.5)
-        
-        reliability = "HIGH" if reliability_val > 0.8 else (
-            "MEDIUM" if reliability_val > 0.6 else "LOW"
-        )
-        
+        reliability = _get_reliability_label(reliability_val)
+
+        published = meta.get("published_at", "")
+        date_str = published[:10] if published else "Unknown"
+
+        recency = item.get("recency_multiplier", 1.0)
+        recency_label = "FRESH" if recency >= 1.5 else ("RECENT" if recency >= 1.0 else "OLDER")
+
         entry = (
-            f"INTEL #{i + 1} [Relevance: {score:.0%}] [Reliability: {reliability}]:\n"
+            f"INTEL #{i + 1} [Relevance: {score:.0%}] [Reliability: {reliability}] [{recency_label}]:\n"
             f"Source: {meta.get('source_name', 'Unknown')} ({meta.get('source_type', 'unknown')})\n"
-            f"Date: {meta.get('published_at', 'Unknown')[:10] if meta.get('published_at') else 'Unknown'}\n"
+            f"Date: {date_str}\n"
             f"Content: {doc}"
         )
         source_entries.append(entry)
@@ -37,7 +57,13 @@ def build_rag_prompt(retrieved_items: list[dict], total_in_memory: int = 0) -> s
 
     memory_note = ""
     if total_in_memory > 0:
-        memory_note = f"\n\n**INSTITUTIONAL MEMORY:** This analysis draws from a database of {total_in_memory:,} indexed news items. The items shown below are the most semantically relevant to South Asia conflict dynamics.\n"
+        memory_note = (
+            f"\n\n**INSTITUTIONAL MEMORY:** This analysis draws from a database of "
+            f"{total_in_memory:,} indexed news items. The items shown below are the most "
+            f"semantically relevant to South Asia conflict dynamics, weighted by recency.\n"
+        )
+
+    tension_criteria = _build_tension_criteria()
 
     return f"""**TOP SECRET // FOR OFFICIAL USE ONLY**
 
@@ -59,6 +85,11 @@ Analyze ALL matters relevant to India's regional relationships, prioritized as:
 **NEUTRALITY DIRECTIVE:**
 Maintain absolute neutrality. Present multiple perspectives when conflicting information exists. Acknowledge information gaps.
 
+**TENSION LEVEL ASSESSMENT CRITERIA:**
+{tension_criteria}
+
+Use these criteria strictly. Your tension_score MUST align with your tension_level.
+
 **INTELLIGENCE FEEDS:**
 ---
 {intelligence_data}
@@ -67,8 +98,9 @@ Maintain absolute neutrality. Present multiple perspectives when conflicting inf
 **ANALYTICAL DIRECTIVES:**
 1. **Synthesize, Do Not Summarize:** Integrate all data into a coherent assessment.
 2. **Impersonal Tone:** Use formal, analytical language.
-3. **No Direct Attribution:** Your report is standalone. Do not attribute to specific sources.
+3. **Ground All Claims:** Every key entity and event must be supported by the intelligence feeds above.
 4. **Acknowledge Uncertainty:** Indicate confidence levels and information gaps.
+5. **Prioritize Recency:** Weight FRESH sources more heavily than OLDER ones.
 
 **REQUIRED OUTPUT FORMAT (Strict JSON):**
 Produce a single, valid JSON object:
@@ -83,8 +115,8 @@ Produce a single, valid JSON object:
         }}
     ],
     "reliability_assessment": {{
-        "source_credibility": "Overall credibility assessment.",
-        "information_gaps": "What critical information is missing.",
+        "source_credibility": "Overall credibility assessment of available sources.",
+        "information_gaps": "What critical information is missing from the intelligence feeds.",
         "confidence_rating": "HIGH, MEDIUM, or LOW with justification."
     }},
     "regional_implications": {{
@@ -93,9 +125,72 @@ Produce a single, valid JSON object:
         "economic": "Potential economic consequences."
     }},
     "tension_level": "LOW|MEDIUM|HIGH|CRITICAL",
-    "tension_rationale": "Justification for the assessed tension level.",
-    "tension_score": "Integer 1-10 (1=calm, 10=active conflict imminent).",
+    "tension_rationale": "Justification for the assessed tension level using specific evidence from sources.",
+    "tension_score": "Integer 1-10 matching to tension level per criteria above.",
     "tension_trend": "INCREASING|DECREASING|STABLE",
     "analysis_type": "Military|Diplomatic|Internal Security|Political|Other",
-    "key_entities": ["3-5 key actors, locations, or organizations."]
+    "key_entities": ["3-5 key actors, locations, or organizations ONLY from the sources above."]
+}}"""
+
+
+def build_extraction_prompt(items: list[dict]) -> str:
+    """Build prompt for fact extraction stage of chain analysis."""
+    source_entries = []
+    for i, item in enumerate(items[:20]):
+        doc = item.get("document", "")
+        source_entries.append(f"SOURCE {i+1}: {doc[:500]}")
+
+    sources_text = "\n\n".join(source_entries)
+
+    return f"""Extract key facts from the following intelligence sources.
+
+SOURCES:
+{sources_text}
+
+OUTPUT FORMAT (JSON):
+{{
+    "facts": [
+        {{
+            "fact": "Clear, objective statement of fact",
+            "source_index": 1,
+            "type": "military|diplomatic|political|economic|other",
+            "date_mentioned": "YYYY-MM-DD or null",
+            "entities": ["entity1", "entity2"]
+        }}
+    ],
+    "total_sources_analyzed": {len(items)}
+}}"""
+
+
+def build_synthesis_prompt(facts: list[dict], historical_context: str = "") -> str:
+    """Build prompt for synthesis stage of chain analysis."""
+    facts_text = "\n".join([f"- {f.get('fact', '')}" for f in facts[:30]])
+
+    return f"""Synthesize the following extracted facts into a coherent assessment.
+
+EXTRACTED FACTS:
+{facts_text}
+
+{f"HISTORICAL CONTEXT: {historical_context}" if historical_context else ""}
+
+TASK:
+1. Identify the 3-5 most significant developments
+2. Assess overall tension level (LOW/MEDIUM/HIGH/CRITICAL)
+3. Identify trends and patterns
+4. Note any contradictions or information gaps
+
+OUTPUT FORMAT (JSON):
+{{
+    "significant_developments": [
+        {{
+            "title": "Development title",
+            "description": "Synthesized description",
+            "supporting_facts": [0, 1, 2]
+        }}
+    ],
+    "preliminary_tension": "LOW|MEDIUM|HIGH|CRITICAL",
+    "tension_reasoning": "Brief reasoning",
+    "trends": ["trend1", "trend2"],
+    "contradictions": ["any contradictory information"],
+    "gaps": ["information gaps identified"]
 }}"""

src/westernfront/repositories/analysis.py

@@ -1,14 +1,14 @@
 """SQLite repository for storing analysis history."""
 
 import json
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from typing import Optional
 
 import aiosqlite
 from loguru import logger
 
 from westernfront.core.enums import AnalysisType, TensionLevel, TensionTrend
+from westernfront.core.exceptions import ServiceNotInitializedError
 from westernfront.core.models import AnalysisSnapshot, ConflictAnalysis
 
 
@@ -16,14 +16,8 @@ class AnalysisRepository:
     """SQLite-based repository for analysis storage and retrieval."""
 
     def __init__(self, db_path: str = "data/westernfront.db") -> None:
-        """
-        Initialize the repository.
-
-        Args:
-            db_path: Path to the SQLite database file.
-        """
         self._db_path = Path(db_path)
-        self._conn: Optional[aiosqlite.Connection] = None
+        self._conn: aiosqlite.Connection | None = None
 
     async def initialize(self) -> None:
         """Initialize the database and create tables."""
@@ -43,7 +37,7 @@ class AnalysisRepository:
     async def _create_tables(self) -> None:
         """Create required database tables."""
         if not self._conn:
-            raise RuntimeError("Repository not initialized")
+            raise ServiceNotInitializedError("Analysis repository not initialized")
 
         await self._conn.execute("""
             CREATE TABLE IF NOT EXISTS analyses (
@@ -70,7 +64,7 @@ class AnalysisRepository:
         if not self._conn:
             return
 
-        cutoff = (datetime.now(timezone.utc) - timedelta(days=retention_days)).isoformat()
+        cutoff = (datetime.now(UTC) - timedelta(days=retention_days)).isoformat()
         cursor = await self._conn.execute(
             "DELETE FROM analyses WHERE generated_at < ?",
             (cutoff,),
@@ -81,14 +75,9 @@ class AnalysisRepository:
             logger.info(f"Cleaned up {cursor.rowcount} old analysis records")
 
     async def save(self, analysis: ConflictAnalysis) -> None:
-        """
-        Save an analysis to the database.
-
-        Args:
-            analysis: The conflict analysis to save.
-        """
+        """Save an analysis to the database."""
         if not self._conn:
-            raise RuntimeError("Repository not initialized")
+            raise ServiceNotInitializedError("Analysis repository not initialized")
 
         await self._conn.execute(
             """
@@ -112,10 +101,10 @@ class AnalysisRepository:
         await self._conn.commit()
         logger.debug(f"Saved analysis {analysis.analysis_id}")
 
-    async def get_latest(self) -> Optional[ConflictAnalysis]:
+    async def get_latest(self) -> ConflictAnalysis | None:
         """Get the most recent analysis."""
         if not self._conn:
-            raise RuntimeError("Repository not initialized")
+            raise ServiceNotInitializedError("Analysis repository not initialized")
 
         cursor = await self._conn.execute(
             "SELECT full_analysis FROM analyses ORDER BY generated_at DESC LIMIT 1"
@@ -131,20 +120,11 @@ class AnalysisRepository:
         days: int = 30,
         limit: int = 100,
     ) -> list[AnalysisSnapshot]:
-        """
-        Get historical analysis snapshots.
-
-        Args:
-            days: Number of days to look back.
-            limit: Maximum number of records to return.
-
-        Returns:
-            List of analysis snapshots.
-        """
+        """Get historical analysis snapshots."""
         if not self._conn:
-            raise RuntimeError("Repository not initialized")
+            raise ServiceNotInitializedError("Analysis repository not initialized")
 
-        cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
+        cutoff = (datetime.now(UTC) - timedelta(days=days)).isoformat()
 
         cursor = await self._conn.execute(
             """
@@ -178,19 +158,11 @@ class AnalysisRepository:
         return snapshots
 
     async def get_tension_history(self, days: int = 30) -> list[dict]:
-        """
-        Get tension score history for graphing.
-
-        Args:
-            days: Number of days to look back.
-
-        Returns:
-            List of date/score pairs.
-        """
+        """Get tension score history for graphing."""
         if not self._conn:
-            raise RuntimeError("Repository not initialized")
+            raise ServiceNotInitializedError("Analysis repository not initialized")
 
-        cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
+        cutoff = (datetime.now(UTC) - timedelta(days=days)).isoformat()
 
         cursor = await self._conn.execute(
             """

src/westernfront/repositories/vectors.py

@@ -1,13 +1,14 @@
 """ChromaDB vector repository for semantic search."""
 
-from datetime import datetime
+import asyncio
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING
 
 import chromadb
 from chromadb.config import Settings as ChromaSettings
 from loguru import logger
 
+from westernfront.core.exceptions import VectorStoreError
 from westernfront.core.models import NewsItem
 
 if TYPE_CHECKING:
@@ -21,16 +22,9 @@ class VectorRepository:
 
     def __init__(
         self,
-        persist_dir: Optional[str] = None,
-        embedding_service: Optional["EmbeddingService"] = None,
+        persist_dir: str | None = None,
+        embedding_service: "EmbeddingService | None" = None,
     ) -> None:
-        """
-        Initialize the vector repository.
-
-        Args:
-            persist_dir: Directory for ChromaDB persistence (defaults to server/data/chroma).
-            embedding_service: Service for generating embeddings.
-        """
         if persist_dir:
             self._persist_dir = Path(persist_dir)
         else:
@@ -39,54 +33,36 @@ class VectorRepository:
         self._persist_dir.mkdir(parents=True, exist_ok=True)
 
         self._embeddings = embedding_service
-        self._client: Optional[chromadb.PersistentClient] = None
+        self._client: chromadb.PersistentClient | None = None
         self._collection = None
 
     def initialize(self) -> bool:
-        """
-        Initialize ChromaDB client and collection.
-
-        Returns:
-            True if initialization was successful.
-        """
-        try:
-            logger.info(f"Initializing ChromaDB at {self._persist_dir}")
+        """Initialize ChromaDB client and collection."""
+        logger.info(f"Initializing ChromaDB at {self._persist_dir}")
 
-            self._client = chromadb.PersistentClient(
-                path=str(self._persist_dir),
-                settings=ChromaSettings(anonymized_telemetry=False),
-            )
-
-            self._collection = self._client.get_or_create_collection(
-                name=self.COLLECTION_NAME,
-                metadata={"hnsw:space": "cosine"},
-            )
+        self._client = chromadb.PersistentClient(
+            path=str(self._persist_dir),
+            settings=ChromaSettings(anonymized_telemetry=False),
+        )
 
-            count = self._collection.count()
-            logger.info(f"ChromaDB initialized with {count} existing documents")
-            return True
+        self._collection = self._client.get_or_create_collection(
+            name=self.COLLECTION_NAME,
+            metadata={"hnsw:space": "cosine"},
+        )
 
-        except Exception as e:
-            logger.error(f"Failed to initialize ChromaDB: {e}")
-            return False
+        count = self._collection.count()
+        logger.info(f"ChromaDB initialized with {count} existing documents")
+        return True
 
     @property
     def is_initialized(self) -> bool:
         """Check if the repository is initialized."""
         return self._collection is not None
 
-    def add_items(self, items: list[NewsItem]) -> int:
-        """
-        Add news items to the vector store.
-
-        Args:
-            items: News items to add.
-
-        Returns:
-            Number of items added.
-        """
+    def _add_items_sync(self, items: list[NewsItem]) -> int:
+        """Synchronous implementation of add_items."""
         if not self._collection or not self._embeddings:
-            raise RuntimeError("Repository not initialized")
+            raise VectorStoreError("Vector repository not initialized")
 
         existing_ids = set(self._collection.get(ids=[item.id for item in items])["ids"])
         new_items = [item for item in items if item.id not in existing_ids]
@@ -117,25 +93,23 @@ class VectorRepository:
         logger.info(f"Added {len(new_items)} items to vector store")
         return len(new_items)
 
+    def add_items(self, items: list[NewsItem]) -> int:
+        """Add news items to the vector store (blocking)."""
+        return self._add_items_sync(items)
+
+    async def add_items_async(self, items: list[NewsItem]) -> int:
+        """Add news items to the vector store (non-blocking)."""
+        return await asyncio.to_thread(self._add_items_sync, items)
+
     def query_similar(
         self,
         query: str,
         n_results: int = 20,
         min_score: float = 0.3,
     ) -> list[dict]:
-        """
-        Query for similar items.
-
-        Args:
-            query: Query text.
-            n_results: Maximum number of results.
-            min_score: Minimum similarity score (0-1).
-
-        Returns:
-            List of similar items with metadata.
-        """
+        """Query for similar items."""
         if not self._collection or not self._embeddings:
-            raise RuntimeError("Repository not initialized")
+            raise VectorStoreError("Vector repository not initialized")
 
         query_embedding = self._embeddings.embed(query)
 
@@ -164,16 +138,7 @@ class VectorRepository:
         topics: list[str],
         n_per_topic: int = 10,
     ) -> list[dict]:
-        """
-        Query for items related to multiple topics.
-
-        Args:
-            topics: List of topic queries.
-            n_per_topic: Results per topic.
-
-        Returns:
-            Deduplicated list of relevant items.
-        """
+        """Query for items related to multiple topics."""
         seen_ids: set[str] = set()
         all_items: list[dict] = []
 

src/westernfront/services/__init__.py

@@ -2,16 +2,28 @@
 
 from westernfront.services.analysis import AnalysisService
 from westernfront.services.cache import CacheService
+from westernfront.services.chain_analysis import ChainAnalysisService
 from westernfront.services.embeddings import EmbeddingService
+from westernfront.services.http import HttpService
 from westernfront.services.newsapi import NewsApiService
+from westernfront.services.parsing import ResponseParser
 from westernfront.services.reddit import RedditService
+from westernfront.services.retrieval import RetrievalService
 from westernfront.services.rss import RssService
+from westernfront.services.scheduler import AnalysisScheduler
+from westernfront.services.validation import AnalysisValidator
 
 __all__ = [
+    "AnalysisScheduler",
     "AnalysisService",
+    "AnalysisValidator",
     "CacheService",
+    "ChainAnalysisService",
     "EmbeddingService",
+    "HttpService",
     "NewsApiService",
     "RedditService",
+    "ResponseParser",
+    "RetrievalService",
     "RssService",
 ]

src/westernfront/services/analysis.py

@@ -1,362 +1,239 @@
-"""AI-powered conflict analysis service with RAG enhancement."""
-
-import asyncio
-import json
-import re
-import uuid
-from datetime import datetime
-from typing import Any, Optional
-
-import google.generativeai as genai
-from loguru import logger
-from tenacity import RetryError, retry, stop_after_attempt, wait_exponential
-
-from westernfront.config import Settings
-from westernfront.core.enums import AnalysisType, SourceType, TensionLevel, TensionTrend
-from westernfront.core.models import (
-    ConflictAnalysis,
-    KeyDevelopment,
-    NewsItem,
-    RegionalImplications,
-    ReliabilityAssessment,
-)
-from westernfront.prompts.analysis import build_rag_prompt
-from westernfront.repositories.analysis import AnalysisRepository
-from westernfront.repositories.vectors import VectorRepository
-from westernfront.services.newsapi import NewsApiService
-from westernfront.services.reddit import RedditService
-from westernfront.services.rss import RssService
-
-
-# Topics for RAG retrieval - these are semantic queries, not keyword filters
-RAG_QUERY_TOPICS = [
-    "India Pakistan military conflict border tensions ceasefire violation",
-    "Kashmir territorial dispute LOC Line of Control",
-    "India China LAC Ladakh Arunachal standoff",
-    "Nepal Bangladesh Sri Lanka India bilateral relations",
-    "South Asia terrorism cross-border insurgency",
-    "India diplomatic relations regional geopolitics",
-    "Military exercises defense buildup South Asia",
-]
-
-
-class AnalysisService:
-    """Service for generating AI-powered conflict analysis with RAG."""
-
-    def __init__(
-        self,
-        gemini_api_key: str,
-        reddit: RedditService,
-        rss: RssService,
-        newsapi: NewsApiService,
-        repository: AnalysisRepository,
-        vectors: VectorRepository,
-        settings: Settings,
-    ) -> None:
-        """
-        Initialize the analysis service.
-
-        Args:
-            gemini_api_key: API key for Google Gemini.
-            reddit: Reddit service for fetching posts.
-            rss: RSS service for fetching articles.
-            newsapi: NewsAPI service for fetching news.
-            repository: Repository for storing analyses.
-            vectors: Vector repository for RAG retrieval.
-            settings: Application settings.
-        """
-        self._api_key = gemini_api_key
-        self._reddit = reddit
-        self._rss = rss
-        self._newsapi = newsapi
-        self._repository = repository
-        self._vectors = vectors
-        self._settings = settings
-        self._model: Optional[genai.GenerativeModel] = None
-        self._update_task: Optional[asyncio.Task[None]] = None
-
-    async def initialize(self) -> None:
-        """Initialize the Gemini model and start background updates."""
-        logger.info("Initializing Gemini AI")
-        genai.configure(api_key=self._api_key)
-        self._model = genai.GenerativeModel(
-            "gemma-3-27b-it",
-            generation_config={
-                "temperature": 0.2,
-                "top_p": 0.95,
-                "top_k": 40,
-            },
-        )
-        logger.info("Gemini AI initialized")
-
-        self._update_task = asyncio.create_task(self._periodic_update())
-
-    async def close(self) -> None:
-        """Clean up resources."""
-        if self._update_task:
-            self._update_task.cancel()
-            try:
-                await self._update_task
-            except asyncio.CancelledError:
-                pass
-        logger.info("Analysis service closed")
-
-    @property
-    def is_initialized(self) -> bool:
-        """Check if the service is initialized."""
-        return self._model is not None
-
-    async def _periodic_update(self) -> None:
-        """Background task for periodic analysis updates."""
-        await asyncio.sleep(5)
-        await self._run_update("startup")
-
-        interval = self._settings.update_interval_minutes * 60
-        while True:
-            await asyncio.sleep(interval)
-            await self._run_update("scheduled")
-
-    async def _run_update(self, trigger: str) -> None:
-        """Execute an analysis update."""
-        try:
-            logger.info(f"Starting analysis update (trigger: {trigger})")
-            analysis = await self.generate_analysis(trigger=trigger)
-            if analysis:
-                logger.info(f"Analysis complete. Tension: {analysis.tension_level.value}")
-            else:
-                logger.warning("No analysis generated")
-        except Exception as e:
-            logger.error(f"Error in update: {e}")
-
-    async def _ingest_all_news(self) -> int:
-        """
-        Ingest ALL news from all sources into vector store.
-        
-        No keyword filtering - we store everything and let vector search
-        determine relevance.
-        
-        Returns:
-            Number of new items added.
-        """
-        days = self._settings.analysis_days_back
-
-        reddit_task = self._reddit.get_all_posts(days)
-        rss_task = self._rss.get_all_articles(days)
-        newsapi_task = self._newsapi.get_related_articles(days_back=days)
-
-        results = await asyncio.gather(
-            reddit_task, rss_task, newsapi_task,
-            return_exceptions=True,
-        )
-
-        all_items: list[NewsItem] = []
-        source_names = ["Reddit", "RSS", "NewsAPI"]
-
-        for i, result in enumerate(results):
-            if isinstance(result, Exception):
-                logger.error(f"Error fetching from {source_names[i]}: {result}")
-            else:
-                all_items.extend(result)
-
-        logger.info(f"Ingested {len(all_items)} total news items from all sources")
-
-        if not self._vectors.is_initialized:
-            logger.warning("Vector store not initialized, skipping ingestion")
-            return 0
-
-        stored = self._vectors.add_items(all_items)
-        total_count = self._vectors.get_count()
-        logger.info(f"Stored {stored} new items. Total in vector store: {total_count}")
-        return stored
-
-    def _retrieve_relevant_items(self, max_items: int = 30) -> list[dict]:
-        """
-        Retrieve relevant items from vector store using semantic search.
-        
-        This is the core of RAG - we query by topic and get the most
-        semantically relevant items regardless of when they were published.
-        
-        Args:
-            max_items: Maximum items to retrieve.
-            
-        Returns:
-            List of relevant items with metadata.
-        """
-        if not self._vectors.is_initialized:
-            logger.warning("Vector store not initialized")
-            return []
-
-        all_results = self._vectors.query_by_topics(
-            RAG_QUERY_TOPICS, 
-            n_per_topic=max_items // len(RAG_QUERY_TOPICS) + 1,
-        )
-
-        results = all_results[:max_items]
-        logger.info(f"Retrieved {len(results)} relevant items via vector search")
-        return results
-
-    @retry(wait=wait_exponential(min=2, max=60), stop=stop_after_attempt(3))
-    async def _call_gemini(self, prompt: str) -> Optional[dict[str, Any]]:
-        """Call Gemini API with retry logic."""
-        if not self._model:
-            raise RuntimeError("Gemini not initialized")
-
-        logger.info("Calling Gemini API")
-        response = await self._model.generate_content_async(prompt)
-        text = response.text
-
-        try:
-            return json.loads(text)
-        except json.JSONDecodeError:
-            json_match = re.search(r"```(?:json)?\n(.*?)\n```", text, re.DOTALL)
-            if json_match:
-                return json.loads(json_match.group(1))
-
-            json_match = re.search(r"\{.*\}", text, re.DOTALL)
-            if json_match:
-                return json.loads(json_match.group(0))
-
-            logger.error(f"Failed to parse JSON: {text[:200]}...")
-            raise ValueError("Could not parse JSON from response")
-
-    def _parse_tension_level(self, value: str) -> TensionLevel:
-        """Parse tension level from string."""
-        value = value.upper()
-        if "CRITICAL" in value:
-            return TensionLevel.CRITICAL
-        if "HIGH" in value:
-            return TensionLevel.HIGH
-        if "MEDIUM" in value:
-            return TensionLevel.MEDIUM
-        return TensionLevel.LOW
-
-    def _parse_tension_trend(self, value: str) -> TensionTrend:
-        """Parse tension trend from string."""
-        value = value.upper()
-        if "INCREASING" in value:
-            return TensionTrend.INCREASING
-        if "DECREASING" in value:
-            return TensionTrend.DECREASING
-        return TensionTrend.STABLE
-
-    def _parse_analysis_type(self, value: str) -> AnalysisType:
-        """Parse analysis type from string."""
-        value = value.upper()
-        if "MILITARY" in value:
-            return AnalysisType.MILITARY
-        if "DIPLOMATIC" in value:
-            return AnalysisType.DIPLOMATIC
-        if "INTERNAL" in value:
-            return AnalysisType.INTERNAL_SECURITY
-        if "POLITICAL" in value:
-            return AnalysisType.POLITICAL
-        return AnalysisType.OTHER
-
-    def _parse_key_developments(self, data: list[dict]) -> list[KeyDevelopment]:
-        """Parse key developments from response data."""
-        developments = []
-        for item in data:
-            if not isinstance(item, dict):
-                continue
-            developments.append(
-                KeyDevelopment(
-                    title=item.get("title", "Unnamed"),
-                    description=item.get("description", "No description"),
-                    sources=item.get("sources", []),
-                    timestamp=datetime.now(),
-                )
-            )
-        return developments
-
-    def _count_sources(self, items: list[dict]) -> dict[str, int]:
-        """Count items by source type from retrieved results."""
-        counts: dict[str, int] = {}
-        for item in items:
-            meta = item.get("metadata", {})
-            key = meta.get("source_type", "unknown")
-            counts[key] = counts.get(key, 0) + 1
-        return counts
-
-    async def generate_analysis(self, trigger: str = "scheduled") -> Optional[ConflictAnalysis]:
-        """
-        Generate a new conflict analysis using RAG.
-
-        Flow:
-        1. Ingest ALL news into vector store
-        2. Retrieve most relevant items via semantic search
-        3. Send retrieved items to Gemini for analysis
-
-        Args:
-            trigger: What triggered this analysis.
-
-        Returns:
-            The generated analysis or None.
-        """
-        await self._ingest_all_news()
-
-        retrieved_items = self._retrieve_relevant_items(
-            max_items=self._settings.max_posts_for_analysis
-        )
-
-        if len(retrieved_items) < self._settings.min_posts_for_analysis:
-            logger.warning(f"Insufficient data: {len(retrieved_items)} items")
-            return None
-
-        prompt = build_rag_prompt(retrieved_items, self._vectors.get_count())
-
-        try:
-            data = await self._call_gemini(prompt)
-        except RetryError as e:
-            logger.error(f"Gemini failed after retries: {e}")
-            return None
-
-        if not data:
-            return None
-
-        tension_score = 1
-        raw_score = data.get("tension_score")
-        if isinstance(raw_score, (int, float)):
-            tension_score = max(1, min(10, int(raw_score)))
-        elif isinstance(raw_score, str) and raw_score.isdigit():
-            tension_score = max(1, min(10, int(raw_score)))
-
-        key_entities = data.get("key_entities", [])
-        if isinstance(key_entities, str):
-            key_entities = [e.strip() for e in key_entities.split(",") if e.strip()]
-
-        reliability_data = data.get("reliability_assessment", {})
-        regional_data = data.get("regional_implications", {})
-
-        analysis = ConflictAnalysis(
-            analysis_id=str(uuid.uuid4()),
-            generated_at=datetime.now(),
-            latest_status=data.get("latest_status", "No status available"),
-            situation_summary=data.get("situation_summary", "No summary available"),
-            key_developments=self._parse_key_developments(data.get("key_developments", [])),
-            reliability_assessment=ReliabilityAssessment(
-                source_credibility=reliability_data.get("source_credibility", "Unknown"),
-                information_gaps=reliability_data.get("information_gaps", "Unknown"),
-                confidence_rating=reliability_data.get("confidence_rating", "LOW"),
-            ),
-            regional_implications=RegionalImplications(
-                security=regional_data.get("security", "No assessment"),
-                diplomatic=regional_data.get("diplomatic", "No assessment"),
-                economic=regional_data.get("economic", "No assessment"),
-            ),
-            tension_level=self._parse_tension_level(data.get("tension_level", "LOW")),
-            tension_rationale=data.get("tension_rationale", "No rationale"),
-            tension_score=tension_score,
-            tension_trend=self._parse_tension_trend(data.get("tension_trend", "STABLE")),
-            analysis_type=self._parse_analysis_type(data.get("analysis_type", "OTHER")),
-            key_entities=key_entities if isinstance(key_entities, list) else [],
-            source_count=len(retrieved_items),
-            source_breakdown=self._count_sources(retrieved_items),
-        )
-
-        await self._repository.save(analysis)
-        logger.info(f"Generated RAG-enhanced analysis {analysis.analysis_id}")
-        return analysis
-
-    async def get_latest(self) -> Optional[ConflictAnalysis]:
-        """Get the latest analysis from the repository."""
-        return await self._repository.get_latest()
+"""AI-powered conflict analysis service with RAG enhancement and quality improvements."""
+
+import asyncio
+import uuid
+from datetime import datetime
+from typing import Any
+
+import google.generativeai as genai
+from loguru import logger
+from tenacity import RetryError, retry, stop_after_attempt, wait_exponential
+
+from westernfront.config import Settings
+from westernfront.core.constants import RAG_QUERY_TOPICS, SEARCH_KEYWORDS
+from westernfront.core.exceptions import ServiceNotInitializedError
+from westernfront.core.models import (
+    ConflictAnalysis,
+    NewsItem,
+    RegionalImplications,
+    ReliabilityAssessment,
+)
+from westernfront.prompts.analysis import build_rag_prompt
+from westernfront.repositories.analysis import AnalysisRepository
+from westernfront.repositories.vectors import VectorRepository
+from westernfront.services.chain_analysis import ChainAnalysisService
+from westernfront.services.newsapi import NewsApiService
+from westernfront.services.parsing import ResponseParser
+from westernfront.services.reddit import RedditService
+from westernfront.services.retrieval import RetrievalService
+from westernfront.services.rss import RssService
+from westernfront.services.scheduler import AnalysisScheduler
+from westernfront.services.validation import AnalysisValidator
+from westernfront.utils import extract_json_from_response
+
+
+class AnalysisService:
+    """Service for generating AI-powered conflict analysis with RAG and quality improvements."""
+
+    def __init__(
+        self,
+        gemini_api_key: str,
+        reddit: RedditService,
+        rss: RssService,
+        newsapi: NewsApiService,
+        repository: AnalysisRepository,
+        vectors: VectorRepository,
+        settings: Settings,
+    ) -> None:
+        self._api_key = gemini_api_key
+        self._reddit = reddit
+        self._rss = rss
+        self._newsapi = newsapi
+        self._repository = repository
+        self._vectors = vectors
+        self._settings = settings
+        self._model: genai.GenerativeModel | None = None
+        self._retrieval: RetrievalService | None = None
+        self._chain: ChainAnalysisService | None = None
+        self._parser = ResponseParser()
+        self._validator = AnalysisValidator()
+        self._scheduler: AnalysisScheduler | None = None
+
+    async def initialize(self) -> None:
+        """Initialize the Gemini model and start background updates."""
+        logger.info("Initializing Gemini AI")
+        genai.configure(api_key=self._api_key)
+        self._model = genai.GenerativeModel(
+            "gemma-3-27b-it",
+            generation_config={
+                "temperature": 0.2,
+                "top_p": 0.95,
+                "top_k": 40,
+            },
+        )
+        logger.info("Gemini AI initialized")
+
+        self._retrieval = RetrievalService(self._vectors)
+        self._chain = ChainAnalysisService(self._model)
+        logger.info("Quality services initialized (retrieval, chain analysis)")
+
+        self._scheduler = AnalysisScheduler(self, self._settings.update_interval_minutes)
+        self._scheduler.start()
+
+    async def close(self) -> None:
+        """Clean up resources."""
+        if self._scheduler:
+            await self._scheduler.stop()
+        logger.info("Analysis service closed")
+
+    @property
+    def is_initialized(self) -> bool:
+        """Check if the service is initialized."""
+        return self._model is not None
+
+    @property
+    def keywords(self) -> list[str]:
+        """Get the search keywords used for analysis."""
+        return SEARCH_KEYWORDS
+
+    async def _ingest_all_news(self) -> int:
+        """Ingest all news from all sources into vector store in parallel."""
+        days = self._settings.analysis_days_back
+
+        results = await asyncio.gather(
+            self._reddit.get_all_posts(days),
+            self._rss.get_all_articles(days),
+            self._newsapi.get_related_articles(days_back=days),
+            return_exceptions=True,
+        )
+
+        all_items: list[NewsItem] = []
+        source_names = ["Reddit", "RSS", "NewsAPI"]
+
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                logger.error(f"Error fetching from {source_names[i]}: {result}")
+            else:
+                all_items.extend(result)
+
+        logger.info(f"Ingested {len(all_items)} total news items from all sources")
+
+        if not self._vectors.is_initialized:
+            logger.warning("Vector store not initialized, skipping ingestion")
+            return 0
+
+        stored = await self._vectors.add_items_async(all_items)
+        total_count = self._vectors.get_count()
+        logger.info(f"Stored {stored} new items. Total in vector store: {total_count}")
+        return stored
+
+    def _retrieve_relevant_items(self, max_items: int = 40) -> list[dict]:
+        """Retrieve relevant items with quality weighting."""
+        if self._retrieval:
+            return self._retrieval.retrieve_with_quality(
+                RAG_QUERY_TOPICS,
+                max_items=max_items,
+            )
+
+        if not self._vectors.is_initialized:
+            logger.warning("Vector store not initialized")
+            return []
+
+        return self._vectors.query_by_topics(
+            RAG_QUERY_TOPICS,
+            n_per_topic=max_items // len(RAG_QUERY_TOPICS) + 1,
+        )[:max_items]
+
+    @retry(wait=wait_exponential(min=2, max=60), stop=stop_after_attempt(3))
+    async def _call_gemini(self, prompt: str) -> dict[str, Any] | None:
+        """Call Gemini API with retry logic."""
+        if not self._model:
+            raise ServiceNotInitializedError("Gemini model not initialized")
+
+        logger.info("Calling Gemini API")
+        response = await self._model.generate_content_async(prompt)
+        result = extract_json_from_response(response.text)
+        if not result:
+            raise ValueError("Could not parse JSON from response")
+        return result
+
+    async def generate_analysis(self, use_chain: bool = True) -> ConflictAnalysis | None:
+        """Generate a new conflict analysis using RAG with quality improvements."""
+        await self._ingest_all_news()
+
+        retrieved_items = self._retrieve_relevant_items(
+            max_items=self._settings.max_posts_for_analysis
+        )
+
+        if len(retrieved_items) < self._settings.min_posts_for_analysis:
+            logger.warning(f"Insufficient data: {len(retrieved_items)} items")
+            return None
+
+        total_count = self._vectors.get_count()
+
+        try:
+            if use_chain and self._chain and len(retrieved_items) >= 10:
+                data = await self._chain.run_chain_analysis(
+                    retrieved_items,
+                    total_in_memory=total_count,
+                    use_full_chain=True,
+                )
+            else:
+                prompt = build_rag_prompt(retrieved_items, total_count)
+                data = await self._call_gemini(prompt)
+        except RetryError as e:
+            logger.error(f"AI analysis failed after retries: {e}")
+            return None
+
+        if not data:
+            return None
+
+        source_texts = [item.get("document", "") for item in retrieved_items]
+        is_valid, issues = self._validator.validate_analysis(data, source_texts)
+        if not is_valid:
+            logger.warning(f"Validation issues: {issues}")
+
+        reliability_data = data.get("reliability_assessment", {})
+        regional_data = data.get("regional_implications", {})
+
+        analysis = ConflictAnalysis(
+            analysis_id=str(uuid.uuid4()),
+            generated_at=datetime.now(),
+            latest_status=data.get("latest_status", "No status available"),
+            situation_summary=data.get("situation_summary", "No summary available"),
+            key_developments=self._parser.parse_key_developments(
+                data.get("key_developments", [])
+            ),
+            reliability_assessment=ReliabilityAssessment(
+                source_credibility=reliability_data.get("source_credibility", "Unknown"),
+                information_gaps=reliability_data.get("information_gaps", "Unknown"),
+                confidence_rating=reliability_data.get("confidence_rating", "LOW"),
+            ),
+            regional_implications=RegionalImplications(
+                security=regional_data.get("security", "No assessment"),
+                diplomatic=regional_data.get("diplomatic", "No assessment"),
+                economic=regional_data.get("economic", "No assessment"),
+            ),
+            tension_level=self._parser.parse_tension_level(
+                data.get("tension_level", "LOW")
+            ),
+            tension_rationale=data.get("tension_rationale", "No rationale"),
+            tension_score=self._parser.parse_tension_score(data.get("tension_score")),
+            tension_trend=self._parser.parse_tension_trend(
+                data.get("tension_trend", "STABLE")
+            ),
+            analysis_type=self._parser.parse_analysis_type(
+                data.get("analysis_type", "OTHER")
+            ),
+            key_entities=self._parser.parse_key_entities(data.get("key_entities")),
+            source_count=len(retrieved_items),
+            source_breakdown=self._parser.count_sources(retrieved_items),
+        )
+
+        await self._repository.save(analysis)
+        logger.info(f"Generated quality-enhanced analysis {analysis.analysis_id}")
+        return analysis
+
+    async def get_latest(self) -> ConflictAnalysis | None:
+        """Get the latest analysis from the repository."""
+        return await self._repository.get_latest()

src/westernfront/services/cache.py

@@ -1,73 +1,90 @@
-"""Thread-safe async cache service."""
+"""Thread-safe async cache service with read-write lock pattern."""
 
 import asyncio
-from typing import Any, Optional
+from typing import Any
 
 from cachetools import TTLCache
 
 
-class CacheService:
-    """Async-safe caching with TTL support."""
+class ReadWriteLock:
+    """Async read-write lock allowing concurrent reads but exclusive writes."""
 
-    def __init__(self, ttl_seconds: int = 3600, max_size: int = 100) -> None:
-        """
-        Initialize the cache service.
+    def __init__(self) -> None:
+        self._readers = 0
+        self._writer = False
+        self._condition = asyncio.Condition()
 
-        Args:
-            ttl_seconds: Time-to-live for cache entries in seconds.
-            max_size: Maximum number of entries in the cache.
-        """
-        self._cache: TTLCache[str, Any] = TTLCache(maxsize=max_size, ttl=ttl_seconds)
-        self._lock = asyncio.Lock()
+    async def acquire_read(self) -> None:
+        """Acquire read lock (allows concurrent readers)."""
+        async with self._condition:
+            while self._writer:
+                await self._condition.wait()
+            self._readers += 1
+
+    async def release_read(self) -> None:
+        """Release read lock."""
+        async with self._condition:
+            self._readers -= 1
+            if self._readers == 0:
+                self._condition.notify_all()
 
-    async def get(self, key: str) -> Optional[Any]:
-        """
-        Get a value from the cache.
+    async def acquire_write(self) -> None:
+        """Acquire exclusive write lock."""
+        async with self._condition:
+            while self._writer or self._readers > 0:
+                await self._condition.wait()
+            self._writer = True
 
-        Args:
-            key: The cache key.
+    async def release_write(self) -> None:
+        """Release write lock."""
+        async with self._condition:
+            self._writer = False
+            self._condition.notify_all()
+
+
+class CacheService:
+    """Async-safe caching with TTL support and read-write lock for concurrent reads."""
 
-        Returns:
-            The cached value or None if not found.
-        """
-        async with self._lock:
+    def __init__(self, ttl_seconds: int = 3600, max_size: int = 100) -> None:
+        self._cache: TTLCache[str, Any] = TTLCache(maxsize=max_size, ttl=ttl_seconds)
+        self._lock = ReadWriteLock()
+
+    async def get(self, key: str) -> Any | None:
+        """Get a value from the cache."""
+        await self._lock.acquire_read()
+        try:
             return self._cache.get(key)
+        finally:
+            await self._lock.release_read()
 
     async def set(self, key: str, value: Any) -> None:
-        """
-        Set a value in the cache.
-
-        Args:
-            key: The cache key.
-            value: The value to cache.
-        """
-        async with self._lock:
+        """Set a value in the cache."""
+        await self._lock.acquire_write()
+        try:
             self._cache[key] = value
+        finally:
+            await self._lock.release_write()
 
     async def delete(self, key: str) -> None:
-        """
-        Delete a value from the cache.
-
-        Args:
-            key: The cache key.
-        """
-        async with self._lock:
+        """Delete a value from the cache."""
+        await self._lock.acquire_write()
+        try:
             self._cache.pop(key, None)
+        finally:
+            await self._lock.release_write()
 
     async def clear(self) -> None:
         """Clear all entries from the cache."""
-        async with self._lock:
+        await self._lock.acquire_write()
+        try:
             self._cache.clear()
+        finally:
+            await self._lock.release_write()
 
     async def has(self, key: str) -> bool:
-        """
-        Check if a key exists in the cache.
-
-        Args:
-            key: The cache key.
-
-        Returns:
-            True if the key exists, False otherwise.
-        """
-        async with self._lock:
+        """Check if a key exists in the cache."""
+        await self._lock.acquire_read()
+        try:
             return key in self._cache
+        finally:
+            await self._lock.release_read()

src/westernfront/services/chain_analysis.py

@@ -0,0 +1,108 @@
+"""Multi-pass chain analysis for improved quality."""
+
+from typing import Any
+
+import google.generativeai as genai
+from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+from westernfront.prompts.analysis import (
+    build_extraction_prompt,
+    build_rag_prompt,
+    build_synthesis_prompt,
+)
+from westernfront.utils import extract_json_from_response
+
+
+class ChainAnalysisService:
+    """Multi-pass analysis using chain-of-thought for better quality."""
+
+    def __init__(self, model: genai.GenerativeModel) -> None:
+        self._model = model
+
+    @retry(wait=wait_exponential(min=2, max=30), stop=stop_after_attempt(3))
+    async def _call_model(self, prompt: str) -> dict[str, Any] | None:
+        """Call model with retry and JSON parsing."""
+        response = await self._model.generate_content_async(prompt)
+        return extract_json_from_response(response.text)
+
+    async def extract_facts(self, items: list[dict]) -> list[dict]:
+        """Stage 1: Extract facts from sources."""
+        logger.info("Chain Analysis Stage 1: Extracting facts")
+        prompt = build_extraction_prompt(items)
+
+        result = await self._call_model(prompt)
+        if not result:
+            logger.warning("Fact extraction failed, continuing with empty facts")
+            return []
+
+        facts = result.get("facts", [])
+        logger.info(f"Extracted {len(facts)} facts from sources")
+        return facts
+
+    async def synthesize_facts(
+        self,
+        facts: list[dict],
+        historical_context: str = "",
+    ) -> dict:
+        """Stage 2: Synthesize facts into preliminary assessment."""
+        logger.info("Chain Analysis Stage 2: Synthesizing facts")
+        prompt = build_synthesis_prompt(facts, historical_context)
+
+        result = await self._call_model(prompt)
+        if not result:
+            logger.warning("Synthesis failed, using defaults")
+            return {
+                "significant_developments": [],
+                "preliminary_tension": "MEDIUM",
+                "tension_reasoning": "Unable to synthesize",
+                "trends": [],
+                "contradictions": [],
+                "gaps": ["Synthesis stage failed"],
+            }
+
+        logger.info(f"Synthesis complete: preliminary tension = {result.get('preliminary_tension')}")
+        return result
+
+    async def generate_final_report(
+        self,
+        items: list[dict],
+        synthesis: dict,
+        total_in_memory: int = 0,
+    ) -> dict[str, Any] | None:
+        """Stage 3: Generate final analysis report."""
+        logger.info("Chain Analysis Stage 3: Generating final report")
+
+        prompt = build_rag_prompt(items, total_in_memory)
+
+        synthesis_context = f"""
+PRELIMINARY ASSESSMENT (from internal analysis):
+- Tension Level Estimate: {synthesis.get('preliminary_tension', 'UNKNOWN')}
+- Reasoning: {synthesis.get('tension_reasoning', 'N/A')}
+- Key Trends: {', '.join(synthesis.get('trends', []))}
+- Information Gaps: {', '.join(synthesis.get('gaps', []))}
+
+Consider this preliminary assessment but verify against the source data.
+"""
+        enhanced_prompt = prompt + "\n\n" + synthesis_context
+
+        result = await self._call_model(enhanced_prompt)
+        if result:
+            logger.info("Final report generated successfully")
+        return result
+
+    async def run_chain_analysis(
+        self,
+        items: list[dict],
+        total_in_memory: int = 0,
+        use_full_chain: bool = True,
+    ) -> dict[str, Any] | None:
+        """Run complete chain analysis."""
+        if not use_full_chain or len(items) < 10:
+            logger.info("Using direct analysis (chain disabled or insufficient data)")
+            prompt = build_rag_prompt(items, total_in_memory)
+            return await self._call_model(prompt)
+
+        facts = await self.extract_facts(items)
+        synthesis = await self.synthesize_facts(facts)
+        return await self.generate_final_report(items, synthesis, total_in_memory)

src/westernfront/services/embeddings.py

@@ -1,25 +1,21 @@
 """Local embedding service using sentence-transformers MiniLM model."""
 
+import asyncio
 import os
 from pathlib import Path
-from typing import Optional
 
 from loguru import logger
 
+from westernfront.core.exceptions import ServiceNotInitializedError
 
 MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
+EMBEDDING_DIMENSION = 384
 
 
 class EmbeddingService:
     """Service for generating text embeddings using local MiniLM model."""
 
-    def __init__(self, model_cache_dir: Optional[str] = None) -> None:
-        """
-        Initialize the embedding service.
-
-        Args:
-            model_cache_dir: Directory to cache the model (defaults to server/models).
-        """
+    def __init__(self, model_cache_dir: str | None = None) -> None:
         if model_cache_dir:
             self._cache_dir = Path(model_cache_dir)
         else:
@@ -32,28 +28,19 @@ class EmbeddingService:
         os.environ["TRANSFORMERS_CACHE"] = str(self._cache_dir)
 
         self._model = None
-        self._dimension = 384
+        self._dimension = EMBEDDING_DIMENSION
 
     def initialize(self) -> bool:
-        """
-        Initialize the embedding model.
-
-        Returns:
-            True if initialization was successful.
-        """
-        try:
-            logger.info(f"Loading embedding model from {self._cache_dir}")
-            from sentence_transformers import SentenceTransformer
-
-            self._model = SentenceTransformer(
-                MODEL_NAME,
-                cache_folder=str(self._cache_dir),
-            )
-            logger.info(f"Embedding model loaded: {MODEL_NAME}")
-            return True
-        except Exception as e:
-            logger.error(f"Failed to load embedding model: {e}")
-            return False
+        """Initialize the embedding model."""
+        logger.info(f"Loading embedding model from {self._cache_dir}")
+        from sentence_transformers import SentenceTransformer
+
+        self._model = SentenceTransformer(
+            MODEL_NAME,
+            cache_folder=str(self._cache_dir),
+        )
+        logger.info(f"Embedding model loaded: {MODEL_NAME}")
+        return True
 
     @property
     def is_initialized(self) -> bool:
@@ -65,34 +52,32 @@ class EmbeddingService:
         """Get the embedding dimension."""
         return self._dimension
 
-    def embed(self, text: str) -> list[float]:
-        """
-        Generate embedding for a single text.
-
-        Args:
-            text: Text to embed.
-
-        Returns:
-            List of floats representing the embedding.
-        """
+    def _embed_sync(self, text: str) -> list[float]:
+        """Synchronous embedding generation."""
         if not self._model:
-            raise RuntimeError("Embedding model not initialized")
-
+            raise ServiceNotInitializedError("Embedding model not initialized")
         embedding = self._model.encode(text, convert_to_numpy=True)
         return embedding.tolist()
 
-    def embed_batch(self, texts: list[str]) -> list[list[float]]:
-        """
-        Generate embeddings for multiple texts.
-
-        Args:
-            texts: List of texts to embed.
-
-        Returns:
-            List of embeddings.
-        """
+    def _embed_batch_sync(self, texts: list[str]) -> list[list[float]]:
+        """Synchronous batch embedding generation."""
         if not self._model:
-            raise RuntimeError("Embedding model not initialized")
-
+            raise ServiceNotInitializedError("Embedding model not initialized")
         embeddings = self._model.encode(texts, convert_to_numpy=True, show_progress_bar=False)
         return embeddings.tolist()
+
+    def embed(self, text: str) -> list[float]:
+        """Generate embedding for a single text (sync version)."""
+        return self._embed_sync(text)
+
+    def embed_batch(self, texts: list[str]) -> list[list[float]]:
+        """Generate embeddings for multiple texts (sync version)."""
+        return self._embed_batch_sync(texts)
+
+    async def embed_async(self, text: str) -> list[float]:
+        """Generate embedding for a single text without blocking the event loop."""
+        return await asyncio.to_thread(self._embed_sync, text)
+
+    async def embed_batch_async(self, texts: list[str]) -> list[list[float]]:
+        """Generate embeddings for multiple texts without blocking the event loop."""
+        return await asyncio.to_thread(self._embed_batch_sync, texts)

src/westernfront/services/http.py

@@ -0,0 +1,57 @@
+"""Shared HTTP client service with connection pooling."""
+
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+
+import aiohttp
+from loguru import logger
+
+from westernfront.core.constants import HTTP_TIMEOUT_SECONDS
+
+
+class HttpService:
+    """Shared HTTP client with connection pooling for all external requests."""
+
+    def __init__(self, timeout_seconds: int = HTTP_TIMEOUT_SECONDS) -> None:
+        self._timeout = aiohttp.ClientTimeout(total=timeout_seconds)
+        self._session: aiohttp.ClientSession | None = None
+
+    async def initialize(self) -> None:
+        """Initialize the shared HTTP session with connection pooling."""
+        if self._session is None or self._session.closed:
+            connector = aiohttp.TCPConnector(
+                limit=100,
+                limit_per_host=10,
+                ttl_dns_cache=300,
+                enable_cleanup_closed=True,
+            )
+            self._session = aiohttp.ClientSession(
+                connector=connector,
+                timeout=self._timeout,
+            )
+            logger.info("HTTP service initialized with connection pooling")
+
+    async def close(self) -> None:
+        """Close the HTTP session."""
+        if self._session and not self._session.closed:
+            await self._session.close()
+            self._session = None
+            logger.info("HTTP service closed")
+
+    @property
+    def session(self) -> aiohttp.ClientSession:
+        """Get the shared HTTP session."""
+        if self._session is None or self._session.closed:
+            raise RuntimeError("HTTP service not initialized")
+        return self._session
+
+    @asynccontextmanager
+    async def get(
+        self,
+        url: str,
+        params: dict | None = None,
+        headers: dict | None = None,
+    ) -> AsyncGenerator[aiohttp.ClientResponse, None]:
+        """Perform a GET request with automatic error handling."""
+        async with self.session.get(url, params=params, headers=headers) as response:
+            yield response

src/westernfront/services/newsapi.py

@@ -1,41 +1,32 @@
-"""NewsAPI integration service."""
+"""NewsAPI integration service with parallel query fetching."""
 
+import asyncio
 import hashlib
-from datetime import datetime, timedelta, timezone
-from typing import Optional
+from datetime import UTC, datetime, timedelta
 
 import aiohttp
 from loguru import logger
 
+from westernfront.core.constants import MAX_CONCURRENT_REQUESTS, NEWSAPI_BASE_URL, NEWSAPI_QUERIES
 from westernfront.core.enums import SourceType
 from westernfront.core.models import NewsItem
 from westernfront.services.cache import CacheService
-
-
-NEWSAPI_BASE_URL = "https://newsapi.org/v2"
-
-INDIA_PAKISTAN_QUERIES = [
-    "India Pakistan",
-    "Kashmir conflict",
-    "India Pakistan border",
-    "LOC firing",
-    "Indo-Pak",
-]
+from westernfront.services.http import HttpService
 
 
 class NewsApiService:
-    """Service for fetching news from NewsAPI.org."""
-
-    def __init__(self, api_key: Optional[str], cache: CacheService) -> None:
-        """
-        Initialize the NewsAPI service.
+    """Service for fetching news from NewsAPI.org with parallel query execution."""
 
-        Args:
-            api_key: NewsAPI API key (optional, service is disabled if None).
-            cache: Cache service for storing results.
-        """
+    def __init__(
+        self,
+        api_key: str | None,
+        cache: CacheService,
+        http: HttpService,
+    ) -> None:
         self._api_key = api_key
         self._cache = cache
+        self._http = http
+        self._semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
 
     @property
     def is_enabled(self) -> bool:
@@ -49,14 +40,14 @@ class NewsApiService:
         raw = f"newsapi:{url}:{title}"
         return hashlib.sha256(raw.encode()).hexdigest()[:16]
 
-    def _parse_date(self, date_str: Optional[str]) -> datetime:
+    def _parse_date(self, date_str: str | None) -> datetime:
         """Parse ISO date string from NewsAPI."""
         if not date_str:
-            return datetime.now(timezone.utc)
+            return datetime.now(UTC)
         try:
             return datetime.fromisoformat(date_str.replace("Z", "+00:00"))
         except ValueError:
-            return datetime.now(timezone.utc)
+            return datetime.now(UTC)
 
     async def _search_news(self, query: str, days_back: int = 2) -> list[dict]:
         """Search NewsAPI for articles matching query."""
@@ -69,22 +60,25 @@ class NewsApiService:
             logger.debug(f"Cache hit for NewsAPI query: {query}")
             return cached
 
-        from_date = (datetime.now(timezone.utc) - timedelta(days=days_back)).strftime("%Y-%m-%d")
-
-        params = {
-            "q": query,
-            "from": from_date,
-            "language": "en",
-            "sortBy": "publishedAt",
-            "pageSize": 50,
-        }
-
-        headers = {"X-Api-Key": self._api_key}
-
-        try:
-            async with aiohttp.ClientSession() as session:
-                url = f"{NEWSAPI_BASE_URL}/everything"
-                async with session.get(url, params=params, headers=headers) as response:
+        async with self._semaphore:
+            from_date = (datetime.now(UTC) - timedelta(days=days_back)).strftime("%Y-%m-%d")
+
+            params = {
+                "q": query,
+                "from": from_date,
+                "language": "en",
+                "sortBy": "publishedAt",
+                "pageSize": 50,
+            }
+
+            headers = {"X-Api-Key": self._api_key}
+
+            try:
+                async with self._http.get(
+                    f"{NEWSAPI_BASE_URL}/everything",
+                    params=params,
+                    headers=headers,
+                ) as response:
                     if response.status == 401:
                         logger.error("NewsAPI: Invalid API key")
                         return []
@@ -101,37 +95,34 @@ class NewsApiService:
                     logger.info(f"NewsAPI: Found {len(articles)} articles for '{query}'")
                     return articles
 
-        except aiohttp.ClientError as e:
-            logger.error(f"NewsAPI HTTP error: {e}")
-            return []
+            except aiohttp.ClientError as e:
+                logger.error(f"NewsAPI HTTP error: {e}")
+                return []
 
     async def get_related_articles(
         self,
-        keywords: Optional[list[str]] = None,
+        keywords: list[str] | None = None,
         days_back: int = 2,
     ) -> list[NewsItem]:
-        """
-        Get articles related to India-Pakistan from NewsAPI.
-
-        Args:
-            keywords: Optional additional keywords (defaults to built-in queries).
-            days_back: How many days back to search.
-
-        Returns:
-            List of news items.
-        """
+        """Get articles related to India-Pakistan from NewsAPI in parallel."""
         if not self.is_enabled:
             logger.debug("NewsAPI service is disabled (no API key)")
             return []
 
-        queries = keywords if keywords else INDIA_PAKISTAN_QUERIES
+        queries = keywords if keywords else NEWSAPI_QUERIES[:3]
+
+        tasks = [self._search_news(query, days_back) for query in queries]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
         seen_urls: set[str] = set()
-        results: list[NewsItem] = []
+        all_articles: list[NewsItem] = []
 
-        for query in queries[:3]:
-            articles = await self._search_news(query, days_back)
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                logger.error(f"NewsAPI query '{queries[i]}' failed: {result}")
+                continue
 
-            for article in articles:
+            for article in result:
                 url = article.get("url", "")
                 if not url or url in seen_urls:
                     continue
@@ -154,9 +145,9 @@ class NewsApiService:
                     reliability_score=0.9,
                     author=article.get("author"),
                 )
-                results.append(item)
+                all_articles.append(item)
                 seen_urls.add(url)
 
-        results.sort(key=lambda i: (-i.published_at.timestamp(), -i.reliability_score))
-        logger.info(f"Found {len(results)} articles from NewsAPI")
-        return results
+        all_articles.sort(key=lambda i: (-i.published_at.timestamp(), -i.reliability_score))
+        logger.info(f"Found {len(all_articles)} articles from NewsAPI")
+        return all_articles

src/westernfront/services/parsing.py

@@ -0,0 +1,88 @@
+"""Response parsing service for AI-generated analysis data."""
+
+from datetime import datetime
+
+from westernfront.core.enums import AnalysisType, TensionLevel, TensionTrend
+from westernfront.core.models import KeyDevelopment
+
+
+class ResponseParser:
+    """Parses and validates AI-generated analysis response data."""
+
+    def parse_tension_level(self, value: str) -> TensionLevel:
+        """Parse tension level from string."""
+        value = value.upper()
+        if "CRITICAL" in value:
+            return TensionLevel.CRITICAL
+        if "HIGH" in value:
+            return TensionLevel.HIGH
+        if "MEDIUM" in value:
+            return TensionLevel.MEDIUM
+        return TensionLevel.LOW
+
+    def parse_tension_trend(self, value: str) -> TensionTrend:
+        """Parse tension trend from string."""
+        value = value.upper()
+        if "INCREASING" in value:
+            return TensionTrend.INCREASING
+        if "DECREASING" in value:
+            return TensionTrend.DECREASING
+        return TensionTrend.STABLE
+
+    def parse_analysis_type(self, value: str) -> AnalysisType:
+        """Parse analysis type from string."""
+        value = value.upper()
+        if "MILITARY" in value:
+            return AnalysisType.MILITARY
+        if "DIPLOMATIC" in value:
+            return AnalysisType.DIPLOMATIC
+        if "INTERNAL" in value:
+            return AnalysisType.INTERNAL_SECURITY
+        if "POLITICAL" in value:
+            return AnalysisType.POLITICAL
+        return AnalysisType.OTHER
+
+    def parse_key_developments(self, data: list[dict]) -> list[KeyDevelopment]:
+        """Parse key developments from response data."""
+        developments = []
+        for item in data:
+            if not isinstance(item, dict):
+                continue
+            developments.append(
+                KeyDevelopment(
+                    title=item.get("title", "Unnamed"),
+                    description=item.get("description", "No description"),
+                    sources=item.get("sources", []),
+                    timestamp=datetime.now(),
+                )
+            )
+        return developments
+
+    def parse_tension_score(self, raw_score: int | float | str | None) -> int:
+        """Parse and clamp tension score to valid range [1, 10]."""
+        if raw_score is None:
+            return 1
+        if isinstance(raw_score, int | float):
+            return max(1, min(10, int(raw_score)))
+        if isinstance(raw_score, str) and raw_score.isdigit():
+            return max(1, min(10, int(raw_score)))
+        return 1
+
+    def parse_key_entities(self, entities: list[str] | str | None) -> list[str]:
+        """Parse key entities from response, handling string or list format."""
+        if entities is None:
+            return []
+        if isinstance(entities, str):
+            return [e.strip() for e in entities.split(",") if e.strip()]
+        if isinstance(entities, list):
+            return entities
+        return []
+
+    def count_sources(self, items: list[dict]) -> dict[str, int]:
+        """Count items by source type."""
+        counts: dict[str, int] = {}
+        for item in items:
+            meta = item.get("metadata", {})
+            key = meta.get("source_type", "unknown")
+            counts[key] = counts.get(key, 0) + 1
+        return counts

src/westernfront/services/reddit.py

@@ -1,7 +1,7 @@
-"""Reddit data collection service using AsyncPRAW."""
+"""Reddit data collection service using AsyncPRAW with full parallelization."""
 
-from datetime import datetime, timedelta, timezone
-from typing import Optional
+import asyncio
+from datetime import UTC, datetime, timedelta
 from urllib.parse import urlparse
 
 import aiohttp
@@ -10,40 +10,19 @@ import asyncprawcore
 from loguru import logger
 from tenacity import retry, stop_after_attempt, wait_exponential
 
+from westernfront.core.constants import (
+    DEFAULT_SUBREDDITS,
+    MAX_CONCURRENT_REQUESTS,
+    RELIABLE_DOMAINS,
+)
 from westernfront.core.enums import SourceType
+from westernfront.core.exceptions import ServiceNotInitializedError
 from westernfront.core.models import NewsItem, SubredditSource
 from westernfront.services.cache import CacheService
 
 
-RELIABLE_DOMAINS = frozenset([
-    "bbc.com", "reuters.com", "apnews.com", "aljazeera.com",
-    "nytimes.com", "wsj.com", "ft.com", "economist.com",
-    "thediplomat.com", "foreignpolicy.com", "foreignaffairs.com",
-    "dawn.com", "timesofindia.indiatimes.com", "ndtv.com", "geo.tv",
-])
-
-DEFAULT_SUBREDDITS = [
-    # High-quality geopolitics sources
-    SubredditSource(name="geopolitics", reliability_score=0.85),
-    SubredditSource(name="CredibleDefense", reliability_score=0.9),
-    SubredditSource(name="worldnews", reliability_score=0.8),
-    SubredditSource(name="neutralnews", reliability_score=0.8),
-    SubredditSource(name="DefenseNews", reliability_score=0.85),
-    # South Asia focused
-    SubredditSource(name="GeopoliticsIndia", reliability_score=0.75),
-    SubredditSource(name="SouthAsia", reliability_score=0.7),
-    SubredditSource(name="india", reliability_score=0.7),
-    SubredditSource(name="pakistan", reliability_score=0.7),
-    # Regional neighbors
-    SubredditSource(name="Nepal", reliability_score=0.65),
-    SubredditSource(name="bangladesh", reliability_score=0.65),
-    SubredditSource(name="srilanka", reliability_score=0.65),
-    SubredditSource(name="China", reliability_score=0.6),
-]
-
-
 class RedditService:
-    """Service for collecting posts from Reddit via AsyncPRAW."""
+    """Service for collecting posts from Reddit via AsyncPRAW with parallel fetching."""
 
     def __init__(
         self,
@@ -52,30 +31,17 @@ class RedditService:
         user_agent: str,
         cache: CacheService,
     ) -> None:
-        """
-        Initialize the Reddit service.
-
-        Args:
-            client_id: Reddit API client ID.
-            client_secret: Reddit API client secret.
-            user_agent: User agent string for API requests.
-            cache: Cache service for storing results.
-        """
         self._client_id = client_id
         self._client_secret = client_secret
         self._user_agent = user_agent
         self._cache = cache
-        self._reddit: Optional[asyncpraw.Reddit] = None
-        self._session: Optional[aiohttp.ClientSession] = None
+        self._reddit: asyncpraw.Reddit | None = None
+        self._session: aiohttp.ClientSession | None = None
         self._sources = list(DEFAULT_SUBREDDITS)
+        self._semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
 
     async def initialize(self) -> bool:
-        """
-        Initialize the Reddit API client.
-
-        Returns:
-            True if initialization was successful.
-        """
+        """Initialize the Reddit API client."""
         logger.info("Initializing Reddit service")
         self._session = aiohttp.ClientSession()
         self._reddit = asyncpraw.Reddit(
@@ -115,7 +81,7 @@ class RedditService:
         try:
             domain = urlparse(url).netloc
             return domain.replace("www.", "")
-        except Exception:
+        except (ValueError, AttributeError):
             return ""
 
     def _calculate_reliability(self, post_url: str, score: int, base_score: float) -> float:
@@ -139,9 +105,9 @@ class RedditService:
         source: SubredditSource,
         limit: int = 100,
     ) -> list[NewsItem]:
-        """Fetch posts from a subreddit with retry logic."""
+        """Fetch posts from a subreddit with retry logic and rate limiting."""
         if not self._reddit:
-            raise RuntimeError("Reddit service not initialized")
+            raise ServiceNotInitializedError("Reddit service not initialized")
 
         cache_key = f"reddit_{source.name}_{limit}"
         cached = await self._cache.get(cache_key)
@@ -149,69 +115,62 @@ class RedditService:
             logger.debug(f"Cache hit for r/{source.name}")
             return cached
 
-        logger.info(f"Fetching posts from r/{source.name}")
-        subreddit = await self._reddit.subreddit(source.name)
-        posts: list[NewsItem] = []
-
-        async for submission in subreddit.new(limit=limit):
-            content = f"{submission.title}\n{getattr(submission, 'selftext', '')}"
-            author = str(submission.author) if submission.author else "[deleted]"
-
-            post = NewsItem(
-                id=submission.id,
-                title=submission.title,
-                content=content,
-                url=submission.url,
-                source_name=f"r/{source.name}",
-                source_type=SourceType.REDDIT,
-                published_at=datetime.fromtimestamp(submission.created_utc, tz=timezone.utc),
-                reliability_score=self._calculate_reliability(
-                    submission.url, submission.score, source.reliability_score
-                ),
-                author=author,
-                score=submission.score,
-            )
-            posts.append(post)
-
-        await self._cache.set(cache_key, posts)
-        logger.info(f"Fetched {len(posts)} posts from r/{source.name}")
-        return posts
-
-    async def get_all_posts(
-        self,
-        days_back: int = 2,
-    ) -> list[NewsItem]:
-        """
-        Get ALL posts from all active subreddits without keyword filtering.
-        
-        Used for RAG ingestion where vector search determines relevance.
+        async with self._semaphore:
+            logger.info(f"Fetching posts from r/{source.name}")
+            subreddit = await self._reddit.subreddit(source.name)
+            posts: list[NewsItem] = []
+
+            async for submission in subreddit.new(limit=limit):
+                content = f"{submission.title}\n{getattr(submission, 'selftext', '')}"
+                author = str(submission.author) if submission.author else "[deleted]"
+
+                post = NewsItem(
+                    id=submission.id,
+                    title=submission.title,
+                    content=content,
+                    url=submission.url,
+                    source_name=f"r/{source.name}",
+                    source_type=SourceType.REDDIT,
+                    published_at=datetime.fromtimestamp(submission.created_utc, tz=UTC),
+                    reliability_score=self._calculate_reliability(
+                        submission.url, submission.score, source.reliability_score
+                    ),
+                    author=author,
+                    score=submission.score,
+                )
+                posts.append(post)
+
+            await self._cache.set(cache_key, posts)
+            logger.info(f"Fetched {len(posts)} posts from r/{source.name}")
+            return posts
+
+    async def get_all_posts(self, days_back: int = 2) -> list[NewsItem]:
+        """Get all posts from all active subreddits in parallel."""
+        active_sources = [s for s in self._sources if s.is_active]
+        cutoff = datetime.now(UTC) - timedelta(days=days_back)
 
-        Args:
-            days_back: How many days back to search.
+        tasks = [self._fetch_subreddit_posts(source) for source in active_sources]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
 
-        Returns:
-            List of all news items.
-        """
-        active_sources = [s for s in self._sources if s.is_active]
-        cutoff = datetime.now(timezone.utc) - timedelta(days=days_back)
         seen_ids: set[str] = set()
-        results: list[NewsItem] = []
-
-        for source in active_sources:
-            try:
-                posts = await self._fetch_subreddit_posts(source)
-                for post in posts:
-                    if post.published_at < cutoff:
-                        continue
-                    if post.id in seen_ids:
-                        continue
-                    results.append(post)
-                    seen_ids.add(post.id)
-
-            except asyncprawcore.exceptions.RequestException as e:
-                logger.error(f"Reddit API error for r/{source.name}: {e}")
-            except Exception as e:
-                logger.error(f"Error fetching r/{source.name}: {e}")
-
-        logger.info(f"Collected {len(results)} total Reddit posts for ingestion")
-        return results
+        all_posts: list[NewsItem] = []
+
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                source_name = active_sources[i].name
+                if isinstance(result, asyncprawcore.exceptions.RequestException):
+                    logger.error(f"Reddit API error for r/{source_name}: {result}")
+                else:
+                    logger.error(f"Error fetching r/{source_name}: {result}")
+                continue
+
+            for post in result:
+                if post.published_at < cutoff:
+                    continue
+                if post.id in seen_ids:
+                    continue
+                all_posts.append(post)
+                seen_ids.add(post.id)
+
+        logger.info(f"Collected {len(all_posts)} total Reddit posts for ingestion")
+        return all_posts

src/westernfront/services/retrieval.py

@@ -0,0 +1,101 @@
+"""Enhanced retrieval service with diversity and recency weighting."""
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from loguru import logger
+
+from westernfront.core.constants import RECENCY_BOOST, SOURCE_DIVERSITY_RULES
+
+if TYPE_CHECKING:
+    from westernfront.repositories.vectors import VectorRepository
+
+
+class RetrievalService:
+    """Enhanced retrieval with source diversity and temporal weighting."""
+
+    def __init__(self, vectors: "VectorRepository") -> None:
+        self._vectors = vectors
+
+    def _calculate_recency_score(self, published_at: str) -> float:
+        """Calculate recency boost based on publication time."""
+        try:
+            pub_date = datetime.fromisoformat(published_at.replace("Z", "+00:00"))
+            now = datetime.now(UTC)
+            hours_old = (now - pub_date).total_seconds() / 3600
+
+            if hours_old <= 24:
+                return RECENCY_BOOST["hours_24"]
+            if hours_old <= 48:
+                return RECENCY_BOOST["hours_48"]
+            if hours_old <= 168:
+                return RECENCY_BOOST["days_7"]
+            return RECENCY_BOOST["older"]
+        except (ValueError, TypeError):
+            return 1.0
+
+    def _apply_recency_boost(self, items: list[dict]) -> list[dict]:
+        """Apply recency boost to similarity scores."""
+        for item in items:
+            meta = item.get("metadata", {})
+            published_at = meta.get("published_at", "")
+            recency_mult = self._calculate_recency_score(published_at)
+            original_score = item.get("similarity_score", 0.5)
+            item["boosted_score"] = min(1.0, original_score * recency_mult)
+            item["recency_multiplier"] = recency_mult
+
+        items.sort(key=lambda x: x.get("boosted_score", 0), reverse=True)
+        return items
+
+    def _enforce_diversity(self, items: list[dict], max_items: int) -> list[dict]:
+        """Enforce source type diversity in results."""
+        by_source: dict[str, list[dict]] = {"reddit": [], "rss": [], "newsapi": []}
+
+        for item in items:
+            meta = item.get("metadata", {})
+            source_type = meta.get("source_type", "unknown")
+            if source_type in by_source:
+                by_source[source_type].append(item)
+
+        result = []
+        for source_type, rules in SOURCE_DIVERSITY_RULES.items():
+            min_count = int(max_items * rules["min_pct"])
+            available = by_source.get(source_type, [])
+            to_add = available[:min_count]
+            result.extend(to_add)
+            logger.debug(f"Diversity: Added {len(to_add)} from {source_type} (min: {min_count})")
+
+        seen_ids = {item["id"] for item in result}
+        remaining_items = [item for item in items if item["id"] not in seen_ids]
+
+        space_left = max_items - len(result)
+        for item in remaining_items[:space_left]:
+            meta = item.get("metadata", {})
+            source_type = meta.get("source_type", "unknown")
+            max_count = int(max_items * SOURCE_DIVERSITY_RULES.get(source_type, {}).get("max_pct", 1.0))
+            current_count = sum(1 for r in result if r.get("metadata", {}).get("source_type") == source_type)
+            if current_count < max_count:
+                result.append(item)
+
+        result.sort(key=lambda x: x.get("boosted_score", 0), reverse=True)
+        return result[:max_items]
+
+    def retrieve_with_quality(
+        self,
+        topics: list[str],
+        max_items: int = 40,
+        n_per_topic: int = 10,
+    ) -> list[dict]:
+        """Retrieve items with recency boost and source diversity."""
+        if not self._vectors.is_initialized:
+            logger.warning("Vector store not initialized")
+            return []
+
+        raw_results = self._vectors.query_by_topics(topics, n_per_topic=n_per_topic)
+        logger.info(f"Retrieved {len(raw_results)} raw items from vector store")
+
+        boosted = self._apply_recency_boost(raw_results)
+        diverse = self._enforce_diversity(boosted, max_items)
+        logger.info(f"After diversity enforcement: {len(diverse)} items")
+
+        return diverse

src/westernfront/services/rss.py

@@ -1,99 +1,29 @@
-"""RSS feed collection service."""
+"""RSS feed collection service with parallel fetching."""
 
+import asyncio
 import hashlib
-from datetime import datetime, timezone
-from typing import Optional
+from datetime import UTC, datetime, timedelta
 from email.utils import parsedate_to_datetime
 
 import aiohttp
 import feedparser
 from loguru import logger
 
+from westernfront.core.constants import DEFAULT_RSS_FEEDS, MAX_CONCURRENT_REQUESTS
 from westernfront.core.enums import SourceType
 from westernfront.core.models import NewsItem, RssFeed
 from westernfront.services.cache import CacheService
-
-
-DEFAULT_RSS_FEEDS = [
-    # Tier 1: Pakistan
-    RssFeed(
-        name="Dawn (Pakistan)",
-        url="https://www.dawn.com/feeds/home",
-        reliability_score=0.85,
-    ),
-    RssFeed(
-        name="Geo News",
-        url="https://www.geo.tv/rss/1/1",
-        reliability_score=0.8,
-    ),
-    RssFeed(
-        name="Express Tribune",
-        url="https://tribune.com.pk/feed/home",
-        reliability_score=0.75,
-    ),
-    # India
-    RssFeed(
-        name="Times of India",
-        url="https://timesofindia.indiatimes.com/rssfeeds/296589292.cms",
-        reliability_score=0.75,
-    ),
-    RssFeed(
-        name="NDTV India",
-        url="https://feeds.feedburner.com/ndtvnews-india-news",
-        reliability_score=0.8,
-    ),
-    RssFeed(
-        name="The Hindu",
-        url="https://www.thehindu.com/news/national/feeder/default.rss",
-        reliability_score=0.85,
-    ),
-    RssFeed(
-        name="Indian Express",
-        url="https://indianexpress.com/section/india/feed/",
-        reliability_score=0.85,
-    ),
-    # Tier 2: China/Nepal/Bangladesh
-    RssFeed(
-        name="South China Morning Post - Asia",
-        url="https://www.scmp.com/rss/91/feed",
-        reliability_score=0.85,
-    ),
-    RssFeed(
-        name="Kathmandu Post",
-        url="https://kathmandupost.com/rss",
-        reliability_score=0.75,
-    ),
-    RssFeed(
-        name="Dhaka Tribune",
-        url="https://www.dhakatribune.com/rss",
-        reliability_score=0.75,
-    ),
-    RssFeed(
-        name="Daily Star Bangladesh",
-        url="https://www.thedailystar.net/rss.xml",
-        reliability_score=0.75,
-    ),
-    # Tier 3: Others
-    RssFeed(
-        name="Daily Mirror Sri Lanka",
-        url="http://www.dailymirror.lk/RSS_Feeds/breaking-news",
-        reliability_score=0.7,
-    ),
-]
+from westernfront.services.http import HttpService
 
 
 class RssService:
-    """Service for collecting news from RSS feeds."""
-
-    def __init__(self, cache: CacheService) -> None:
-        """
-        Initialize the RSS service.
+    """Service for collecting news from RSS feeds with parallel fetching."""
 
-        Args:
-            cache: Cache service for storing results.
-        """
+    def __init__(self, cache: CacheService, http: HttpService) -> None:
         self._cache = cache
+        self._http = http
         self._feeds = list(DEFAULT_RSS_FEEDS)
+        self._semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
 
     @property
     def feeds(self) -> list[RssFeed]:
@@ -112,11 +42,11 @@ class RssService:
                 try:
                     parsed = parsedate_to_datetime(entry[date_field])
                     if parsed.tzinfo is None:
-                        return parsed.replace(tzinfo=timezone.utc)
+                        return parsed.replace(tzinfo=UTC)
                     return parsed
                 except (ValueError, TypeError):
                     pass
-        return datetime.now(timezone.utc)
+        return datetime.now(UTC)
 
     def _generate_id(self, entry: dict, feed_name: str) -> str:
         """Generate a unique ID for an entry."""
@@ -133,81 +63,73 @@ class RssService:
             logger.debug(f"Cache hit for RSS: {feed.name}")
             return cached
 
-        logger.info(f"Fetching RSS feed: {feed.name}")
-        items: list[NewsItem] = []
+        async with self._semaphore:
+            logger.info(f"Fetching RSS feed: {feed.name}")
+            items: list[NewsItem] = []
 
-        try:
-            async with aiohttp.ClientSession() as session:
-                async with session.get(feed.url, timeout=30) as response:
+            try:
+                async with self._http.get(feed.url) as response:
                     if response.status != 200:
                         logger.warning(f"RSS feed {feed.name} returned {response.status}")
                         return []
                     content = await response.text()
 
-            parsed = feedparser.parse(content)
-
-            for entry in parsed.entries:
-                title = entry.get("title", "").strip()
-                description = entry.get("description", "") or entry.get("summary", "")
-                link = entry.get("link", "")
-
-                if not title or not link:
-                    continue
+                parsed = feedparser.parse(content)
+
+                for entry in parsed.entries:
+                    title = entry.get("title", "").strip()
+                    description = entry.get("description", "") or entry.get("summary", "")
+                    link = entry.get("link", "")
+
+                    if not title or not link:
+                        continue
+
+                    item = NewsItem(
+                        id=self._generate_id(entry, feed.name),
+                        title=title,
+                        content=f"{title}\n{description}",
+                        url=link,
+                        source_name=feed.name,
+                        source_type=SourceType.RSS,
+                        published_at=self._parse_date(entry),
+                        reliability_score=feed.reliability_score,
+                        author=entry.get("author"),
+                    )
+                    items.append(item)
+
+                await self._cache.set(cache_key, items)
+                logger.info(f"Fetched {len(items)} items from {feed.name}")
+
+            except aiohttp.ClientError as e:
+                logger.error(f"HTTP error fetching {feed.name}: {e}")
+            except feedparser.CharacterEncodingOverride as e:
+                logger.error(f"Parse error for {feed.name}: {e}")
+
+            return items
+
+    async def get_all_articles(self, days_back: int = 2) -> list[NewsItem]:
+        """Get all articles from all active feeds in parallel."""
+        active_feeds = [f for f in self._feeds if f.is_active]
+        cutoff = datetime.now(UTC) - timedelta(days=days_back)
 
-                item = NewsItem(
-                    id=self._generate_id(entry, feed.name),
-                    title=title,
-                    content=f"{title}\n{description}",
-                    url=link,
-                    source_name=feed.name,
-                    source_type=SourceType.RSS,
-                    published_at=self._parse_date(entry),
-                    reliability_score=feed.reliability_score,
-                    author=entry.get("author"),
-                )
-                items.append(item)
-
-            await self._cache.set(cache_key, items)
-            logger.info(f"Fetched {len(items)} items from {feed.name}")
-
-        except aiohttp.ClientError as e:
-            logger.error(f"HTTP error fetching {feed.name}: {e}")
-        except Exception as e:
-            logger.error(f"Error parsing {feed.name}: {e}")
-
-        return items
-
-    async def get_all_articles(
-        self,
-        days_back: int = 2,
-    ) -> list[NewsItem]:
-        """
-        Get ALL articles from all active feeds without keyword filtering.
-        
-        Used for RAG ingestion where vector search determines relevance.
-
-        Args:
-            days_back: How many days back to search.
-
-        Returns:
-            List of all news items.
-        """
-        from datetime import timedelta
+        tasks = [self._fetch_feed(feed) for feed in active_feeds]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
 
-        active_feeds = [f for f in self._feeds if f.is_active]
-        cutoff = datetime.now(timezone.utc) - timedelta(days=days_back)
         seen_urls: set[str] = set()
-        results: list[NewsItem] = []
+        all_articles: list[NewsItem] = []
+
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                logger.error(f"Error fetching {active_feeds[i].name}: {result}")
+                continue
 
-        for feed in active_feeds:
-            items = await self._fetch_feed(feed)
-            for item in items:
+            for item in result:
                 if item.published_at < cutoff:
                     continue
                 if item.url in seen_urls:
                     continue
-                results.append(item)
+                all_articles.append(item)
                 seen_urls.add(item.url)
 
-        logger.info(f"Collected {len(results)} total RSS articles for ingestion")
-        return results
+        logger.info(f"Collected {len(all_articles)} total RSS articles for ingestion")
+        return all_articles

src/westernfront/services/scheduler.py

@@ -0,0 +1,69 @@
+"""Background task scheduler for periodic analysis updates."""
+
+import asyncio
+import contextlib
+from typing import TYPE_CHECKING, Protocol
+
+from loguru import logger
+
+if TYPE_CHECKING:
+    from westernfront.core.models import ConflictAnalysis
+
+
+class AnalysisGenerator(Protocol):
+    """Protocol for analysis generation."""
+
+    async def generate_analysis(self) -> "ConflictAnalysis | None":
+        """Generate a new conflict analysis."""
+        ...
+
+
+class AnalysisScheduler:
+    """Manages periodic background analysis updates."""
+
+    def __init__(
+        self,
+        generator: AnalysisGenerator,
+        interval_minutes: int = 60,
+    ) -> None:
+        self._generator = generator
+        self._interval_seconds = interval_minutes * 60
+        self._task: asyncio.Task[None] | None = None
+
+    def start(self) -> None:
+        """Start the periodic update task."""
+        if self._task is None:
+            self._task = asyncio.create_task(self._periodic_update())
+            logger.info(f"Scheduler started with {self._interval_seconds // 60}min interval")
+
+    async def stop(self) -> None:
+        """Stop the periodic update task."""
+        if self._task:
+            self._task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._task
+            self._task = None
+            logger.info("Scheduler stopped")
+
+    @property
+    def is_running(self) -> bool:
+        """Check if the scheduler is running."""
+        return self._task is not None and not self._task.done()
+
+    async def _periodic_update(self) -> None:
+        """Background task for periodic analysis updates."""
+        await asyncio.sleep(5)
+        await self._run_update()
+
+        while True:
+            await asyncio.sleep(self._interval_seconds)
+            await self._run_update()
+
+    async def _run_update(self) -> None:
+        """Execute a single analysis update cycle."""
+        logger.info("Starting scheduled analysis update")
+        analysis = await self._generator.generate_analysis()
+        if analysis:
+            logger.info(f"Scheduled analysis complete. Tension: {analysis.tension_level.value}")
+        else:
+            logger.warning("Scheduled analysis produced no result")

src/westernfront/services/validation.py

@@ -0,0 +1,119 @@
+"""Analysis validation service for quality assurance."""
+
+from datetime import datetime
+
+from westernfront.core.constants import TENSION_LEVEL_CRITERIA
+from westernfront.core.enums import TensionLevel
+
+
+class AnalysisValidator:
+    """Validates AI-generated analysis for quality and consistency."""
+
+    def validate_tension_consistency(
+        self,
+        level: TensionLevel,
+        score: int,
+        rationale: str,
+    ) -> tuple[bool, list[str]]:
+        """Validate tension level matches score and rationale."""
+        issues = []
+        criteria = TENSION_LEVEL_CRITERIA.get(level.value.upper(), {})
+        expected_range = criteria.get("score_range", (1, 10))
+
+        if not (expected_range[0] <= score <= expected_range[1]):
+            issues.append(
+                f"Score {score} inconsistent with {level.value} level "
+                f"(expected {expected_range[0]}-{expected_range[1]})"
+            )
+
+        level_keywords = {
+            TensionLevel.LOW: ["calm", "normal", "routine", "stable", "peaceful"],
+            TensionLevel.MEDIUM: ["elevated", "heightened", "tension", "concern", "monitoring"],
+            TensionLevel.HIGH: ["serious", "alert", "mobilization", "firing", "escalation"],
+            TensionLevel.CRITICAL: ["urgent", "imminent", "emergency", "active", "combat"],
+        }
+
+        keywords = level_keywords.get(level, [])
+        rationale_lower = rationale.lower()
+        if keywords and not any(kw in rationale_lower for kw in keywords):
+            issues.append(
+                f"Rationale may not justify {level.value} level "
+                f"(expected keywords like: {', '.join(keywords[:3])})"
+            )
+
+        return len(issues) == 0, issues
+
+    def validate_entities(
+        self,
+        entities: list[str],
+        source_texts: list[str],
+    ) -> tuple[bool, list[str]]:
+        """Validate that key entities are grounded in source texts."""
+        combined_text = " ".join(source_texts).lower()
+        ungrounded = []
+
+        for entity in entities:
+            entity_lower = entity.lower()
+            entity_words = entity_lower.split()
+            found = any(word in combined_text for word in entity_words if len(word) > 3)
+            if not found:
+                ungrounded.append(entity)
+
+        is_valid = len(ungrounded) <= 1
+        return is_valid, ungrounded
+
+    def validate_dates(
+        self,
+        developments: list[dict],
+    ) -> tuple[bool, list[str]]:
+        """Validate that development timestamps are reasonable."""
+        issues = []
+        now = datetime.now()
+
+        for dev in developments:
+            title = dev.get("title", "Unknown")
+            timestamp = dev.get("timestamp")
+            if timestamp and isinstance(timestamp, datetime) and timestamp > now:
+                issues.append(f"Future date in development: {title}")
+
+        return len(issues) == 0, issues
+
+    def validate_analysis(
+        self,
+        analysis_data: dict,
+        source_texts: list[str],
+    ) -> tuple[bool, list[str]]:
+        """Perform comprehensive validation on analysis."""
+        all_issues = []
+
+        level_str = analysis_data.get("tension_level", "LOW")
+        try:
+            level = TensionLevel(level_str)
+        except ValueError:
+            level = TensionLevel.LOW
+            all_issues.append(f"Invalid tension level: {level_str}")
+
+        score = analysis_data.get("tension_score", 1)
+        if not isinstance(score, int):
+            try:
+                score = int(score)
+            except (ValueError, TypeError):
+                score = 1
+                all_issues.append(f"Invalid tension score: {analysis_data.get('tension_score')}")
+
+        rationale = analysis_data.get("tension_rationale", "")
+        _, tension_issues = self.validate_tension_consistency(level, score, rationale)
+        all_issues.extend(tension_issues)
+
+        entities = analysis_data.get("key_entities", [])
+        if isinstance(entities, list):
+            _, ungrounded = self.validate_entities(entities, source_texts)
+            if ungrounded:
+                all_issues.append(f"Possibly ungrounded entities: {', '.join(ungrounded)}")
+
+        developments = analysis_data.get("key_developments", [])
+        if isinstance(developments, list):
+            _, date_issues = self.validate_dates(developments)
+            all_issues.extend(date_issues)
+
+        return len(all_issues) == 0, all_issues

src/westernfront/utils/__init__.py

@@ -0,0 +1,5 @@
+"""Utilities package exports."""
+
+from westernfront.utils.json_parser import extract_json_from_response
+
+__all__ = ["extract_json_from_response"]

src/westernfront/utils/json_parser.py

@@ -0,0 +1,42 @@
+"""Utility functions for parsing JSON from LLM responses."""
+
+import json
+import re
+from typing import Any
+
+from loguru import logger
+
+
+def extract_json_from_response(text: str) -> dict[str, Any] | None:
+    """
+    Extract JSON from an LLM response that may contain markdown code blocks or raw JSON.
+
+    Tries multiple strategies:
+    1. Direct JSON parsing
+    2. Extract from markdown code block
+    3. Find JSON object in text
+    """
+    # Try direct parsing first
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        pass
+
+    # Try extracting from markdown code block
+    json_match = re.search(r"```(?:json)?\n(.*?)\n```", text, re.DOTALL)
+    if json_match:
+        try:
+            return json.loads(json_match.group(1))
+        except json.JSONDecodeError:
+            pass
+
+    # Try finding JSON object in text
+    json_match = re.search(r"\{.*\}", text, re.DOTALL)
+    if json_match:
+        try:
+            return json.loads(json_match.group(0))
+        except json.JSONDecodeError:
+            pass
+
+    logger.warning(f"Failed to parse JSON from response: {text[:200]}...")
+    return None

tests/test_api.py

@@ -0,0 +1,71 @@
+"""Tests for API routes."""
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from westernfront.main import app
+
+
+@pytest.fixture
+async def client():
+    """Create test client."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        yield client
+
+
+class TestPublicEndpoints:
+    """Tests for public API endpoints."""
+
+    async def test_root_endpoint(self, client):
+        """Test root endpoint returns API info."""
+        response = await client.get("/")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["name"] == "WesternFront API"
+        assert "version" in data
+        assert data["status"] in ["ready", "initializing"]
+
+    async def test_health_endpoint(self, client):
+        """Test health endpoint returns status."""
+        response = await client.get("/health")
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] in ["healthy", "initializing"]
+        assert "version" in data
+        assert "timestamp" in data
+        assert "components" in data
+
+    async def test_health_head_endpoint(self, client):
+        """Test health HEAD request."""
+        response = await client.head("/health")
+        assert response.status_code == 200
+
+    async def test_tension_levels_endpoint(self, client):
+        """Test tension levels endpoint."""
+        response = await client.get(
+            "/tension-levels",
+            headers={"X-API-Key": "test-key"},
+        )
+        # Will return 401 if no valid key, or 200 with levels
+        assert response.status_code in [200, 401]
+
+
+class TestProtectedEndpoints:
+    """Tests for protected API endpoints."""
+
+    async def test_analysis_requires_auth(self, client):
+        """Test analysis endpoint requires API key."""
+        response = await client.get("/analysis")
+        assert response.status_code == 401
+        assert "API key" in response.json()["detail"]
+
+    async def test_sources_requires_auth(self, client):
+        """Test sources endpoint requires API key."""
+        response = await client.get("/sources")
+        assert response.status_code == 401
+
+    async def test_keywords_requires_auth(self, client):
+        """Test keywords endpoint requires API key."""
+        response = await client.get("/keywords")
+        assert response.status_code == 401

tests/test_parsing.py

@@ -0,0 +1,111 @@
+"""Tests for ResponseParser service."""
+
+import pytest
+from datetime import datetime
+
+from westernfront.core.enums import AnalysisType, TensionLevel, TensionTrend
+from westernfront.services.parsing import ResponseParser
+
+
+class TestResponseParser:
+    """Tests for ResponseParser methods."""
+
+    @pytest.fixture
+    def parser(self):
+        """Create parser instance."""
+        return ResponseParser()
+
+    def test_parse_tension_level_low(self, parser):
+        """Test parsing LOW tension level."""
+        assert parser.parse_tension_level("low") == TensionLevel.LOW
+        assert parser.parse_tension_level("LOW") == TensionLevel.LOW
+        assert parser.parse_tension_level("unknown") == TensionLevel.LOW
+
+    def test_parse_tension_level_medium(self, parser):
+        """Test parsing MEDIUM tension level."""
+        assert parser.parse_tension_level("medium") == TensionLevel.MEDIUM
+        assert parser.parse_tension_level("MEDIUM") == TensionLevel.MEDIUM
+
+    def test_parse_tension_level_high(self, parser):
+        """Test parsing HIGH tension level."""
+        assert parser.parse_tension_level("high") == TensionLevel.HIGH
+        assert parser.parse_tension_level("HIGH") == TensionLevel.HIGH
+
+    def test_parse_tension_level_critical(self, parser):
+        """Test parsing CRITICAL tension level."""
+        assert parser.parse_tension_level("critical") == TensionLevel.CRITICAL
+        assert parser.parse_tension_level("CRITICAL") == TensionLevel.CRITICAL
+
+    def test_parse_tension_trend(self, parser):
+        """Test parsing tension trends."""
+        assert parser.parse_tension_trend("increasing") == TensionTrend.INCREASING
+        assert parser.parse_tension_trend("decreasing") == TensionTrend.DECREASING
+        assert parser.parse_tension_trend("stable") == TensionTrend.STABLE
+        assert parser.parse_tension_trend("unknown") == TensionTrend.STABLE
+
+    def test_parse_analysis_type(self, parser):
+        """Test parsing analysis types."""
+        assert parser.parse_analysis_type("military") == AnalysisType.MILITARY
+        assert parser.parse_analysis_type("diplomatic") == AnalysisType.DIPLOMATIC
+        assert parser.parse_analysis_type("internal security") == AnalysisType.INTERNAL_SECURITY
+        assert parser.parse_analysis_type("political") == AnalysisType.POLITICAL
+        assert parser.parse_analysis_type("unknown") == AnalysisType.OTHER
+
+    def test_parse_tension_score_valid(self, parser):
+        """Test parsing valid tension scores."""
+        assert parser.parse_tension_score(5) == 5
+        assert parser.parse_tension_score(5.7) == 5
+        assert parser.parse_tension_score("7") == 7
+
+    def test_parse_tension_score_clamped(self, parser):
+        """Test tension scores are clamped to valid range."""
+        assert parser.parse_tension_score(0) == 1
+        assert parser.parse_tension_score(-5) == 1
+        assert parser.parse_tension_score(15) == 10
+        assert parser.parse_tension_score(None) == 1
+
+    def test_parse_key_entities_list(self, parser):
+        """Test parsing entities from list."""
+        result = parser.parse_key_entities(["India", "Pakistan", "Kashmir"])
+        assert result == ["India", "Pakistan", "Kashmir"]
+
+    def test_parse_key_entities_string(self, parser):
+        """Test parsing entities from comma-separated string."""
+        result = parser.parse_key_entities("India, Pakistan, Kashmir")
+        assert result == ["India", "Pakistan", "Kashmir"]
+
+    def test_parse_key_entities_none(self, parser):
+        """Test parsing None returns empty list."""
+        assert parser.parse_key_entities(None) == []
+
+    def test_parse_key_developments_valid(self, parser):
+        """Test parsing key developments."""
+        data = [
+            {
+                "title": "Test Event",
+                "description": "Test description",
+                "sources": ["Military Activity"],
+            }
+        ]
+        result = parser.parse_key_developments(data)
+        assert len(result) == 1
+        assert result[0].title == "Test Event"
+        assert result[0].description == "Test description"
+        assert result[0].sources == ["Military Activity"]
+
+    def test_parse_key_developments_skips_non_dict(self, parser):
+        """Test that non-dict items are skipped."""
+        data = [{"title": "Valid"}, "invalid", None]
+        result = parser.parse_key_developments(data)
+        assert len(result) == 1
+
+    def test_count_sources(self, parser):
+        """Test source counting."""
+        items = [
+            {"metadata": {"source_type": "reddit"}},
+            {"metadata": {"source_type": "reddit"}},
+            {"metadata": {"source_type": "rss"}},
+            {"metadata": {}},
+        ]
+        result = parser.count_sources(items)
+        assert result == {"reddit": 2, "rss": 1, "unknown": 1}