RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: make FLUSH_ENABLED a config value (#361) 

* fix: use cashews for DERIVER_FLUSH_KEY

* fix: make FLUSH_ENABLED a config value
This commit is contained in:
Rajat Ahuja 2026-02-03 10:53:14 -05:00 committed by GitHub
parent c50362d505
commit d3637a9d73
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 13 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.env.template
View File

@ -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=

1
config.toml.example
View File

@ -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]

27
src/cache/client.py vendored
View File

@ -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",
]

3
src/config.py
View File

@ -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:

9
src/deriver/queue_manager.py
View File

@ -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(

2
tests/bench/harness.py
View File

@ -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"],

26
tests/bench/runner_common.py
View File

@ -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")

21
tests/unified/runner.py
View File

@ -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()