From 4536612b8fc212a4a3fb85ecd4dd5c28f3d274f8 Mon Sep 17 00:00:00 2001 From: Eugene Eisenstein Date: Wed, 8 Jul 2026 11:58:13 -0400 Subject: [PATCH 1/3] add .omc to gitignore --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index e6cbf1d7..efb1f8ce 100644 --- a/.gitignore +++ b/.gitignore @@ -193,3 +193,6 @@ metrics.jsonl AGENTS.md lancedb_data/ grafana-data/ + +# Claude Code addon stuff +.omc \ No newline at end of file From be26c859addb4b9baf7506b0ff9c32c2935f3e32 Mon Sep 17 00:00:00 2001 From: Eugene Eisenstein Date: Wed, 8 Jul 2026 12:24:55 -0400 Subject: [PATCH 2/3] split config entry into 2 --- .env.template | 3 +- config.toml.example | 3 +- docs/v3/contributing/configuration.mdx | 3 +- docs/v3/contributing/troubleshooting.mdx | 2 +- src/config.py | 41 ++++++++++++++++-- src/deriver/consumer.py | 2 +- src/deriver/deriver.py | 2 +- src/deriver/queue_manager.py | 20 +++++---- src/telemetry/events/representation.py | 2 +- tests/bench/calculate_expected_events.py | 2 +- tests/deriver/test_queue_processing.py | 24 ++++++----- tests/test_config.py | 54 +++++++++++++++++++++++- 12 files changed, 126 insertions(+), 32 deletions(-) diff --git a/.env.template b/.env.template index 24014ee2..fbbd2a50 100644 --- a/.env.template +++ b/.env.template @@ -131,7 +131,8 @@ LLM_OPENAI_API_KEY=your-api-key-here # DERIVER_MAX_INPUT_TOKENS=25000 # DERIVER_MAX_CUSTOM_INSTRUCTIONS_TOKENS=2000 # DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100 -# DERIVER_REPRESENTATION_BATCH_MAX_TOKENS=1024 +# DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=512 # Min tokens a work unit accumulates before the deriver claims it; 0 disables the gate +# DERIVER_REPRESENTATION_BATCH_LLM_MAX_TOKENS=1024 # Max context-window tokens per deriver LLM call # DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS=1800 # DERIVER_FLUSH_ENABLED=false # Bypass batch token threshold, process work immediately # DERIVER_MODEL_CONFIG__FALLBACK__MODEL= diff --git a/config.toml.example b/config.toml.example index 54bf1ffd..60dfb323 100644 --- a/config.toml.example +++ b/config.toml.example @@ -109,7 +109,8 @@ LOG_OBSERVATIONS = false MAX_INPUT_TOKENS = 25000 MAX_CUSTOM_INSTRUCTIONS_TOKENS = 2000 WORKING_REPRESENTATION_MAX_OBSERVATIONS = 100 -REPRESENTATION_BATCH_MAX_TOKENS = 1024 +REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS = 512 # Min tokens a work unit accumulates before the deriver claims it; 0 disables the gate +REPRESENTATION_BATCH_LLM_MAX_TOKENS = 1024 # Max context-window tokens per deriver LLM call REPRESENTATION_BATCH_MAX_AGE_SECONDS = 1800 FLUSH_ENABLED = false # Bypass batch token threshold, process work immediately diff --git a/docs/v3/contributing/configuration.mdx b/docs/v3/contributing/configuration.mdx index e770e96b..eeda7b7a 100644 --- a/docs/v3/contributing/configuration.mdx +++ b/docs/v3/contributing/configuration.mdx @@ -408,7 +408,8 @@ DERIVER_QUEUE_ERROR_RETENTION_SECONDS=2592000 # 30 days DERIVER_DEDUPLICATE=true DERIVER_LOG_OBSERVATIONS=false DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100 -DERIVER_REPRESENTATION_BATCH_MAX_TOKENS=1024 +DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=512 +DERIVER_REPRESENTATION_BATCH_LLM_MAX_TOKENS=1024 DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS=1800 ``` diff --git a/docs/v3/contributing/troubleshooting.mdx b/docs/v3/contributing/troubleshooting.mdx index 0443eff7..7db76b34 100644 --- a/docs/v3/contributing/troubleshooting.mdx +++ b/docs/v3/contributing/troubleshooting.mdx @@ -109,7 +109,7 @@ Messages are stored but no observations, summaries, or representations are being ```bash DERIVER_WORKERS=4 ``` -5. **Representation Batch Max** — By default the deriver buffers representation work until a session has enough tokens for that representation, set via `DERIVER_REPRESENTATION_BATCH_MAX_TOKENS`. Sub-threshold tails become eligible after `DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS` (default 1800 seconds), so quiet sessions eventually flush without disabling batching globally. Set the age to `0` for legacy behavior where sub-threshold tails wait indefinitely. See [token batching](/v3/documentation/core-concepts/reasoning#token-batching) for more details +5. **Representation Batching** — By default the deriver buffers representation work until a work unit has accumulated enough tokens, set via `DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS` (`0` disables the accumulation gate). A separate setting, `DERIVER_REPRESENTATION_BATCH_LLM_MAX_TOKENS`, caps the conversation window fed to each deriver LLM call when draining a claimed work unit. Sub-threshold tails become eligible after `DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS` (default 1800 seconds), so quiet sessions eventually flush without disabling batching globally. Set the age to `0` for legacy behavior where sub-threshold tails wait indefinitely. See [token batching](/v3/documentation/core-concepts/reasoning#token-batching) for more details ## Alternative Provider Issues diff --git a/src/config.py b/src/config.py index 0e39fdcc..977b02d7 100644 --- a/src/config.py +++ b/src/config.py @@ -857,7 +857,20 @@ class DeriverSettings(HonchoSettings): int, Field(default=100, gt=0, le=1000) ] = 100 - REPRESENTATION_BATCH_MAX_TOKENS: Annotated[ + # Minimum tokens a representation work unit must accumulate (summed over + # its own unprocessed messages) before it becomes claimable. Bypassed by + # FLUSH_ENABLED and by REPRESENTATION_BATCH_MAX_AGE_SECONDS age-flushing. + # 0 disables the accumulation gate entirely (equivalent to FLUSH_ENABLED + # for claiming): work units are claimable as soon as anything is pending. + REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS: Annotated[ + int, + Field(default=512, ge=0, le=16_384), + ] = 512 + # Cumulative-token cap on the conversation window (queued messages plus + # interleaved context) fed to a single deriver LLM call when draining a + # claimed work unit. The first unprocessed message is always included, + # even if it alone exceeds the cap. + REPRESENTATION_BATCH_LLM_MAX_TOKENS: Annotated[ int, Field(default=1024, ge=128, le=16_384), ] = 1024 @@ -881,11 +894,33 @@ class DeriverSettings(HonchoSettings): ) return data # pyright: ignore[reportUnknownVariableType] + @model_validator(mode="before") + @classmethod + def _reject_removed_batch_max_tokens(cls, data: Any) -> Any: + """Fail fast on the removed REPRESENTATION_BATCH_MAX_TOKENS setting. + + The old single setting was split into + REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS (claim gate) and + REPRESENTATION_BATCH_LLM_MAX_TOKENS (per-LLM-call window cap). + `extra="ignore"` would otherwise silently drop the old key and revert + both roles to defaults — an operator-hostile failure mode for a + batching knob — so reject it loudly instead. + """ + legacy_in_data = isinstance(data, dict) and any( + str(key).upper() == "REPRESENTATION_BATCH_MAX_TOKENS" + for key in cast(dict[str, Any], data) + ) + if legacy_in_data or "DERIVER_REPRESENTATION_BATCH_MAX_TOKENS" in os.environ: + raise ValueError( + "REPRESENTATION_BATCH_MAX_TOKENS has been split into REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS (minimum tokens a work unit must accumulate before it is claimed) and REPRESENTATION_BATCH_LLM_MAX_TOKENS (token cap on the context window per deriver LLM call). Set those instead." + ) + return data # pyright: ignore[reportUnknownVariableType] + @model_validator(mode="after") def validate_batch_tokens_vs_context_limit(self): - if self.REPRESENTATION_BATCH_MAX_TOKENS > self.MAX_INPUT_TOKENS: + if self.REPRESENTATION_BATCH_LLM_MAX_TOKENS > self.MAX_INPUT_TOKENS: raise ValueError( - f"REPRESENTATION_BATCH_MAX_TOKENS ({self.REPRESENTATION_BATCH_MAX_TOKENS}) cannot exceed max deriver input tokens ({self.MAX_INPUT_TOKENS})" + f"REPRESENTATION_BATCH_LLM_MAX_TOKENS ({self.REPRESENTATION_BATCH_LLM_MAX_TOKENS}) cannot exceed max deriver input tokens ({self.MAX_INPUT_TOKENS})" ) return self diff --git a/src/deriver/consumer.py b/src/deriver/consumer.py index 6e35d614..9a961df9 100644 --- a/src/deriver/consumer.py +++ b/src/deriver/consumer.py @@ -176,7 +176,7 @@ async def process_representation_batch( queue_item_message_ids: Message IDs from queue items hit_batch_token_cap: whether the queue batcher clamped this batch to fit was_flush_enabled: snapshot of DERIVER.FLUSH_ENABLED at fetch time - batch_max_tokens: DERIVER.REPRESENTATION_BATCH_MAX_TOKENS snapshot + batch_max_tokens: DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS snapshot """ if not messages or not messages[0]: logger.debug("process_representation_batch received no messages") diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index 0db9477d..93f18eb1 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -58,7 +58,7 @@ async def process_representation_tasks_batch( queue_item_message_ids: Message IDs from queue items being processed hit_batch_token_cap: queue batcher clamped this batch to fit was_flush_enabled: DERIVER.FLUSH_ENABLED snapshot at batch time - batch_max_tokens: DERIVER.REPRESENTATION_BATCH_MAX_TOKENS snapshot + batch_max_tokens: DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS snapshot """ if not messages: return diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 93bed1a1..145a6252 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -331,16 +331,18 @@ class QueueManager: """ Get available work units that aren't being processed. For representation tasks, only returns work units whose accumulated - tokens reach REPRESENTATION_BATCH_MAX_TOKENS or whose oldest pending - item exceeds REPRESENTATION_BATCH_MAX_AGE_SECONDS, unless - FLUSH_ENABLED is True. + tokens reach REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS or whose + oldest pending item exceeds REPRESENTATION_BATCH_MAX_AGE_SECONDS, + 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()) if limit == 0: return {} - batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + work_unit_target_tokens = ( + settings.DERIVER.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS + ) async with tracked_db("get_available_work_units") as db: representation_prefix = "representation:" @@ -396,11 +398,11 @@ class QueueManager: ) # Apply batch threshold filter (skip if FLUSH_ENABLED is True) - if not settings.DERIVER.FLUSH_ENABLED and batch_max_tokens > 0: + if not settings.DERIVER.FLUSH_ENABLED and work_unit_target_tokens > 0: max_age_seconds = settings.DERIVER.REPRESENTATION_BATCH_MAX_AGE_SECONDS threshold_clause = ( func.coalesce(token_stats_subq.c.total_tokens, 0) - >= batch_max_tokens + >= work_unit_target_tokens ) if max_age_seconds > 0: threshold_clause = or_( @@ -426,13 +428,13 @@ class QueueManager: not settings.DERIVER.FLUSH_ENABLED and settings.DERIVER.REPRESENTATION_BATCH_MAX_AGE_SECONDS > 0 and work_unit_key.startswith(representation_prefix) - and int(total_tokens or 0) < batch_max_tokens + and int(total_tokens or 0) < work_unit_target_tokens ): logger.info( "age-flushing work unit %s (tokens=%s < %s, oldest=%s)", work_unit_key, total_tokens or 0, - batch_max_tokens, + work_unit_target_tokens, oldest_created_at, ) if not available_units: @@ -816,7 +818,7 @@ class QueueManager: f"{task_type} tasks are not supported for get_queue_item_batch" ) - batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS was_flush_enabled = settings.DERIVER.FLUSH_ENABLED parsed_key = parse_work_unit_key(work_unit_key) messages_context: list[models.Message] = [] diff --git a/src/telemetry/events/representation.py b/src/telemetry/events/representation.py index 77d6b292..dcad8a12 100644 --- a/src/telemetry/events/representation.py +++ b/src/telemetry/events/representation.py @@ -104,7 +104,7 @@ class RepresentationCompletedEvent(BaseEvent): # Cap configuration + hit flags () batch_max_tokens: int = Field( default=0, - description="settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS at fetch time", + description="settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS at fetch time", ) max_input_tokens: int = Field( default=0, description="settings.DERIVER.MAX_INPUT_TOKENS at call time" diff --git a/tests/bench/calculate_expected_events.py b/tests/bench/calculate_expected_events.py index 348cd95c..45eea82b 100644 --- a/tests/bench/calculate_expected_events.py +++ b/tests/bench/calculate_expected_events.py @@ -246,7 +246,7 @@ def calculate_question_events( # Calculate representation events # Each unique (session, observed) pair generates one representation event - # (assuming messages fit within REPRESENTATION_BATCH_MAX_TOKENS) + # (assuming messages fit within REPRESENTATION_BATCH_LLM_MAX_TOKENS) # When merge_sessions=True, all messages go into one session if merge_sessions: # One merged session = one representation event diff --git a/tests/deriver/test_queue_processing.py b/tests/deriver/test_queue_processing.py index 68d4852c..e860f6cd 100644 --- a/tests/deriver/test_queue_processing.py +++ b/tests/deriver/test_queue_processing.py @@ -359,7 +359,7 @@ class TestQueueProcessing: peer = peers[0] # Create messages with token counts that exceed batch limit - limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS token_counts = [limit // 2, limit // 2, limit // 2] # Create and save messages to the database first @@ -479,7 +479,7 @@ class TestQueueProcessing: session, peers = sample_session_with_peers peer = peers[0] - cap = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + cap = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS # M1 + M2 sum to exactly the cap; M3 pushes over it. After SQL, # messages_context = [M1, M2]; the cap is genuinely binding *on the @@ -581,7 +581,7 @@ class TestQueueProcessing: session, peers = sample_session_with_peers peer_a = peers[0] peer_b = peers[1] if len(peers) > 1 else peers[0] - cap = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + cap = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS # Layout: 4 messages, ordered. # M1 (peer_a, queue, 200) @@ -744,7 +744,9 @@ class TestQueueProcessing: qm = QueueManager() # Mock the token limit to 2000 for this test - with patch.object(settings.DERIVER, "REPRESENTATION_BATCH_MAX_TOKENS", 2000): + with patch.object( + settings.DERIVER, "REPRESENTATION_BATCH_LLM_MAX_TOKENS", 2000 + ): # Test alice's work unit alice_work_unit_key = alice_queue_items[0].work_unit_key alice_aqs = models.ActiveQueueSession(work_unit_key=alice_work_unit_key) @@ -919,7 +921,9 @@ class TestQueueProcessing: qm = QueueManager() # Mock the token limit to 1500 for this test - with patch.object(settings.DERIVER, "REPRESENTATION_BATCH_MAX_TOKENS", 1500): + with patch.object( + settings.DERIVER, "REPRESENTATION_BATCH_LLM_MAX_TOKENS", 1500 + ): # Test alice's work unit # With per-work-unit anchoring + preceding context: # Alice starts at message 3, includes preceding message 2 (steve) for context @@ -1133,7 +1137,7 @@ class TestQueueProcessing: peer = peers[0] # Create messages where first message exceeds the batch limit - limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS token_counts = [limit + 1000, 100, 200] # First message way over limit # Create and save messages to the database first @@ -1250,7 +1254,7 @@ class TestQueueProcessing: peer = peers[0] # Create messages that test the exact boundary - limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS token_counts = [ limit // 2, limit // 2, @@ -1381,7 +1385,7 @@ class TestQueueProcessing: peer = peers[0] # Create messages with tokens BELOW the threshold - limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS token_counts = [100, 100, 100] # Total 300, way below 4096 messages: list[models.Message] = [] @@ -1652,7 +1656,7 @@ class TestQueueProcessing: session, peers = sample_session_with_peers peer = peers[0] - limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS # Create a single message that exceeds the threshold message = models.Message( @@ -1758,7 +1762,7 @@ class TestQueueProcessing: session, peers = sample_session_with_peers peer = peers[0] - limit = settings.DERIVER.REPRESENTATION_BATCH_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS # Create messages that sum to exactly the threshold token_counts = [limit // 2, limit // 2] diff --git a/tests/test_config.py b/tests/test_config.py index 28b58d81..707b6e8a 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -7,7 +7,8 @@ def _make_deriver_settings( *, MAX_INPUT_TOKENS: int = 25000, MAX_CUSTOM_INSTRUCTIONS_TOKENS: int = 2000, - REPRESENTATION_BATCH_MAX_TOKENS: int = 1024, + REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS: int = 512, + REPRESENTATION_BATCH_LLM_MAX_TOKENS: int = 1024, REPRESENTATION_BATCH_MAX_AGE_SECONDS: int = 1800, ) -> DeriverSettings: return DeriverSettings( @@ -17,7 +18,8 @@ def _make_deriver_settings( ), MAX_INPUT_TOKENS=MAX_INPUT_TOKENS, MAX_CUSTOM_INSTRUCTIONS_TOKENS=MAX_CUSTOM_INSTRUCTIONS_TOKENS, - REPRESENTATION_BATCH_MAX_TOKENS=REPRESENTATION_BATCH_MAX_TOKENS, + REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS, + REPRESENTATION_BATCH_LLM_MAX_TOKENS=REPRESENTATION_BATCH_LLM_MAX_TOKENS, REPRESENTATION_BATCH_MAX_AGE_SECONDS=REPRESENTATION_BATCH_MAX_AGE_SECONDS, ) @@ -50,3 +52,51 @@ def test_representation_batch_age_can_be_disabled_with_zero() -> None: def test_representation_batch_age_rejects_negative_values() -> None: with pytest.raises(ValueError, match="greater than or equal to 0"): _make_deriver_settings(REPRESENTATION_BATCH_MAX_AGE_SECONDS=-1) + + +def test_representation_batch_work_unit_target_can_be_disabled_with_zero() -> None: + settings = _make_deriver_settings(REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=0) + + assert settings.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS == 0 + + +def test_representation_batch_work_unit_target_rejects_negative_values() -> None: + with pytest.raises(ValueError, match="greater than or equal to 0"): + _make_deriver_settings(REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=-1) + + +def test_representation_batch_tokens_can_diverge() -> None: + settings = _make_deriver_settings( + REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=4096, + REPRESENTATION_BATCH_LLM_MAX_TOKENS=1024, + ) + + assert settings.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS == 4096 + assert settings.REPRESENTATION_BATCH_LLM_MAX_TOKENS == 1024 + + +def test_representation_batch_llm_max_cannot_exceed_max_input_tokens() -> None: + with pytest.raises(ValueError, match="cannot exceed max deriver input tokens"): + _make_deriver_settings( + MAX_INPUT_TOKENS=1000, + REPRESENTATION_BATCH_LLM_MAX_TOKENS=2048, + ) + + +def test_legacy_representation_batch_max_tokens_is_rejected() -> None: + with pytest.raises(ValueError, match="has been split into"): + DeriverSettings( + MODEL_CONFIG=ConfiguredModelSettings( + model="gpt-5.4-mini", + transport="openai", + ), + REPRESENTATION_BATCH_MAX_TOKENS=1024, # pyright: ignore[reportCallIssue] + ) + + +def test_legacy_representation_batch_max_tokens_env_var_is_rejected( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("DERIVER_REPRESENTATION_BATCH_MAX_TOKENS", "1024") + with pytest.raises(ValueError, match="has been split into"): + _make_deriver_settings() From 43d962d1609076f45e5b5edc9e937c76ada6b95b Mon Sep 17 00:00:00 2001 From: Eugene Eisenstein Date: Thu, 9 Jul 2026 10:42:52 -0400 Subject: [PATCH 3/3] rename to REPRESENTATION_BATCH_TARGET_INPUT_TOKENS --- .env.template | 2 +- config.toml.example | 2 +- docs/v3/contributing/configuration.mdx | 2 +- docs/v3/contributing/troubleshooting.mdx | 2 +- src/config.py | 10 +++++----- src/deriver/consumer.py | 2 +- src/deriver/deriver.py | 2 +- src/deriver/queue_manager.py | 2 +- src/telemetry/events/representation.py | 2 +- tests/bench/calculate_expected_events.py | 2 +- tests/deriver/test_queue_processing.py | 14 +++++++------- tests/test_config.py | 12 ++++++------ 12 files changed, 27 insertions(+), 27 deletions(-) diff --git a/.env.template b/.env.template index fbbd2a50..167fa857 100644 --- a/.env.template +++ b/.env.template @@ -132,7 +132,7 @@ LLM_OPENAI_API_KEY=your-api-key-here # DERIVER_MAX_CUSTOM_INSTRUCTIONS_TOKENS=2000 # DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100 # DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=512 # Min tokens a work unit accumulates before the deriver claims it; 0 disables the gate -# DERIVER_REPRESENTATION_BATCH_LLM_MAX_TOKENS=1024 # Max context-window tokens per deriver LLM call +# DERIVER_REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=1024 # Max context-window tokens per deriver LLM call # DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS=1800 # DERIVER_FLUSH_ENABLED=false # Bypass batch token threshold, process work immediately # DERIVER_MODEL_CONFIG__FALLBACK__MODEL= diff --git a/config.toml.example b/config.toml.example index 60dfb323..e64d90f0 100644 --- a/config.toml.example +++ b/config.toml.example @@ -110,7 +110,7 @@ MAX_INPUT_TOKENS = 25000 MAX_CUSTOM_INSTRUCTIONS_TOKENS = 2000 WORKING_REPRESENTATION_MAX_OBSERVATIONS = 100 REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS = 512 # Min tokens a work unit accumulates before the deriver claims it; 0 disables the gate -REPRESENTATION_BATCH_LLM_MAX_TOKENS = 1024 # Max context-window tokens per deriver LLM call +REPRESENTATION_BATCH_TARGET_INPUT_TOKENS = 1024 # Max context-window tokens per deriver LLM call REPRESENTATION_BATCH_MAX_AGE_SECONDS = 1800 FLUSH_ENABLED = false # Bypass batch token threshold, process work immediately diff --git a/docs/v3/contributing/configuration.mdx b/docs/v3/contributing/configuration.mdx index eeda7b7a..8f4fb443 100644 --- a/docs/v3/contributing/configuration.mdx +++ b/docs/v3/contributing/configuration.mdx @@ -409,7 +409,7 @@ DERIVER_DEDUPLICATE=true DERIVER_LOG_OBSERVATIONS=false DERIVER_WORKING_REPRESENTATION_MAX_OBSERVATIONS=100 DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=512 -DERIVER_REPRESENTATION_BATCH_LLM_MAX_TOKENS=1024 +DERIVER_REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=1024 DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS=1800 ``` diff --git a/docs/v3/contributing/troubleshooting.mdx b/docs/v3/contributing/troubleshooting.mdx index 7db76b34..d9b745a1 100644 --- a/docs/v3/contributing/troubleshooting.mdx +++ b/docs/v3/contributing/troubleshooting.mdx @@ -109,7 +109,7 @@ Messages are stored but no observations, summaries, or representations are being ```bash DERIVER_WORKERS=4 ``` -5. **Representation Batching** — By default the deriver buffers representation work until a work unit has accumulated enough tokens, set via `DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS` (`0` disables the accumulation gate). A separate setting, `DERIVER_REPRESENTATION_BATCH_LLM_MAX_TOKENS`, caps the conversation window fed to each deriver LLM call when draining a claimed work unit. Sub-threshold tails become eligible after `DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS` (default 1800 seconds), so quiet sessions eventually flush without disabling batching globally. Set the age to `0` for legacy behavior where sub-threshold tails wait indefinitely. See [token batching](/v3/documentation/core-concepts/reasoning#token-batching) for more details +5. **Representation Batching** — By default the deriver buffers representation work until a work unit has accumulated enough tokens, set via `DERIVER_REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS` (`0` disables the accumulation gate). A separate setting, `DERIVER_REPRESENTATION_BATCH_TARGET_INPUT_TOKENS`, caps the conversation window fed to each deriver LLM call when draining a claimed work unit. Sub-threshold tails become eligible after `DERIVER_REPRESENTATION_BATCH_MAX_AGE_SECONDS` (default 1800 seconds), so quiet sessions eventually flush without disabling batching globally. Set the age to `0` for legacy behavior where sub-threshold tails wait indefinitely. See [token batching](/v3/documentation/core-concepts/reasoning#token-batching) for more details ## Alternative Provider Issues diff --git a/src/config.py b/src/config.py index 977b02d7..a5cf9bf7 100644 --- a/src/config.py +++ b/src/config.py @@ -870,7 +870,7 @@ class DeriverSettings(HonchoSettings): # interleaved context) fed to a single deriver LLM call when draining a # claimed work unit. The first unprocessed message is always included, # even if it alone exceeds the cap. - REPRESENTATION_BATCH_LLM_MAX_TOKENS: Annotated[ + REPRESENTATION_BATCH_TARGET_INPUT_TOKENS: Annotated[ int, Field(default=1024, ge=128, le=16_384), ] = 1024 @@ -901,7 +901,7 @@ class DeriverSettings(HonchoSettings): The old single setting was split into REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS (claim gate) and - REPRESENTATION_BATCH_LLM_MAX_TOKENS (per-LLM-call window cap). + REPRESENTATION_BATCH_TARGET_INPUT_TOKENS (per-LLM-call window cap). `extra="ignore"` would otherwise silently drop the old key and revert both roles to defaults — an operator-hostile failure mode for a batching knob — so reject it loudly instead. @@ -912,15 +912,15 @@ class DeriverSettings(HonchoSettings): ) if legacy_in_data or "DERIVER_REPRESENTATION_BATCH_MAX_TOKENS" in os.environ: raise ValueError( - "REPRESENTATION_BATCH_MAX_TOKENS has been split into REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS (minimum tokens a work unit must accumulate before it is claimed) and REPRESENTATION_BATCH_LLM_MAX_TOKENS (token cap on the context window per deriver LLM call). Set those instead." + "REPRESENTATION_BATCH_MAX_TOKENS has been split into REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS (minimum tokens a work unit must accumulate before it is claimed) and REPRESENTATION_BATCH_TARGET_INPUT_TOKENS (token cap on the context window per deriver LLM call). Set those instead." ) return data # pyright: ignore[reportUnknownVariableType] @model_validator(mode="after") def validate_batch_tokens_vs_context_limit(self): - if self.REPRESENTATION_BATCH_LLM_MAX_TOKENS > self.MAX_INPUT_TOKENS: + if self.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS > self.MAX_INPUT_TOKENS: raise ValueError( - f"REPRESENTATION_BATCH_LLM_MAX_TOKENS ({self.REPRESENTATION_BATCH_LLM_MAX_TOKENS}) cannot exceed max deriver input tokens ({self.MAX_INPUT_TOKENS})" + f"REPRESENTATION_BATCH_TARGET_INPUT_TOKENS ({self.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS}) cannot exceed max deriver input tokens ({self.MAX_INPUT_TOKENS})" ) return self diff --git a/src/deriver/consumer.py b/src/deriver/consumer.py index 9a961df9..708d05f6 100644 --- a/src/deriver/consumer.py +++ b/src/deriver/consumer.py @@ -176,7 +176,7 @@ async def process_representation_batch( queue_item_message_ids: Message IDs from queue items hit_batch_token_cap: whether the queue batcher clamped this batch to fit was_flush_enabled: snapshot of DERIVER.FLUSH_ENABLED at fetch time - batch_max_tokens: DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS snapshot + batch_max_tokens: DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS snapshot """ if not messages or not messages[0]: logger.debug("process_representation_batch received no messages") diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index 93f18eb1..3e68d084 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -58,7 +58,7 @@ async def process_representation_tasks_batch( queue_item_message_ids: Message IDs from queue items being processed hit_batch_token_cap: queue batcher clamped this batch to fit was_flush_enabled: DERIVER.FLUSH_ENABLED snapshot at batch time - batch_max_tokens: DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS snapshot + batch_max_tokens: DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS snapshot """ if not messages: return diff --git a/src/deriver/queue_manager.py b/src/deriver/queue_manager.py index 145a6252..498372ed 100644 --- a/src/deriver/queue_manager.py +++ b/src/deriver/queue_manager.py @@ -818,7 +818,7 @@ class QueueManager: f"{task_type} tasks are not supported for get_queue_item_batch" ) - batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS + batch_max_tokens = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS was_flush_enabled = settings.DERIVER.FLUSH_ENABLED parsed_key = parse_work_unit_key(work_unit_key) messages_context: list[models.Message] = [] diff --git a/src/telemetry/events/representation.py b/src/telemetry/events/representation.py index dcad8a12..92db6806 100644 --- a/src/telemetry/events/representation.py +++ b/src/telemetry/events/representation.py @@ -104,7 +104,7 @@ class RepresentationCompletedEvent(BaseEvent): # Cap configuration + hit flags () batch_max_tokens: int = Field( default=0, - description="settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS at fetch time", + description="settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS at fetch time", ) max_input_tokens: int = Field( default=0, description="settings.DERIVER.MAX_INPUT_TOKENS at call time" diff --git a/tests/bench/calculate_expected_events.py b/tests/bench/calculate_expected_events.py index 45eea82b..b9696262 100644 --- a/tests/bench/calculate_expected_events.py +++ b/tests/bench/calculate_expected_events.py @@ -246,7 +246,7 @@ def calculate_question_events( # Calculate representation events # Each unique (session, observed) pair generates one representation event - # (assuming messages fit within REPRESENTATION_BATCH_LLM_MAX_TOKENS) + # (assuming messages fit within REPRESENTATION_BATCH_TARGET_INPUT_TOKENS) # When merge_sessions=True, all messages go into one session if merge_sessions: # One merged session = one representation event diff --git a/tests/deriver/test_queue_processing.py b/tests/deriver/test_queue_processing.py index e860f6cd..81dd0632 100644 --- a/tests/deriver/test_queue_processing.py +++ b/tests/deriver/test_queue_processing.py @@ -359,7 +359,7 @@ class TestQueueProcessing: peer = peers[0] # Create messages with token counts that exceed batch limit - limit = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS token_counts = [limit // 2, limit // 2, limit // 2] # Create and save messages to the database first @@ -479,7 +479,7 @@ class TestQueueProcessing: session, peers = sample_session_with_peers peer = peers[0] - cap = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS + cap = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS # M1 + M2 sum to exactly the cap; M3 pushes over it. After SQL, # messages_context = [M1, M2]; the cap is genuinely binding *on the @@ -581,7 +581,7 @@ class TestQueueProcessing: session, peers = sample_session_with_peers peer_a = peers[0] peer_b = peers[1] if len(peers) > 1 else peers[0] - cap = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS + cap = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS # Layout: 4 messages, ordered. # M1 (peer_a, queue, 200) @@ -745,7 +745,7 @@ class TestQueueProcessing: # Mock the token limit to 2000 for this test with patch.object( - settings.DERIVER, "REPRESENTATION_BATCH_LLM_MAX_TOKENS", 2000 + settings.DERIVER, "REPRESENTATION_BATCH_TARGET_INPUT_TOKENS", 2000 ): # Test alice's work unit alice_work_unit_key = alice_queue_items[0].work_unit_key @@ -922,7 +922,7 @@ class TestQueueProcessing: # Mock the token limit to 1500 for this test with patch.object( - settings.DERIVER, "REPRESENTATION_BATCH_LLM_MAX_TOKENS", 1500 + settings.DERIVER, "REPRESENTATION_BATCH_TARGET_INPUT_TOKENS", 1500 ): # Test alice's work unit # With per-work-unit anchoring + preceding context: @@ -1137,7 +1137,7 @@ class TestQueueProcessing: peer = peers[0] # Create messages where first message exceeds the batch limit - limit = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS token_counts = [limit + 1000, 100, 200] # First message way over limit # Create and save messages to the database first @@ -1254,7 +1254,7 @@ class TestQueueProcessing: peer = peers[0] # Create messages that test the exact boundary - limit = settings.DERIVER.REPRESENTATION_BATCH_LLM_MAX_TOKENS + limit = settings.DERIVER.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS token_counts = [ limit // 2, limit // 2, diff --git a/tests/test_config.py b/tests/test_config.py index 707b6e8a..9b242e55 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -8,7 +8,7 @@ def _make_deriver_settings( MAX_INPUT_TOKENS: int = 25000, MAX_CUSTOM_INSTRUCTIONS_TOKENS: int = 2000, REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS: int = 512, - REPRESENTATION_BATCH_LLM_MAX_TOKENS: int = 1024, + REPRESENTATION_BATCH_TARGET_INPUT_TOKENS: int = 1024, REPRESENTATION_BATCH_MAX_AGE_SECONDS: int = 1800, ) -> DeriverSettings: return DeriverSettings( @@ -19,7 +19,7 @@ def _make_deriver_settings( MAX_INPUT_TOKENS=MAX_INPUT_TOKENS, MAX_CUSTOM_INSTRUCTIONS_TOKENS=MAX_CUSTOM_INSTRUCTIONS_TOKENS, REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS, - REPRESENTATION_BATCH_LLM_MAX_TOKENS=REPRESENTATION_BATCH_LLM_MAX_TOKENS, + REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=REPRESENTATION_BATCH_TARGET_INPUT_TOKENS, REPRESENTATION_BATCH_MAX_AGE_SECONDS=REPRESENTATION_BATCH_MAX_AGE_SECONDS, ) @@ -68,18 +68,18 @@ def test_representation_batch_work_unit_target_rejects_negative_values() -> None def test_representation_batch_tokens_can_diverge() -> None: settings = _make_deriver_settings( REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS=4096, - REPRESENTATION_BATCH_LLM_MAX_TOKENS=1024, + REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=1024, ) assert settings.REPRESENTATION_BATCH_WORK_UNIT_TARGET_TOKENS == 4096 - assert settings.REPRESENTATION_BATCH_LLM_MAX_TOKENS == 1024 + assert settings.REPRESENTATION_BATCH_TARGET_INPUT_TOKENS == 1024 -def test_representation_batch_llm_max_cannot_exceed_max_input_tokens() -> None: +def test_representation_batch_target_input_cannot_exceed_max_input_tokens() -> None: with pytest.raises(ValueError, match="cannot exceed max deriver input tokens"): _make_deriver_settings( MAX_INPUT_TOKENS=1000, - REPRESENTATION_BATCH_LLM_MAX_TOKENS=2048, + REPRESENTATION_BATCH_TARGET_INPUT_TOKENS=2048, )