diff --git a/.env.template b/.env.template index a91ede16..f4369f65 100644 --- a/.env.template +++ b/.env.template @@ -21,6 +21,8 @@ LOG_LEVEL=INFO # LANGFUSE_HOST= # LANGFUSE_PUBLIC_KEY= +# NAMESPACE="honcho" + # ============================================================================= # Database Settings (REQUIRED) # ============================================================================= @@ -144,5 +146,14 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here # ============================================================================= # Metrics (Optional) # ============================================================================= -# ENABLED=false -# NAMESPACE=honcho +# METRICS_ENABLED=false +# METRICS_NAMESPACE=honcho + +# ============================================================================= +# Cache +# ============================================================================= +# CACHE_ENABLED=false +# CACHE_URL="redis://localhost:6379/0" +# CACHE_NAMESPACE="honcho" +# CACHE_DEFAULT_TTL_SECONDS=300 +# CACHE_DEFAULT_LOCK_TTL_SECONDS=5 diff --git a/.gitignore b/.gitignore index f75448c4..17d9597c 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ api/docker-compose.yml *.db data +redis-data docker-compose.yml compose.yml diff --git a/README.md b/README.md index baac9e39..b83733f9 100644 --- a/README.md +++ b/README.md @@ -396,13 +396,18 @@ cp config.toml.example config.toml Then modify the values as needed. The TOML file is organized into sections: -- `[app]` - Application-level settings (log level, host, port, embedding settings) +- `[app]` - Application-level settings (log level, session limits, embedding settings, metrics) - `[db]` - Database connection and pool settings - `[auth]` - Authentication configuration +- `[cache]` - Redis cache configuration - `[llm]` - LLM provider API keys and general settings - `[dialectic]` - Dialectic API configuration (provider, model, search settings) - `[deriver]` - Background worker settings and theory of mind configuration +- `[peer_card]` - Peer card generation settings - `[summary]` - Session summarization settings +- `[dream]` - Dream processing configuration +- `[webhook]` - Webhook configuration +- `[metrics]` - Metrics collection settings - `[sentry]` - Error tracking and monitoring settings ### Using Environment Variables diff --git a/config.toml.example b/config.toml.example index 50f618ba..8667cb15 100644 --- a/config.toml.example +++ b/config.toml.example @@ -15,7 +15,7 @@ MAX_EMBEDDING_TOKENS = 8192 MAX_EMBEDDING_TOKENS_PER_REQUEST = 300000 # LANGFUSE_HOST = "https://api.langfuse.com" # LANGFUSE_PUBLIC_KEY = "your-public-key-here" - +NAMESPACE="honcho" # Database settings [db] @@ -111,3 +111,11 @@ MAX_WORKSPACE_LIMIT = 10 [metrics] ENABLED = false NAMESPACE = "honcho" + +# Cache settings +[cache] +ENABLED = false +URL = "redis://localhost:6379/0" +NAMESPACE="honcho" +DEFAULT_TTL_SECONDS = 300 +DEFAULT_LOCK_TTL_SECONDS = 5 diff --git a/docker-compose.yml.example b/docker-compose.yml.example index e11e93ad..123ec5c5 100644 --- a/docker-compose.yml.example +++ b/docker-compose.yml.example @@ -45,3 +45,15 @@ services: interval: 5s timeout: 5s retries: 5 + redis: + image: redis:8.2 + restart: always + ports: + - 6379:6379 + volumes: + - ./redis-data:/data + healthcheck: + test: ["CMD-SHELL", "redis-cli ping"] + interval: 5s + timeout: 5s + retries: 5 diff --git a/docs/v2/contributing/configuration.mdx b/docs/v2/contributing/configuration.mdx index da5f2d6a..1233f511 100644 --- a/docs/v2/contributing/configuration.mdx +++ b/docs/v2/contributing/configuration.mdx @@ -46,14 +46,19 @@ cp config.toml.example config.toml Then modify the values as needed. The TOML file is organized into sections: -- `[app]` - Application-level settings (log level, host, port, embedding settings) -- `[db]` - Database connection and pool settings -- `[auth]` - Authentication configuration -- `[llm]` - LLM provider API keys and general settings -- `[dialectic]` - Dialectic API configuration (provider, model, search settings) -- `[deriver]` - Background worker settings and theory of mind configuration -- `[summary]` - Session summarization settings -- `[sentry]` - Error tracking and monitoring settings +- `[app]` - Application-level settings (log level, session limits, embedding settings, Langfuse integration, local metrics collection) +- `[db]` - Database connection and pool settings (connection URI, pool size, timeouts, connection recycling) +- `[auth]` - Authentication configuration (enable/disable auth, JWT secret) +- `[cache]` - Redis cache configuration (enable/disable caching, Redis URL, TTL settings, lock configuration for cache stampede prevention) +- `[llm]` - LLM provider API keys (Anthropic, OpenAI, Gemini, Groq, OpenAI-compatible endpoints) and general LLM settings +- `[dialectic]` - Dialectic API configuration (provider, model, query generation settings, semantic search parameters, context window size) +- `[deriver]` - Background worker settings (worker count, polling intervals, queue management) and theory of mind configuration (model, tokens, observation limits) +- `[peer_card]` - Peer card generation settings (provider, model, token limits) +- `[summary]` - Session summarization settings (frequency thresholds, provider, model, token limits for short and long summaries) +- `[dream]` - Dream processing configuration (enable/disable, thresholds, idle timeouts, dream types, LLM settings) +- `[webhook]` - Webhook configuration (webhook secret, workspace limits) +- `[metrics]` - Metrics collection settings (enable/disable metrics, namespace) +- `[sentry]` - Error tracking and monitoring settings (enable/disable, DSN, environment, sample rates) ### Using Environment Variables @@ -107,28 +112,33 @@ The application will use the production connection URI while keeping the pool si ### Application Settings +Application-level settings control core behavior of the Honcho server including logging, session limits, message handling, and optional integrations. + **Basic Application Configuration:** ```bash # Logging and server settings -LOG_LEVEL=INFO # DEBUG, INFO, WARNING, ERROR -SESSION_PEERS_LIMIT=10 -GET_CONTEXT_MAX_TOKENS=100000 +LOG_LEVEL=INFO # DEBUG, INFO, WARNING, ERROR, CRITICAL -# Embedding settings (optional) -EMBED_MESSAGES=false -MAX_EMBEDDING_TOKENS=8192 -MAX_EMBEDDING_TOKENS_PER_REQUEST=300000 +# Session and context limits +SESSION_OBSERVERS_LIMIT=10 # Maximum number of observers per session +GET_CONTEXT_MAX_TOKENS=100000 # Maximum tokens for context retrieval +MAX_MESSAGE_SIZE=25000 # Maximum message size in characters + +# Embedding settings +EMBED_MESSAGES=true # Enable vector embeddings for messages +MAX_EMBEDDING_TOKENS=8192 # Maximum tokens per embedding +MAX_EMBEDDING_TOKENS_PER_REQUEST=300000 # Batch embedding limit ``` -**Environment-specific settings:** +**Optional Integrations:** ```bash -# Development -LOG_LEVEL=DEBUG -FASTAPI_HOST=127.0.0.1 +# Langfuse integration for LLM observability +LANGFUSE_HOST=https://cloud.langfuse.com +LANGFUSE_PUBLIC_KEY=your-langfuse-public-key -# Production -LOG_LEVEL=WARNING -FASTAPI_HOST=0.0.0.0 +# Local metrics collection +COLLECT_METRICS_LOCAL=false +LOCAL_METRICS_FILE=metrics.jsonl ``` ### Database Configuration @@ -196,6 +206,36 @@ AUTH_JWT_SECRET=your-super-secret-jwt-key python scripts/generate_jwt_secret.py ``` +### Cache Configuration + +Honcho supports Redis caching to improve performance by caching frequently accessed data like peers, sessions, and working representations. Caching also includes lock mechanisms to prevent cache stampede scenarios. + +**Redis Cache Settings:** +```bash +# Enable/disable Redis caching +CACHE_ENABLED=false # Set to true to enable caching + +# Redis connection +CACHE_URL=redis://localhost:6379/0?suppress=false + +# Cache namespace and TTL +CACHE_NAMESPACE=honcho # Prefix for all cache keys +CACHE_DEFAULT_TTL_SECONDS=300 # How long items stay in cache (5 minutes) + +# Lock settings for preventing cache stampede +CACHE_DEFAULT_LOCK_TTL_SECONDS=5 # Lock duration when fetching from DB on cache miss +``` + +**When to Enable Caching:** +- High-traffic production environments +- Applications with many repeated reads of the same data +- When you need to reduce database load + +**Note:** Caching requires a Redis instance. You can run Redis locally with Docker: +```bash +docker run -d -p 6379:6379 redis:latest +``` + ## LLM Provider Configuration Honcho supports multiple LLM providers for different tasks. API keys are configured in the `[llm]` section, while specific features use their own configuration sections. @@ -221,6 +261,9 @@ LLM_OPENAI_COMPATIBLE_BASE_URL=https://your-openai-compatible-endpoint.com ```bash # Default settings for all LLM calls LLM_DEFAULT_MAX_TOKENS=2500 + +# Embedding provider (used when EMBED_MESSAGES=true) +LLM_EMBEDDING_PROVIDER=openai # Options: openai, gemini ``` ### Feature-Specific Model Configuration @@ -228,67 +271,147 @@ LLM_DEFAULT_MAX_TOKENS=2500 Different features can use different providers and models: **Dialectic API:** + +The Dialectic API provides theory-of-mind informed responses by integrating long-term facts with current context. + ```bash # Main dialectic model (default: Anthropic) DIALECTIC_PROVIDER=anthropic DIALECTIC_MODEL=claude-sonnet-4-20250514 DIALECTIC_MAX_OUTPUT_TOKENS=2500 -DIALECTIC_THINKING_BUDGET_TOKENS=1024 +DIALECTIC_THINKING_BUDGET_TOKENS=1024 # Only used with Anthropic provider +DIALECTIC_CONTEXT_WINDOW_SIZE=100000 # Maximum context window tokens -# Query generation for dialectic (default: Groq) +# Query generation for dialectic searches +DIALECTIC_PERFORM_QUERY_GENERATION=false # Enable query generation for semantic search DIALECTIC_QUERY_GENERATION_PROVIDER=groq DIALECTIC_QUERY_GENERATION_MODEL=llama-3.1-8b-instant # Semantic search settings -DIALECTIC_SEMANTIC_SEARCH_TOP_K=10 -DIALECTIC_SEMANTIC_SEARCH_MAX_DISTANCE=0.85 +DIALECTIC_SEMANTIC_SEARCH_TOP_K=10 # Number of results to retrieve +DIALECTIC_SEMANTIC_SEARCH_MAX_DISTANCE=0.85 # Maximum distance for relevance ``` -**Deriver:** +**Deriver (Theory of Mind):** + +The Deriver is a background processing system that extracts facts from messages and builds theory-of-mind representations of peers. + ```bash -# Deriver model (default: Google) +# LLM settings for deriver DERIVER_PROVIDER=google -DERIVER_MODEL=gemini-2.0-flash-lite +DERIVER_MODEL=gemini-2.5-flash-lite +DERIVER_MAX_OUTPUT_TOKENS=10000 +DERIVER_THINKING_BUDGET_TOKENS=1024 # Only used with Anthropic provider +DERIVER_MAX_INPUT_TOKENS=23000 # Maximum input tokens for deriver # Worker settings -DERIVER_WORKERS=1 -DERIVER_STALE_SESSION_TIMEOUT_MINUTES=5 -DERIVER_POLLING_SLEEP_INTERVAL_SECONDS=1.0 +DERIVER_WORKERS=1 # Number of background worker processes +DERIVER_POLLING_SLEEP_INTERVAL_SECONDS=1.0 # Time between queue checks +DERIVER_STALE_SESSION_TIMEOUT_MINUTES=5 # Timeout for stale sessions -# Peer card settings -DERIVER_PEER_CARD_PROVIDER=openai -DERIVER_PEER_CARD_MODEL=gpt-5-nano-2025-08-07 -DERIVER_PEER_CARD_MAX_OUTPUT_TOKENS=2000 +# Queue management +DERIVER_QUEUE_ERROR_RETENTION_SECONDS=2592000 # Keep errored items for 30 days -# Maximum number of observations to store in working representation -# This is applied to both explicit and deductive observations -DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100 +# Working representation settings +DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=50 # Max observations stored +DERIVER_REPRESENTATION_BATCH_MAX_TOKENS=4096 # Max tokens per batch +``` + +**Peer Card:** + +Peer cards are short, structured summaries of peer identity and characteristics. + +```bash +# Enable/disable peer card generation +PEER_CARD_ENABLED=true + +# LLM settings for peer card generation +PEER_CARD_PROVIDER=openai +PEER_CARD_MODEL=gpt-5-nano-2025-08-07 +PEER_CARD_MAX_OUTPUT_TOKENS=4000 # Includes thinking tokens for GPT-5 models ``` **Summary Generation:** -```bash -# Summary model (default: Google) -SUMMARY_PROVIDER=google -SUMMARY_MODEL=gemini-1.5-flash-latest -SUMMARY_MAX_TOKENS_SHORT=1000 -SUMMARY_MAX_TOKENS_LONG=2000 -SUMMARY_THINKING_BUDGET_TOKENS=512 -# Summary frequency -SUMMARY_MESSAGES_PER_SHORT_SUMMARY=20 -SUMMARY_MESSAGES_PER_LONG_SUMMARY=60 +Session summaries provide compressed context for long conversations. Honcho creates two types: short summaries (frequent) and long summaries (comprehensive). + +```bash +# Enable/disable summarization +SUMMARY_ENABLED=true + +# LLM settings for summary generation +SUMMARY_PROVIDER=openai +SUMMARY_MODEL=gpt-4o-mini-2024-07-18 +SUMMARY_MAX_TOKENS_SHORT=1000 # Max tokens for short summaries +SUMMARY_MAX_TOKENS_LONG=4000 # Max tokens for long summaries +SUMMARY_THINKING_BUDGET_TOKENS=512 # Only used with Anthropic provider + +# Summary frequency thresholds +SUMMARY_MESSAGES_PER_SHORT_SUMMARY=20 # Create short summary every N messages +SUMMARY_MESSAGES_PER_LONG_SUMMARY=60 # Create long summary every N messages ``` ### Default Provider Usage By default, Honcho uses: -- **Anthropic** for dialectic API responses -- **Groq** for query generation -- **Google** for deriving theory of mind and summarization +- **Anthropic** (Claude) for dialectic API responses +- **Groq** for query generation (fast, cost-effective) +- **Google** (Gemini) for theory of mind derivation +- **OpenAI** (GPT) for peer cards and summarization - **OpenAI** for embeddings (if `EMBED_MESSAGES=true`) -You only need to set the API keys for the providers you plan to use. +You only need to set the API keys for the providers you plan to use. All providers are configurable per feature. +## Additional Features Configuration + +### Dream Processing + +Dream processing consolidates and refines peer representations during idle periods, similar to how human memory consolidation works during sleep. + +**Dream Settings:** +```bash +# Enable/disable dream processing +DREAM_ENABLED=true + +# Trigger thresholds +DREAM_DOCUMENT_THRESHOLD=50 # Minimum documents to trigger a dream +DREAM_IDLE_TIMEOUT_MINUTES=60 # Minutes of inactivity before dream can start +DREAM_MIN_HOURS_BETWEEN_DREAMS=8 # Minimum hours between dreams for a peer + +# Dream types to enable +DREAM_ENABLED_TYPES=["consolidate"] # Currently supported: consolidate + +# LLM settings for dream processing +DREAM_PROVIDER=openai +DREAM_MODEL=gpt-4o-mini-2024-07-18 +DREAM_MAX_OUTPUT_TOKENS=2000 +``` + +### Webhook Configuration + +Webhooks allow you to receive real-time notifications when events occur in Honcho (e.g., new messages, session updates). + +**Webhook Settings:** +```bash +# Webhook secret for signing payloads (optional but recommended) +WEBHOOK_SECRET=your-webhook-signing-secret + +# Limit on webhooks per workspace +WEBHOOK_MAX_WORKSPACE_LIMIT=10 +``` + +### Metrics Collection + +Enable metrics collection for monitoring Honcho performance and usage. + +**Metrics Settings:** +```bash +# Enable/disable metrics collection +METRICS_ENABLED=false + +# Namespace for metrics (used in metric names) +METRICS_NAMESPACE=honcho +``` ## Monitoring Configuration @@ -296,13 +419,17 @@ You only need to set the API keys for the providers you plan to use. **Sentry Settings:** ```bash -# Enable/disable Sentry +# Enable/disable Sentry error tracking SENTRY_ENABLED=false # Sentry configuration SENTRY_DSN=https://your-sentry-dsn@sentry.io/project-id -SENTRY_TRACES_SAMPLE_RATE=0.1 -SENTRY_PROFILES_SAMPLE_RATE=0.1 +SENTRY_RELEASE=2.4.0 # Optional: track which version errors come from +SENTRY_ENVIRONMENT=production # Environment name (development, staging, production) + +# Sampling rates (0.0 to 1.0) +SENTRY_TRACES_SAMPLE_RATE=0.1 # 10% of transactions tracked +SENTRY_PROFILES_SAMPLE_RATE=0.1 # 10% of transactions profiled ``` ## Environment-Specific Examples @@ -313,7 +440,7 @@ SENTRY_PROFILES_SAMPLE_RATE=0.1 ```toml [app] LOG_LEVEL = "DEBUG" -SESSION_PEERS_LIMIT = 10 +SESSION_OBSERVERS_LIMIT = 10 EMBED_MESSAGES = false [db] @@ -323,21 +450,40 @@ POOL_SIZE = 5 [auth] USE_AUTH = false +[cache] +ENABLED = false + [dialectic] PROVIDER = "anthropic" MODEL = "claude-sonnet-4-20250514" -QUERY_GENERATION_PROVIDER = "groq" -QUERY_GENERATION_MODEL = "llama-3.1-8b-instant" +PERFORM_QUERY_GENERATION = false MAX_OUTPUT_TOKENS = 2500 -[summary] -PROVIDER = "google" -MODEL = "gemini-1.5-flash-latest" -MAX_TOKENS_SHORT = 1000 -MAX_TOKENS_LONG = 2000 - [deriver] WORKERS = 1 +PROVIDER = "google" +MODEL = "gemini-2.5-flash-lite" + +[peer_card] +ENABLED = true +PROVIDER = "openai" +MODEL = "gpt-5-nano-2025-08-07" + +[summary] +ENABLED = true +PROVIDER = "openai" +MODEL = "gpt-4o-mini-2024-07-18" +MAX_TOKENS_SHORT = 1000 +MAX_TOKENS_LONG = 4000 + +[dream] +ENABLED = true + +[webhook] +MAX_WORKSPACE_LIMIT = 10 + +[metrics] +ENABLED = false [sentry] ENABLED = false @@ -349,7 +495,12 @@ ENABLED = false LOG_LEVEL=DEBUG DB_CONNECTION_URI=postgresql+psycopg://postgres:postgres@localhost:5432/honcho_dev AUTH_USE_AUTH=false -ANTHROPIC_API_KEY=your-dev-anthropic-key +CACHE_ENABLED=false + +# LLM Provider API Keys +LLM_ANTHROPIC_API_KEY=your-dev-anthropic-key +LLM_OPENAI_API_KEY=your-dev-openai-key +LLM_GEMINI_API_KEY=your-dev-gemini-key ``` ### Production Configuration @@ -358,7 +509,7 @@ ANTHROPIC_API_KEY=your-dev-anthropic-key ```toml [app] LOG_LEVEL = "WARNING" -SESSION_PEERS_LIMIT = 10 +SESSION_OBSERVERS_LIMIT = 10 EMBED_MESSAGES = true [db] @@ -369,27 +520,50 @@ MAX_OVERFLOW = 40 [auth] USE_AUTH = true +[cache] +ENABLED = true +URL = "redis://redis:6379/0" +DEFAULT_TTL_SECONDS = 300 + [dialectic] PROVIDER = "anthropic" MODEL = "claude-sonnet-4-20250514" -QUERY_GENERATION_PROVIDER = "groq" -QUERY_GENERATION_MODEL = "llama-3.1-8b-instant" +PERFORM_QUERY_GENERATION = false MAX_OUTPUT_TOKENS = 2500 -[summary] -PROVIDER = "google" -MODEL = "gemini-1.5-flash-latest" -MAX_TOKENS_SHORT = 1000 -MAX_TOKENS_LONG = 2000 - [deriver] WORKERS = 4 PROVIDER = "google" -MODEL = "gemini-2.0-flash-lite" +MODEL = "gemini-2.5-flash-lite" + +[peer_card] +ENABLED = true +PROVIDER = "openai" +MODEL = "gpt-5-nano-2025-08-07" + +[summary] +ENABLED = true +PROVIDER = "openai" +MODEL = "gpt-4o-mini-2024-07-18" +MAX_TOKENS_SHORT = 1000 +MAX_TOKENS_LONG = 4000 + +[dream] +ENABLED = true +PROVIDER = "openai" +MODEL = "gpt-4o-mini-2024-07-18" + +[webhook] +MAX_WORKSPACE_LIMIT = 10 + +[metrics] +ENABLED = true [sentry] ENABLED = true +ENVIRONMENT = "production" TRACES_SAMPLE_RATE = 0.1 +PROFILES_SAMPLE_RATE = 0.1 ``` **Environment variables for production:** @@ -397,11 +571,27 @@ TRACES_SAMPLE_RATE = 0.1 # .env.production LOG_LEVEL=WARNING DB_CONNECTION_URI=postgresql+psycopg://honcho_user:secure_password@prod-db:5432/honcho_prod + +# Authentication AUTH_USE_AUTH=true AUTH_JWT_SECRET=your-super-secret-jwt-key -ANTHROPIC_API_KEY=your-prod-anthropic-key -GEMINI_API_KEY=your-prod-gemini-key + +# Cache +CACHE_ENABLED=true +CACHE_URL=redis://redis:6379/0 + +# LLM Provider API Keys +LLM_ANTHROPIC_API_KEY=your-prod-anthropic-key +LLM_OPENAI_API_KEY=your-prod-openai-key +LLM_GEMINI_API_KEY=your-prod-gemini-key +LLM_GROQ_API_KEY=your-prod-groq-key + +# Webhooks +WEBHOOK_SECRET=your-webhook-signing-secret + +# Monitoring SENTRY_DSN=https://your-sentry-dsn@sentry.io/project-id +SENTRY_ENVIRONMENT=production ``` ## Migration Management diff --git a/pyproject.toml b/pyproject.toml index f39dcec5..8c0bf144 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,8 @@ dependencies = [ "typing-extensions>=4.11.0", "prometheus-client>=0.20.0", "json-repair>=0.49.0", + "redis>=6.0.0", + "cashews[redis]>=7.4.3", ] [tool.uv] dev-dependencies = [ @@ -47,6 +49,7 @@ dev-dependencies = [ "pre-commit>=4.2.0", "pytest-cov>=6.2.1", "honcho-ai", + "fakeredis>=2.32.0", ] [tool.uv.workspace] diff --git a/src/cache/__init__.py b/src/cache/__init__.py new file mode 100644 index 00000000..44e5b5fe --- /dev/null +++ b/src/cache/__init__.py @@ -0,0 +1,11 @@ +"""Cache module for Honcho.""" + +from src.cache.client import ( + close_cache, + init_cache, +) + +__all__ = [ + "close_cache", + "init_cache", +] diff --git a/src/cache/client.py b/src/cache/client.py new file mode 100644 index 00000000..8f06ca46 --- /dev/null +++ b/src/cache/client.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +import asyncio +import logging +from typing import cast + +import sentry_sdk +from cashews import cache +from cashews.picklers import PicklerType +from redis import exceptions as redis_exc +from tenacity import ( + AsyncRetrying, + retry_if_exception_type, + stop_after_delay, + wait_exponential_jitter, +) + +from src.config import settings + +logger = logging.getLogger(__name__) + + +_cache_lock = asyncio.Lock() + + +def is_cache_enabled() -> bool: + return settings.CACHE.ENABLED + + +def get_cache_namespace() -> str: + # CACHE.NAMESPACE is guaranteed to be non-None by AppSettings.propagate_namespace validator + return cast(str, settings.CACHE.NAMESPACE) + + +async def init_cache() -> None: + """Initialize and verify cache connection if enabled.""" + async with _cache_lock: + # Close existing backends to force recreation with new ContextVars + await cache.close() + + if not is_cache_enabled(): + # Use in-memory cache when caching is disabled + logger.info("Cache disabled, using in-memory cache") + cache.setup("mem://", pickle_type=PicklerType.SQLALCHEMY) # pyright: ignore[reportUnknownMemberType] + return + + # Setup cache with Redis backend + try: + cache.setup( # pyright: ignore[reportUnknownMemberType] + settings.CACHE.URL, + pickle_type=PicklerType.SQLALCHEMY, + ) + + except Exception as setup_err: + logger.warning( + "Cache setup failed for %s: %s. Falling back to in-memory cache", + settings.CACHE.URL, + setup_err, + ) + if settings.SENTRY.ENABLED: + sentry_sdk.capture_exception(setup_err) + # Fallback to in-memory cache + cache.setup("mem://", pickle_type=PicklerType.SQLALCHEMY) # pyright: ignore[reportUnknownMemberType] + return + + cache.enable() + # Retry Redis ping with exponential backoff + try: + async for attempt in AsyncRetrying( + wait=wait_exponential_jitter(initial=0.2, max=2.0), + stop=stop_after_delay(5), # give it a bit more headroom + retry=retry_if_exception_type( + ( + redis_exc.TimeoutError, + redis_exc.ConnectionError, + asyncio.TimeoutError, + TimeoutError, + ) + ), + reraise=True, + ): + with attempt: + async with asyncio.timeout(2): + await cache.ping() + logger.info("Connected to cache at %s", settings.CACHE.URL) + except ( + redis_exc.TimeoutError, + redis_exc.ConnectionError, + asyncio.TimeoutError, + TimeoutError, + ) as e: + logger.warning( + "Failed to connect to cache at %s: %s. Falling back to in-memory cache", + settings.CACHE.URL, + e, + ) + if settings.SENTRY.ENABLED: + sentry_sdk.capture_exception(e) + # Fallback to in-memory cache + await cache.close() + cache.setup("mem://", pickle_type=PicklerType.SQLALCHEMY) # pyright: ignore[reportUnknownMemberType] + except Exception as e: + logger.warning( + "Unexpected cache error at %s: %s. Falling back to in-memory cache", + settings.CACHE.URL, + e, + ) + if settings.SENTRY.ENABLED: + sentry_sdk.capture_exception(e) + # Fallback to in-memory cache + await cache.close() + cache.setup("mem://", pickle_type=PicklerType.SQLALCHEMY) # pyright: ignore[reportUnknownMemberType] + + +async def close_cache() -> None: + await cache.close() + + +__all__ = [ + "init_cache", + "close_cache", + "cache", +] diff --git a/src/config.py b/src/config.py index 6458717e..82e2b97a 100644 --- a/src/config.py +++ b/src/config.py @@ -50,6 +50,7 @@ class TomlConfigSettingsSource(PydanticBaseSettingsSource): "DB": "db", "AUTH": "auth", "SENTRY": "sentry", + "CACHE": "cache", "LLM": "llm", "DERIVER": "deriver", "PEER_CARD": "peer_card", @@ -290,7 +291,22 @@ class MetricsSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="METRICS_", extra="ignore") # pyright: ignore ENABLED: bool = False - NAMESPACE: str = "honcho" + NAMESPACE: str | None = None + + +class CacheSettings(HonchoSettings): + model_config = SettingsConfigDict(env_prefix="CACHE_", extra="ignore") # pyright: ignore + + ENABLED: bool = False + URL: str = "redis://localhost:6379/0?suppress=false" + NAMESPACE: str | None = None + DEFAULT_TTL_SECONDS: Annotated[int, Field(default=300, ge=1, le=86_400)] = ( + 300 # how long to keep items in cache + ) + + DEFAULT_LOCK_TTL_SECONDS: Annotated[int, Field(default=5, ge=1, le=86_400)] = ( + 5 # how long to hold a lock on a resource when fetching DB after cache miss + ) class DreamSettings(HonchoSettings): @@ -334,6 +350,8 @@ class AppSettings(HonchoSettings): COLLECT_METRICS_LOCAL: bool = False LOCAL_METRICS_FILE: str = "metrics.jsonl" + NAMESPACE: str = "honcho" # Top-level namespace for all settings, can be overridden by nested-model settings + # Nested settings models DB: DBSettings = Field(default_factory=DBSettings) AUTH: AuthSettings = Field(default_factory=AuthSettings) @@ -345,6 +363,7 @@ class AppSettings(HonchoSettings): SUMMARY: SummarySettings = Field(default_factory=SummarySettings) WEBHOOK: WebhookSettings = Field(default_factory=WebhookSettings) METRICS: MetricsSettings = Field(default_factory=MetricsSettings) + CACHE: CacheSettings = Field(default_factory=CacheSettings) DREAM: DreamSettings = Field(default_factory=DreamSettings) @field_validator("LOG_LEVEL") @@ -354,6 +373,19 @@ class AppSettings(HonchoSettings): raise ValueError(f"Invalid log level: {v}") return log_level + @model_validator(mode="after") + def propagate_namespace(self) -> "AppSettings": + """Propagate top-level NAMESPACE to nested settings if not explicitly set. + + After this validator runs, CACHE.NAMESPACE and METRICS.NAMESPACE are guaranteed + to exist. + """ + if self.CACHE.NAMESPACE is None: + self.CACHE.NAMESPACE = self.NAMESPACE + if self.METRICS.NAMESPACE is None: + self.METRICS.NAMESPACE = self.NAMESPACE + return self + # Create a single global instance of the settings settings: AppSettings = AppSettings() diff --git a/src/crud/peer.py b/src/crud/peer.py index cad234c8..acacd235 100644 --- a/src/crud/peer.py +++ b/src/crud/peer.py @@ -1,16 +1,35 @@ from logging import getLogger from typing import Any +from cashews import NOT_NONE from sqlalchemy import Select, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from src import models, schemas +from src.cache.client import cache, get_cache_namespace +from src.config import settings +from src.crud.workspace import get_or_create_workspace from src.exceptions import ConflictException, ResourceNotFoundException from src.utils.filter import apply_filter logger = getLogger(__name__) +PEER_CACHE_KEY_TEMPLATE = "workspace:{workspace_name}:peer:{peer_name}" +PEER_LOCK_PREFIX = f"{get_cache_namespace()}:lock" + + +def peer_cache_key(workspace_name: str, peer_name: str) -> str: + """Generate cache key for peer.""" + return ( + get_cache_namespace() + + ":" + + PEER_CACHE_KEY_TEMPLATE.format( + workspace_name=workspace_name, + peer_name=peer_name, + ) + ) + async def get_or_create_peers( db: AsyncSession, @@ -35,6 +54,8 @@ async def get_or_create_peers( Raises: ConflictException: If we fail to get or create the peers """ + + await get_or_create_workspace(db, schemas.WorkspaceCreate(name=workspace_name)) peer_names = [p.name for p in peers] stmt = ( select(models.Peer) @@ -75,8 +96,6 @@ async def get_or_create_peers( try: db.add_all(new_peers) await db.commit() - # Return combined list of existing and new peers - return existing_peers + new_peers except IntegrityError: await db.rollback() if _retry: @@ -85,6 +104,37 @@ async def get_or_create_peers( ) from None return await get_or_create_peers(db, workspace_name, peers, _retry=True) + # Invalidate cache for all updated/created peers - read-through pattern + for peer_obj in existing_peers + new_peers: + cache_key = peer_cache_key(workspace_name, peer_obj.name) + await cache.delete(cache_key) + + # Return combined list of existing and new peers + return existing_peers + new_peers + + +@cache( + key=PEER_CACHE_KEY_TEMPLATE, + ttl=f"{settings.CACHE.DEFAULT_TTL_SECONDS}s", + prefix=get_cache_namespace(), + condition=NOT_NONE, +) +@cache.locked( + key=PEER_CACHE_KEY_TEMPLATE, + ttl=f"{settings.CACHE.DEFAULT_LOCK_TTL_SECONDS}s", + prefix=PEER_LOCK_PREFIX, +) +async def _fetch_peer( + db: AsyncSession, + workspace_name: str, + peer_name: str, +) -> models.Peer | None: + return await db.scalar( + select(models.Peer) + .where(models.Peer.workspace_name == workspace_name) + .where(models.Peer.name == peer_name) + ) + async def get_peer( db: AsyncSession, @@ -105,21 +155,16 @@ async def get_peer( Raises: ResourceNotFoundException: If the peer does not exist """ - # Try to get the existing peer - stmt = ( - select(models.Peer) - .where(models.Peer.workspace_name == workspace_name) - .where(models.Peer.name == peer.name) - ) - result = await db.execute(stmt) - existing_peer = result.scalar_one_or_none() + existing_peer = await _fetch_peer(db, workspace_name, peer.name) + if existing_peer is None: + raise ResourceNotFoundException( + f"Peer {peer.name} not found in workspace {workspace_name}" + ) - if existing_peer is not None: - return existing_peer + # Merge cached object into session (cached objects are detached) + existing_peer = await db.merge(existing_peer, load=False) - raise ResourceNotFoundException( - f"Peer {peer.name} not found in workspace {workspace_name}" - ) + return existing_peer async def get_peers( @@ -166,7 +211,13 @@ async def update_peer( honcho_peer.configuration = peer.configuration await db.commit() - logger.debug(f"Peer {peer_name} updated successfully") + await db.refresh(honcho_peer) + + # Invalidate cache - read-through pattern + cache_key = peer_cache_key(workspace_name, honcho_peer.name) + await cache.delete(cache_key) + + logger.debug("Peer %s updated successfully", peer_name) return honcho_peer diff --git a/src/crud/peer_card.py b/src/crud/peer_card.py index 295d0de8..f04205ed 100644 --- a/src/crud/peer_card.py +++ b/src/crud/peer_card.py @@ -7,7 +7,8 @@ from sqlalchemy import update from sqlalchemy.ext.asyncio import AsyncSession from src import exceptions, models, schemas -from src.crud.peer import get_peer +from src.cache.client import cache +from src.crud.peer import get_peer, peer_cache_key logger = logging.getLogger(__name__) @@ -81,14 +82,20 @@ async def set_peer_card( } ) ) + .returning(models.Peer) ) result = await db.execute(stmt) - if result.rowcount == 0: + updated_peer = result.scalar_one_or_none() + if updated_peer is None: raise exceptions.ResourceNotFoundException( f"Peer {observer} not found in workspace {workspace_name}" ) await db.commit() + # Invalidate cache - read-through pattern + cache_key = peer_cache_key(workspace_name, observer) + await cache.delete(cache_key) + def construct_peer_card_label(*, observer: str, observed: str) -> str: if observer == observed: diff --git a/src/crud/session.py b/src/crud/session.py index cfb77660..625626e7 100644 --- a/src/crud/session.py +++ b/src/crud/session.py @@ -1,6 +1,7 @@ from logging import getLogger from typing import Any +from cashews import NOT_NONE from nanoid import generate as generate_nanoid from sqlalchemy import Select, case, cast, func, insert, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert @@ -9,6 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.types import BigInteger, Boolean from src import models, schemas +from src.cache.client import cache, get_cache_namespace from src.config import settings from src.exceptions import ( ConflictException, @@ -18,12 +20,48 @@ from src.exceptions import ( from src.utils.filter import apply_filter from .peer import get_or_create_peers, get_peer - -# Import workspace and peer functions that are needed from .workspace import get_or_create_workspace logger = getLogger(__name__) +SESSION_CACHE_KEY_TEMPLATE = "workspace:{workspace_name}:session:{session_name}" +SESSION_LOCK_PREFIX = f"{get_cache_namespace()}:lock" + + +def session_cache_key(workspace_name: str, session_name: str) -> str: + """Generate cache key for session.""" + return ( + get_cache_namespace() + + ":" + + SESSION_CACHE_KEY_TEMPLATE.format( + workspace_name=workspace_name, + session_name=session_name, + ) + ) + + +@cache( + key=SESSION_CACHE_KEY_TEMPLATE, + ttl=f"{settings.CACHE.DEFAULT_TTL_SECONDS}s", + prefix=get_cache_namespace(), + condition=NOT_NONE, +) +@cache.locked( + key=SESSION_CACHE_KEY_TEMPLATE, + ttl=f"{settings.CACHE.DEFAULT_LOCK_TTL_SECONDS}s", + prefix=SESSION_LOCK_PREFIX, +) +async def _fetch_session( + db: AsyncSession, + workspace_name: str, + session_name: str, +) -> models.Session | None: + return await db.scalar( + select(models.Session) + .where(models.Session.workspace_name == workspace_name) + .where(models.Session.name == session_name) + ) + def count_observers_in_config( peer_configs: dict[str, schemas.SessionPeerConfig], @@ -79,15 +117,14 @@ async def get_or_create_session( ConflictException: If we fail to get or create the session """ - stmt = ( - select(models.Session) - .where(models.Session.workspace_name == workspace_name) - .where(models.Session.name == session.name) - ) + if not session.name: + raise ValueError("Session name must be provided") - result = await db.execute(stmt) + honcho_session = await _fetch_session(db, workspace_name, session.name) - honcho_session = result.scalar_one_or_none() + # Merge cached object into session if it exists (cached objects are detached) + if honcho_session is not None: + honcho_session = await db.merge(honcho_session, load=False) # Check if session already exists if honcho_session is None: @@ -148,6 +185,12 @@ async def get_or_create_session( ) await db.commit() + await db.refresh(honcho_session) + + cache_key = session_cache_key(workspace_name, session.name) + await cache.set( + cache_key, honcho_session, expire=settings.CACHE.DEFAULT_TTL_SECONDS + ) return honcho_session @@ -170,22 +213,17 @@ async def get_session( Raises: ResourceNotFoundException: If the session does not exist """ - stmt = ( - select(models.Session) - .where(models.Session.workspace_name == workspace_name) - .where(models.Session.name == session_name) - ) + session = await _fetch_session(db, workspace_name, session_name) - result = await db.execute(stmt) - - honcho_session = result.scalar_one_or_none() - - if honcho_session is None: + if session is None: raise ResourceNotFoundException( f"Session {session_name} not found in workspace {workspace_name}" ) - return honcho_session + # Merge cached object into session (cached objects are detached) + session = await db.merge(session, load=False) + + return session async def update_session( @@ -220,6 +258,12 @@ async def update_session( honcho_session.configuration = session.configuration await db.commit() + await db.refresh(honcho_session) + + # Invalidate cache - read-through pattern + cache_key = session_cache_key(workspace_name, session_name) + await cache.delete(cache_key) + logger.debug("Session %s updated successfully", session_name) return honcho_session @@ -241,22 +285,16 @@ async def delete_session( Raises: ResourceNotFoundException: If the session does not exist """ - stmt = ( - select(models.Session) - .where(models.Session.workspace_name == workspace_name) - .where(models.Session.name == session_name) - ) - result = await db.execute(stmt) - honcho_session = result.scalar_one_or_none() - - if honcho_session is None: - logger.warning( - f"Session {session_name} not found in workspace {workspace_name}" - ) - raise ResourceNotFoundException("Session not found") + honcho_session = await get_session(db, session_name, workspace_name) honcho_session.is_active = False await db.commit() + await db.refresh(honcho_session) + + # Invalidate cache - read-through pattern + cache_key = session_cache_key(workspace_name, session_name) + await cache.delete(cache_key) + logger.debug("Session %s marked as inactive", session_name) return True @@ -359,7 +397,10 @@ async def clone_session( db.add(new_session_peer) await db.commit() + await db.refresh(new_session) logger.debug("Session %s cloned successfully", original_session_name) + + # Cache will be populated on next read - read-through pattern return new_session @@ -385,18 +426,7 @@ async def remove_peers_from_session( ResourceNotFoundException: If the session does not exist """ # Verify session exists - stmt = ( - select(models.Session) - .where(models.Session.workspace_name == workspace_name) - .where(models.Session.name == session_name) - ) - result = await db.execute(stmt) - session = result.scalar_one_or_none() - - if session is None: - raise ResourceNotFoundException( - f"Session {session_name} not found in workspace {workspace_name}" - ) + await get_session(db, session_name, workspace_name) # Soft delete specified session peers by setting left_at timestamp update_stmt = ( @@ -409,7 +439,7 @@ async def remove_peers_from_session( ) .values(left_at=func.now()) ) - result = await db.execute(update_stmt) + await db.execute(update_stmt) await db.commit() return True diff --git a/src/crud/workspace.py b/src/crud/workspace.py index d9af943c..c872b728 100644 --- a/src/crud/workspace.py +++ b/src/crud/workspace.py @@ -1,16 +1,51 @@ from logging import getLogger from typing import Any +from cashews import NOT_NONE from sqlalchemy import Select, delete, func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from src import models, schemas +from src.cache.client import cache, get_cache_namespace +from src.config import settings from src.exceptions import ConflictException, ResourceNotFoundException from src.utils.filter import apply_filter logger = getLogger(__name__) +WORKSPACE_CACHE_KEY_TEMPLATE = "workspace:{workspace_name}" +WORKSPACE_LOCK_PREFIX = f"{get_cache_namespace()}:lock" + + +def workspace_cache_key(workspace_name: str) -> str: + """Generate cache key for workspace.""" + return ( + get_cache_namespace() + + ":" + + WORKSPACE_CACHE_KEY_TEMPLATE.format(workspace_name=workspace_name) + ) + + +@cache( + key=WORKSPACE_CACHE_KEY_TEMPLATE, + ttl=f"{settings.CACHE.DEFAULT_TTL_SECONDS}s", + prefix=get_cache_namespace(), + condition=NOT_NONE, +) +@cache.locked( + key=WORKSPACE_CACHE_KEY_TEMPLATE, + ttl=f"{settings.CACHE.DEFAULT_LOCK_TTL_SECONDS}s", + prefix=WORKSPACE_LOCK_PREFIX, +) +async def _fetch_workspace( + db: AsyncSession, workspace_name: str +) -> models.Workspace | None: + """Fetch a workspace from the database.""" + return await db.scalar( + select(models.Workspace).where(models.Workspace.name == workspace_name) + ) + async def get_or_create_workspace( db: AsyncSession, @@ -31,14 +66,17 @@ async def get_or_create_workspace( Raises: ConflictException: If we fail to get or create the workspace """ - # Try to get the existing workspace - stmt = select(models.Workspace).where(models.Workspace.name == workspace.name) - result = await db.execute(stmt) - existing_workspace = result.scalar_one_or_none() + if not workspace.name: + raise ValueError("Workspace name must be provided") + + # Check if workspace already exists + existing_workspace = await _fetch_workspace(db, workspace.name) if existing_workspace is not None: # Workspace already exists logger.debug("Found existing workspace: %s", workspace.name) + # Merge cached object into session (cached objects are detached) + existing_workspace = await db.merge(existing_workspace, load=False) return existing_workspace # Workspace doesn't exist, create a new one @@ -50,7 +88,14 @@ async def get_or_create_workspace( try: db.add(honcho_workspace) await db.commit() + await db.refresh(honcho_workspace) + logger.debug("Workspace created successfully: %s", workspace.name) + + cache_key = workspace_cache_key(workspace.name) + await cache.set( + cache_key, honcho_workspace, expire=settings.CACHE.DEFAULT_TTL_SECONDS + ) return honcho_workspace except IntegrityError: await db.rollback() @@ -94,15 +139,15 @@ async def get_workspace( Raises: ResourceNotFoundException: If the workspace does not exist """ - # Try to get the existing peer - stmt = select(models.Workspace).where(models.Workspace.name == workspace_name) - result = await db.execute(stmt) - existing_workspace = result.scalar_one_or_none() + existing_workspace = await _fetch_workspace(db, workspace_name) - if existing_workspace is not None: - return existing_workspace + if existing_workspace is None: + raise ResourceNotFoundException(f"Workspace {workspace_name} not found") - raise ResourceNotFoundException(f"Workspace {workspace_name} not found") + # Merge cached object into session (cached objects are detached) + existing_workspace = await db.merge(existing_workspace, load=False) + + return existing_workspace async def update_workspace( @@ -134,6 +179,12 @@ async def update_workspace( honcho_workspace.configuration = workspace.configuration await db.commit() + await db.refresh(honcho_workspace) + + # Invalidate cache + cache_key = workspace_cache_key(workspace_name) + await cache.delete(cache_key) + logger.debug("Workspace with id %s updated successfully", honcho_workspace.id) return honcho_workspace @@ -241,10 +292,18 @@ async def delete_workspace(db: AsyncSession, workspace_name: str) -> schemas.Wor ) await db.delete(honcho_workspace) await db.commit() + + cache_key = workspace_cache_key(workspace_name) + workspace_pattern = f"{cache_key}*" + await cache.delete_match(workspace_pattern) + logger.debug("Workspace %s deleted", workspace_name) - except Exception as e: - logger.error("Failed to delete workspace %s: %s", workspace_name, e) + except Exception: + logger.exception( + "Failed to delete workspace %s", + workspace_name, + ) await db.rollback() - raise e + raise return workspace_snapshot diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 356ae3ba..953895a2 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -16,6 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import func from src import models, prometheus +from src.cache.client import close_cache, init_cache from src.config import settings from src.dependencies import tracked_db from src.deriver.consumer import ( @@ -766,6 +767,13 @@ class QueueManager: async def main(): logger.debug("Starting queue manager") + try: + await init_cache() + except Exception as e: + logger.warning( + "Error initializing cache in queue manager; proceeding without cache: %s", e + ) + manager = QueueManager() try: await manager.initialize() @@ -773,4 +781,5 @@ async def main(): logger.error(f"Error in main: {str(e)}") sentry_sdk.capture_exception(e) finally: + await close_cache() logger.debug("Main function exiting") diff --git a/src/main.py b/src/main.py index 0ea88272..bb354dec 100644 --- a/src/main.py +++ b/src/main.py @@ -16,6 +16,7 @@ from sentry_sdk.integrations.fastapi import FastApiIntegration from sentry_sdk.integrations.starlette import StarletteIntegration from src import prometheus +from src.cache.client import close_cache, init_cache from src.config import settings from src.db import engine, request_context from src.exceptions import HonchoException @@ -109,8 +110,18 @@ if SENTRY_ENABLED: @asynccontextmanager async def lifespan(_: FastAPI): - yield - await engine.dispose() + try: + await init_cache() + except Exception as e: + logger.warning( + "Error initializing cache in api process; proceeding without cache: %s", e + ) + + try: + yield + finally: + await close_cache() + await engine.dispose() app = FastAPI( diff --git a/src/prometheus.py b/src/prometheus.py index 85bda9ec..f19d30f3 100644 --- a/src/prometheus.py +++ b/src/prometheus.py @@ -5,6 +5,7 @@ This module defines all Prometheus metrics for all Honcho processes and exposes """ import logging +from typing import cast from prometheus_client import ( CONTENT_TYPE_LATEST, @@ -28,7 +29,8 @@ class NamespacedCounter(Counter): def labels(self, **kwargs: str) -> "NamespacedCounter": """Override labels to automatically appends namespace label""" - kwargs["namespace"] = settings.METRICS.NAMESPACE + # METRICS.NAMESPACE is guaranteed to be non-None by AppSettings.propagate_namespace validator + kwargs["namespace"] = cast(str, settings.METRICS.NAMESPACE) return super().labels(**kwargs) diff --git a/tests/bench/harness.py b/tests/bench/harness.py index 09bb7840..1e3f1a2c 100755 --- a/tests/bench/harness.py +++ b/tests/bench/harness.py @@ -11,6 +11,7 @@ This script: """ import argparse +import asyncio import os import shutil import subprocess @@ -22,6 +23,8 @@ from pathlib import Path import yaml +from src.cache.client import close_cache, init_cache + class HonchoHarness: """ @@ -65,6 +68,9 @@ class HonchoHarness: # Update the database port compose_data["services"]["database"]["ports"] = [f"{self.db_port}:5432"] + # Update the Redis port + compose_data["services"]["redis"]["ports"] = ["6379:6379"] + # Add a unique project name to avoid conflicts compose_data["name"] = f"honcho_harness_{self.db_port}" @@ -111,13 +117,15 @@ class HonchoHarness: def get_database_env_vars(self) -> dict[str, str]: """ - Get environment variables for database configuration and required API keys. + Get environment variables for database configuration, cache configuration, and required API keys. Returns: - Dictionary of environment variables for database connection and API keys + Dictionary of environment variables for database connection, cache, and API keys """ return { "DB_CONNECTION_URI": f"postgresql+psycopg://testuser:testpwd@localhost:{self.db_port}/honcho", + "CACHE_ENABLED": "true", + "CACHE_URL": "redis://localhost:6379/0", } def start_database(self) -> None: @@ -149,6 +157,99 @@ class HonchoHarness: print("Database started successfully") + def start_redis(self) -> None: + """ + Start the Redis cache server using Docker Compose. + """ + print("Starting Redis cache server on port 6379...") + + # Change to the temp directory and start the redis service + result = subprocess.run( + [ + "docker-compose", + "-f", + str(self.docker_compose_file), + "-p", + f"honcho_harness_{self.db_port}", + "up", + "-d", + "redis", + ], + cwd=self.temp_dir, + capture_output=True, + text=True, + ) + + if result.returncode != 0: + print(f"Failed to start Redis: {result.stderr}") + sys.exit(1) + + print("Redis started successfully") + + def wait_for_redis(self, timeout: int = 30) -> bool: + """ + Wait for Redis to be ready. + + Args: + timeout: Maximum time to wait in seconds + + Returns: + True if Redis is ready, False otherwise + """ + print("Waiting for Redis to be ready...") + start_time = time.time() + redis_port = 6379 + + while time.time() - start_time < timeout: + try: + import redis + + # Test Redis connection + r = redis.Redis( + host="localhost", port=redis_port, decode_responses=True + ) + r.ping() # pyright: ignore[reportUnknownMemberType] + print("Redis is ready!") + return True + except Exception: + pass + + time.sleep(1) # Check every second + + print("Redis failed to become ready within timeout") + return False + + async def init_cache(self) -> None: + """ + Initialize the Redis cache connection. + """ + try: + # Add the project root to the path so we can import Honcho modules + sys.path.insert(0, str(self.project_root)) + + # Set environment variables for cache configuration + env = self.get_database_env_vars() + for key, value in env.items(): + os.environ[key] = value + + await init_cache() + print(f"[Instance {self.instance_id}] Cache initialized successfully") + except Exception as e: + print(f"[Instance {self.instance_id}] Failed to initialize cache: {e}") + + async def close_cache(self) -> None: + """ + Close the Redis cache connection. + """ + try: + # Add the project root to the path so we can import Honcho modules + sys.path.insert(0, str(self.project_root)) + + await close_cache() + print(f"[Instance {self.instance_id}] Cache closed successfully") + except Exception as e: + print(f"[Instance {self.instance_id}] Failed to close cache: {e}") + def wait_for_database(self, timeout: int = 60) -> bool: """ Wait for the database to be ready. @@ -576,10 +677,16 @@ except Exception as e: except Exception as e: print(f"Error removing temp directory: {e}") + # Close cache + try: + asyncio.run(self.close_cache()) + except Exception as e: + print(f"Error closing cache: {e}") + # Restore .env file self.restore_env_file() - def run(self) -> None: + async def run(self) -> None: """ Run the complete Honcho harness. """ @@ -606,11 +713,22 @@ except Exception as e: # Start database self.start_database() + # Start Redis + self.start_redis() + # Wait for database to be ready if not self.wait_for_database(): print("Database failed to start. Exiting.") sys.exit(1) + # Wait for Redis to be ready + if not self.wait_for_redis(): + print("Redis failed to start. Exiting.") + sys.exit(1) + + # Initialize cache + await self.init_cache() + # Provision database self.provision_database() @@ -700,7 +818,7 @@ class HonchoHarnessPool: ) self.harnesses.append(harness) - def run(self) -> None: + async def run(self) -> None: """ Run all Honcho harnesses in the pool. """ @@ -748,6 +866,9 @@ class HonchoHarnessPool: # Start database harness.start_database() + # Start Redis + harness.start_redis() + # Wait for database to be ready if not harness.wait_for_database(): print( @@ -755,6 +876,16 @@ class HonchoHarnessPool: ) sys.exit(1) + # Wait for Redis to be ready + if not harness.wait_for_redis(): + print( + f"Redis failed to start for instance {harness.instance_id}. Exiting." + ) + sys.exit(1) + + # Initialize cache + await harness.init_cache() + # Provision database harness.provision_database() @@ -903,7 +1034,7 @@ Examples: base_api_port=args.api_port, project_root=args.project_root, ) - pool.run() + asyncio.run(pool.run()) else: harness = HonchoHarness( db_port=args.port, @@ -911,7 +1042,7 @@ Examples: project_root=args.project_root, instance_id=0, ) - harness.run() + asyncio.run(harness.run()) if __name__ == "__main__": diff --git a/tests/conftest.py b/tests/conftest.py index 4b7cb9c1..1278ce78 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,6 +6,9 @@ from unittest.mock import AsyncMock, MagicMock, patch import jwt import pytest import pytest_asyncio +from cashews.backends.interface import ControlMixin +from cashews.picklers import PicklerType +from fakeredis import FakeAsyncRedis from fastapi import Request from fastapi.responses import JSONResponse from fastapi.testclient import TestClient @@ -26,6 +29,7 @@ from sqlalchemy_utils import ( ) from src import models +from src.cache.client import cache from src.config import settings from src.db import Base from src.dependencies import get_db @@ -164,8 +168,79 @@ async def db_session(db_engine: AsyncEngine): await session.rollback() +@pytest_asyncio.fixture(scope="session") +async def fake_cache_session(): + """Set up fakeredis for caching once per test session.""" + # Store original settings + original_enabled = settings.CACHE.ENABLED + original_url = settings.CACHE.URL + + # Create a fake redis instance that persists for the session + fake_redis = FakeAsyncRedis(decode_responses=True) + + # Patch redis creation to use fakeredis + # Cashews uses redis.asyncio.from_url to create connections + def fake_redis_from_url(*_args: Any, **_kwargs: Any): + return fake_redis + + # Patch the cashews backend's _disable property to avoid ContextVar issues + # This works around cashews' ContextVar not being properly initialized in TestClient context + + original_disable_property = ControlMixin._disable # pyright: ignore[reportPrivateUsage] + + @property # type: ignore + def patched_disable_property(self): # pyright: ignore + try: + return original_disable_property.fget(self) # pyright: ignore[reportOptionalCall] + except LookupError: + # Return empty set as default if ContextVar not set in current context + return set() # pyright: ignore + + # Start patching + redis_patch = patch("redis.asyncio.from_url", fake_redis_from_url) + redis_patch.start() + ControlMixin._disable = patched_disable_property # pyright: ignore[reportPrivateUsage, reportAttributeAccessIssue] + + try: + # Enable caching and set URL for tests + settings.CACHE.ENABLED = True + settings.CACHE.URL = "redis://fake-redis:6379/0" + + # Setup cache for tests that don't use TestClient (direct CRUD tests) + # For TestClient tests, the app's lifespan handler will also call cache.setup() + # The ContextVar patch above handles any context issues + cache.setup( # pyright: ignore[reportUnknownMemberType] + "redis://fake-redis:6379/0", pickle_type=PicklerType.SQLALCHEMY, enable=True + ) + + yield fake_redis + finally: + # Stop the patches + redis_patch.stop() + ControlMixin._disable = original_disable_property # pyright: ignore[reportPrivateUsage, reportAttributeAccessIssue] + + # Restore original settings + settings.CACHE.ENABLED = original_enabled + settings.CACHE.URL = original_url + + +@pytest_asyncio.fixture(scope="function", autouse=True) +async def fake_cache(fake_cache_session: FakeAsyncRedis): + """Clear cache between tests.""" + # Clear cache before each test + await fake_cache_session.flushall() # pyright: ignore[reportUnknownMemberType] + + yield cache + + # Clear cache after each test + await fake_cache_session.flushall() # pyright: ignore[reportUnknownMemberType] + + @pytest.fixture(scope="function") -async def client(db_session: AsyncSession): +async def client( + db_session: AsyncSession, + fake_cache_session: FakeAsyncRedis, # pyright: ignore[reportUnusedParameter] +) -> AsyncGenerator[TestClient, Any]: """Create a FastAPI TestClient for the scope of a single test function""" # Register exception handlers for tests @@ -491,6 +566,7 @@ def mock_tracked_db(db_session: AsyncSession): patch("src.dependencies.tracked_db", mock_tracked_db_context), patch("src.deriver.queue_manager.tracked_db", mock_tracked_db_context), patch("src.routers.sessions.tracked_db", mock_tracked_db_context), + patch("src.routers.peers.tracked_db", mock_tracked_db_context), patch("src.crud.representation.tracked_db", mock_tracked_db_context), patch("src.routers.peers.tracked_db", mock_tracked_db_context), patch("src.dreamer.dreamer.tracked_db", mock_tracked_db_context), diff --git a/uv.lock b/uv.lock index 43347634..87614adf 100644 --- a/uv.lock +++ b/uv.lock @@ -70,6 +70,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/6f/12/e5e0282d673bb9746bacfb6e2dba8719989d3660cdb2ea79aee9a9651afb/anyio-4.10.0-py3-none-any.whl", hash = "sha256:60e474ac86736bbfd6f210f7a61218939c318f43f9972497381f1c5e930ed3d1", size = 107213, upload-time = "2025-08-04T08:54:24.882Z" }, ] +[[package]] +name = "async-timeout" +version = "5.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a5/ae/136395dfbfe00dfc94da3f3e136d0b13f394cba8f4841120e34226265780/async_timeout-5.0.1.tar.gz", hash = "sha256:d9321a7a3d5a6a5e187e824d2fa0793ce379a202935782d555d6e9d2735677d3", size = 9274, upload-time = "2024-11-06T16:41:39.6Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fe/ba/e2081de779ca30d473f21f5b30e0e737c438205440784c7dfc81efc2b029/async_timeout-5.0.1-py3-none-any.whl", hash = "sha256:39e3809566ff85354557ec2398b55e096c8364bacac9405a7a1fa429e77fe76c", size = 6233, upload-time = "2024-11-06T16:41:37.9Z" }, +] + [[package]] name = "attrs" version = "25.3.0" @@ -118,6 +127,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/72/76/20fa66124dbe6be5cafeb312ece67de6b61dd91a0247d1ea13db4ebb33c2/cachetools-5.5.2-py3-none-any.whl", hash = "sha256:d26a22bcc62eb95c3beabd9f1ee5e820d3d2704fe2967cbe350e20c8ffcd3f0a", size = 10080, upload-time = "2025-02-20T21:01:16.647Z" }, ] +[[package]] +name = "cashews" +version = "7.4.3" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/52/d9/e7b3daee814680e9a4a02835beb8c13b489a4422e902a7f39371c811ed2b/cashews-7.4.3.tar.gz", hash = "sha256:72cc1931b558e1bbe1e80395f9be6707131ffe21768f55e359d3e760c802fbb1", size = 92426, upload-time = "2025-10-11T21:08:41.761Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c5/6b/40059c0d5fe32b6434aae14b55afe6a7e6d158826582a5a44101bb5daaed/cashews-7.4.3-py3-none-any.whl", hash = "sha256:33449b26b44c36d3ab6e73f91f71ade1e29e918e85a820238f1947beaae26322", size = 79729, upload-time = "2025-10-11T21:08:40.236Z" }, +] + +[package.optional-dependencies] +redis = [ + { name = "redis" }, +] + [[package]] name = "certifi" version = "2025.8.3" @@ -464,6 +487,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/36/f4/c6e662dade71f56cd2f3735141b265c3c79293c109549c1e6933b0651ffc/exceptiongroup-1.3.0-py3-none-any.whl", hash = "sha256:4d111e6e0c13d0644cad6ddaa7ed0261a0b36971f6d23e7ec9b4b9097da78a10", size = 16674, upload-time = "2025-05-10T17:42:49.33Z" }, ] +[[package]] +name = "fakeredis" +version = "2.32.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "redis" }, + { name = "sortedcontainers" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e1/2e/94ca3f2ff35f086d7d3eeb924054e328b2ac851f0a20302d942c8d29726c/fakeredis-2.32.0.tar.gz", hash = "sha256:63d745b40eb6c8be4899cf2a53187c097ccca3afbca04fdbc5edc8b936cd1d59", size = 171097, upload-time = "2025-10-07T10:46:58.876Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/0e/1b/84ab7fd197eba5243b6625c78fbcffaa4cf6ac7dda42f95d22165f52187e/fakeredis-2.32.0-py3-none-any.whl", hash = "sha256:c9da8228de84060cfdb72c3cf4555c18c59ba7a5ae4d273f75e4822d6f01ecf8", size = 118422, upload-time = "2025-10-07T10:46:57.643Z" }, +] + [[package]] name = "fastapi" version = "0.116.1" @@ -677,6 +714,7 @@ version = "2.4.2" source = { virtual = "." } dependencies = [ { name = "alembic" }, + { name = "cashews", extra = ["redis"] }, { name = "fastapi", extra = ["standard"] }, { name = "fastapi-pagination" }, { name = "google-genai" }, @@ -695,6 +733,7 @@ dependencies = [ { name = "pydantic-settings" }, { name = "pyjwt" }, { name = "python-dotenv" }, + { name = "redis" }, { name = "rich" }, { name = "sentry-sdk", extra = ["anthropic", "fastapi", "sqlalchemy"] }, { name = "sqlalchemy" }, @@ -707,6 +746,7 @@ dependencies = [ dev = [ { name = "basedpyright" }, { name = "coverage" }, + { name = "fakeredis" }, { name = "honcho-ai" }, { name = "interrogate" }, { name = "pre-commit" }, @@ -721,6 +761,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "alembic", specifier = ">=1.14.0" }, + { name = "cashews", extras = ["redis"], specifier = ">=7.4.3" }, { name = "fastapi", extras = ["standard"], specifier = ">=0.111.0" }, { name = "fastapi-pagination", specifier = ">=0.12.24" }, { name = "google-genai", specifier = ">=1.32.0" }, @@ -739,6 +780,7 @@ requires-dist = [ { name = "pydantic-settings", specifier = ">=2.10.1" }, { name = "pyjwt", specifier = ">=2.10.0" }, { name = "python-dotenv", specifier = ">=1.0.0" }, + { name = "redis", specifier = ">=6.0.0" }, { name = "rich", specifier = ">=13.7.1" }, { name = "sentry-sdk", extras = ["anthropic", "fastapi", "sqlalchemy"], specifier = ">=2.3.1" }, { name = "sqlalchemy", specifier = ">=2.0.30" }, @@ -751,6 +793,7 @@ requires-dist = [ dev = [ { name = "basedpyright", specifier = ">=1.29.4" }, { name = "coverage", specifier = ">=7.6.0" }, + { name = "fakeredis", specifier = ">=2.32.0" }, { name = "honcho-ai", editable = "sdks/python" }, { name = "interrogate", specifier = ">=1.7.0" }, { name = "pre-commit", specifier = ">=4.2.0" }, @@ -1967,6 +2010,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fa/de/02b54f42487e3d3c6efb3f89428677074ca7bf43aae402517bc7cca949f3/PyYAML-6.0.2-cp313-cp313-win_amd64.whl", hash = "sha256:8388ee1976c416731879ac16da0aff3f63b286ffdd57cdeb95f3f2e085687563", size = 156446, upload-time = "2024-08-06T20:33:04.33Z" }, ] +[[package]] +name = "redis" +version = "6.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "async-timeout", marker = "python_full_version < '3.11.3'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/0d/d6/e8b92798a5bd67d659d51a18170e91c16ac3b59738d91894651ee255ed49/redis-6.4.0.tar.gz", hash = "sha256:b01bc7282b8444e28ec36b261df5375183bb47a07eb9c603f284e89cbc5ef010", size = 4647399, upload-time = "2025-08-07T08:10:11.441Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e8/02/89e2ed7e85db6c93dfa9e8f691c5087df4e3551ab39081a4d7c6d1f90e05/redis-6.4.0-py3-none-any.whl", hash = "sha256:f0544fa9604264e9464cdf4814e7d4830f74b165d52f2a330a760a88dd248b7f", size = 279847, upload-time = "2025-08-07T08:10:09.84Z" }, +] + [[package]] name = "regex" version = "2025.7.34" @@ -2261,6 +2316,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235, upload-time = "2024-02-25T23:20:01.196Z" }, ] +[[package]] +name = "sortedcontainers" +version = "2.4.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/e8/c4/ba2f8066cceb6f23394729afe52f3bf7adec04bf9ed2c820b39e19299111/sortedcontainers-2.4.0.tar.gz", hash = "sha256:25caa5a06cc30b6b83d11423433f65d1f9d76c4c6a0c90e3379eaa43b9bfdb88", size = 30594, upload-time = "2021-05-16T22:03:42.897Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/32/46/9cb0e58b2deb7f82b84065f37f3bffeb12413f947f9388e4cac22c4621ce/sortedcontainers-2.4.0-py2.py3-none-any.whl", hash = "sha256:a163dcaede0f1c021485e957a39245190e74249897e2ae4b2aa38595db237ee0", size = 29575, upload-time = "2021-05-16T22:03:41.177Z" }, +] + [[package]] name = "sqlalchemy" version = "2.0.42"