diff --git a/.env.template b/.env.template index 1aa0f586..c922fe54 100644 --- a/.env.template +++ b/.env.template @@ -106,6 +106,7 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here # DERIVER_MAX_INPUT_TOKENS=23000 # DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100 # DERIVER_REPRESENTATION_BATCH_MAX_TOKENS=1024 +# DERIVER_FLUSH_ENABLED=false # Bypass batch token threshold, process work immediately # DERIVER_BACKUP_PROVIDER= # DERIVER_BACKUP_MODEL= diff --git a/config.toml.example b/config.toml.example index d8de12c0..b6b407dc 100644 --- a/config.toml.example +++ b/config.toml.example @@ -86,6 +86,7 @@ LOG_OBSERVATIONS = false MAX_INPUT_TOKENS = 23000 WORKING_REPRESENTATION_MAX_OBSERVATIONS = 100 REPRESENTATION_BATCH_MAX_TOKENS = 1024 +FLUSH_ENABLED = false # Bypass batch token threshold, process work immediately # Peer card settings [peer_card] diff --git a/src/cache/client.py b/src/cache/client.py index 0b8480ca..d4005fb1 100644 --- a/src/cache/client.py +++ b/src/cache/client.py @@ -3,9 +3,7 @@ from __future__ import annotations import asyncio import logging from typing import cast -from urllib.parse import urlparse, urlunparse -import redis.asyncio as aioredis import sentry_sdk from cashews import cache from cashews.picklers import PicklerType @@ -118,33 +116,8 @@ async def close_cache() -> None: await cache.close() -# Deriver flush mode - bypasses batch token threshold when enabled -# Uses direct Redis access to avoid cashews serialization/namespace issues -DERIVER_FLUSH_KEY = "honcho:deriver:flush_mode" - - -async def is_deriver_flush_enabled() -> bool: - """Check if deriver flush mode is enabled (bypasses batch threshold).""" - if not is_cache_enabled(): - return False - try: - # Strip query parameters - redis-py doesn't support custom params like ?suppress=true - parsed = urlparse(settings.CACHE.URL) - clean_url = urlunparse(parsed._replace(query="")) - - redis_client = aioredis.from_url(clean_url) # pyright: ignore[reportUnknownMemberType] - try: - result = await redis_client.get(DERIVER_FLUSH_KEY) - return result == b"1" - finally: - await redis_client.aclose() - except Exception: - return False - - __all__ = [ "init_cache", "close_cache", "cache", - "is_deriver_flush_enabled", ] diff --git a/src/config.py b/src/config.py index 5198056f..f01c451b 100644 --- a/src/config.py +++ b/src/config.py @@ -273,6 +273,9 @@ class DeriverSettings(BackupLLMSettingsMixin, HonchoSettings): Field(default=1024, ge=128, le=16_384), ] = 1024 + # When enabled, bypasses the batch token threshold and processes work immediately + FLUSH_ENABLED: bool = False + @model_validator(mode="after") def validate_batch_tokens_vs_context_limit(self): if self.REPRESENTATION_BATCH_MAX_TOKENS > self.MAX_INPUT_TOKENS: diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index c1f48d86..856334b9 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -16,7 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import func from src import models -from src.cache.client import close_cache, init_cache, is_deriver_flush_enabled +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 ( @@ -216,7 +216,7 @@ class QueueManager: """ Get available work units that aren't being processed. For representation tasks, only returns work units with accumulated tokens - >= REPRESENTATION_BATCH_MAX_TOKENS (forced batching), unless flush mode is enabled. + >= REPRESENTATION_BATCH_MAX_TOKENS (forced batching), unless FLUSH_ENABLED is True. Returns a dict mapping work_unit_key to aqs_id. """ limit: int = max(0, self.workers - self.get_total_owned_work_units()) @@ -224,7 +224,6 @@ class QueueManager: return {} batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS - flush_enabled = await is_deriver_flush_enabled() async with tracked_db("get_available_work_units") as db: representation_prefix = "representation:" @@ -267,8 +266,8 @@ class QueueManager: ) ) - # Apply batch threshold filter unless flush mode is enabled - if not flush_enabled and batch_max_tokens > 0: + # Apply batch threshold filter (skip if FLUSH_ENABLED is True) + if not settings.DERIVER.FLUSH_ENABLED and batch_max_tokens > 0: query = query.where( or_( ~work_units_subq.c.work_unit_key.startswith( diff --git a/tests/bench/harness.py b/tests/bench/harness.py index f8fef32e..9b9c5f51 100755 --- a/tests/bench/harness.py +++ b/tests/bench/harness.py @@ -489,6 +489,8 @@ class HonchoHarness: # Create environment with instance-specific database connection env = os.environ.copy() env.update(self.get_database_env_vars()) + # Enable flush mode for tests - process messages immediately without waiting for batch threshold + env["DERIVER_FLUSH_ENABLED"] = "true" process = subprocess.Popen( [sys.executable, "-m", "src.deriver"], diff --git a/tests/bench/runner_common.py b/tests/bench/runner_common.py index 0cfcc79b..31855479 100644 --- a/tests/bench/runner_common.py +++ b/tests/bench/runner_common.py @@ -17,12 +17,10 @@ from logging import Logger from pathlib import Path from typing import Any, Generic, Literal, TypeVar -import redis.asyncio as aioredis from anthropic import AsyncAnthropic from honcho import Honcho from honcho.api_types import SessionConfiguration, SummaryConfiguration from openai import AsyncOpenAI -from redis.asyncio.client import Redis from src.telemetry.metrics_collector import MetricsCollector @@ -44,7 +42,6 @@ class RunnerConfig: batch_size: int = 10 cleanup_workspace: bool = False use_get_context: bool = False - redis_url: str = "redis://localhost:6379/0" reasoning_level: ReasoningLevel | None = None base_url: str | None = None api_key: str | None = None @@ -66,7 +63,6 @@ class RunnerConfig: batch_size=args.batch_size, cleanup_workspace=args.cleanup_workspace, use_get_context=args.use_get_context, - redis_url=args.redis_url, reasoning_level=args.reasoning_level, base_url=args.base_url, api_key=args.api_key, @@ -155,13 +151,6 @@ def add_common_arguments(parser: argparse.ArgumentParser) -> None: help="Use get_context + judge LLM instead of dialectic .chat endpoint (default: False)", ) - parser.add_argument( - "--redis-url", - type=str, - default="redis://localhost:6379/0", - help="Redis URL for flush mode signaling (default: redis://localhost:6379/0)", - ) - parser.add_argument( "--reasoning-level", type=str, @@ -521,7 +510,6 @@ class BaseRunner(ABC, Generic[ResultT]): # Wait for deriver queue print(f"[{workspace_id}] Waiting for deriver queue to empty...") await asyncio.sleep(1) # Give time for tasks to be queued - await self._flush_deriver_queue() queue_empty = await self._wait_for_queue_empty(ctx.honcho_client) if not queue_empty: @@ -621,19 +609,6 @@ class BaseRunner(ABC, Generic[ResultT]): """Get default session configuration with summaries disabled.""" return SessionConfiguration(summary=SummaryConfiguration(enabled=False)) - async def _flush_deriver_queue(self) -> None: - """Enable deriver flush mode to bypass batch token threshold.""" - default_redis_url = "redis://localhost:6379/0" - if self.config.base_url and self.config.redis_url == default_redis_url: - print("Skipping flush mode (remote instance, no --redis-url provided)") - return - redis_client: Redis = aioredis.from_url(self.config.redis_url) # pyright: ignore[reportUnknownMemberType] - try: - await redis_client.set("honcho:deriver:flush_mode", "1", ex=3600) - print(f"Enabled deriver flush mode via {self.config.redis_url}") - finally: - await redis_client.aclose() - async def _wait_for_queue_empty( self, honcho_client: Honcho, session_id: str | None = None ) -> bool: @@ -689,7 +664,6 @@ class BaseRunner(ABC, Generic[ResultT]): # Wait for dream to complete await asyncio.sleep(2) - await self._flush_deriver_queue() success = await self._wait_for_queue_empty(honcho_client) if success: print(f"[{workspace_id}] Dream for {observer} completed") diff --git a/tests/unified/runner.py b/tests/unified/runner.py index 104f804c..2168d1c4 100644 --- a/tests/unified/runner.py +++ b/tests/unified/runner.py @@ -10,7 +10,6 @@ from pathlib import Path from typing import Any, cast import httpx -import redis.asyncio as aioredis from anthropic import AsyncAnthropic from honcho.api_types import ( MessageCreateParams, @@ -182,11 +181,9 @@ class UnifiedTestExecutor: self, honcho_client: Honcho, anthropic_client: AsyncAnthropic | None, - redis_url: str, ): self.client: Honcho = honcho_client self.anthropic: AsyncAnthropic | None = anthropic_client - self.redis_url: str = redis_url async def execute(self, test_def: TestDefinition, test_name: str) -> bool: logger.info(f"Starting test: {test_name}") @@ -288,8 +285,7 @@ class UnifiedTestExecutor: if step.duration: await asyncio.sleep(step.duration) if step.target == "queue_empty": - if step.flush: - await self.flush_deriver_queue() + # Flush mode is enabled by default in the harness (DERIVER_FLUSH_ENABLED=true) await self.wait_for_queue(step.timeout) elif isinstance(step, ScheduleDreamAction): @@ -304,17 +300,6 @@ class UnifiedTestExecutor: for assertion in step.assertions: await self.check_assertion(result, assertion) - async def flush_deriver_queue(self): - """Enable deriver flush mode to bypass batch token threshold.""" - # Use direct Redis connection to set the flush key - # This avoids issues with settings being loaded before env vars are set - redis_client = aioredis.from_url(self.redis_url) # pyright: ignore[reportUnknownMemberType] - try: - await redis_client.set("honcho:deriver:flush_mode", "1", ex=60) - logger.info("Enabled deriver flush mode") - finally: - await redis_client.aclose() - async def wait_for_queue(self, timeout: int): # Poll deriver status # Wait for potential background tasks to enqueue @@ -541,7 +526,6 @@ class UnifiedTestRunner: if not self.harness.wait_for_redis(): raise RuntimeError("Redis failed to start") - await self.harness.init_cache() self.harness.provision_database() self.harness.verify_empty_database() @@ -581,9 +565,8 @@ class UnifiedTestRunner: base_url=f"http://localhost:{self.harness.api_port}", workspace_id="default", # Will be overridden per test ) - redis_url = f"redis://localhost:{self.harness.redis_port}/0" - executor = UnifiedTestExecutor(client, self.anthropic, redis_url) + executor = UnifiedTestExecutor(client, self.anthropic) suite_start_time = time.time()