Coverage for src / harnessutils / config.py: 84%
164 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-18 08:30 -0600
1"""Configuration schema for harness-utils."""
3from dataclasses import dataclass, field
4from pathlib import Path
5from typing import Any
8@dataclass
9class TruncationConfig:
10 """Configuration for Tier 1: Output truncation."""
12 max_lines: int = 2000
13 max_bytes: int = 50 * 1024 # 50KB
14 direction: str = "head" # "head" or "tail"
16 # Phase 2: Content-aware truncation
17 max_tokens: int = 2000 # Token-based limit
18 use_content_aware: bool = True # Enable content-aware truncation
19 preserve_errors: bool = True # Keep all errors/warnings in logs
20 json_array_limit: int = 10 # Keep first/last N items in JSON arrays
21 stacktrace_frame_limit: int = 20 # Keep top/bottom N frames in stacktraces
23 def validate(self) -> None:
24 """Validate truncation configuration.
26 Raises:
27 ConfigurationError: If configuration is invalid
28 """
29 from harnessutils.exceptions import ConfigurationError
31 if self.max_lines <= 0:
32 raise ConfigurationError(
33 f"max_lines must be positive, got {self.max_lines}",
34 "Set max_lines to a positive integer (e.g., 2000)",
35 )
37 if self.max_bytes <= 0:
38 raise ConfigurationError(
39 f"max_bytes must be positive, got {self.max_bytes}",
40 "Set max_bytes to a positive integer (e.g., 50000)",
41 )
43 if self.max_tokens <= 0:
44 raise ConfigurationError(
45 f"max_tokens must be positive, got {self.max_tokens}",
46 "Set max_tokens to a positive integer (e.g., 2000)",
47 )
49 if self.direction not in ["head", "tail"]:
50 raise ConfigurationError(
51 f"direction must be 'head' or 'tail', got '{self.direction}'",
52 "Set direction to either 'head' (keep beginning) or 'tail' (keep end)",
53 )
55 if self.json_array_limit <= 0:
56 raise ConfigurationError(
57 f"json_array_limit must be positive, got {self.json_array_limit}",
58 "Set json_array_limit to a positive integer (e.g., 10)",
59 )
61 if self.stacktrace_frame_limit <= 0:
62 raise ConfigurationError(
63 f"stacktrace_frame_limit must be positive, got {self.stacktrace_frame_limit}",
64 "Set stacktrace_frame_limit to a positive integer (e.g., 20)",
65 )
68@dataclass
69class PruningConfig:
70 """Configuration for Tier 2: Selective pruning."""
72 prune_protect: int = 40_000 # Keep recent 40K tokens
73 prune_minimum: int = 20_000 # Only prune if saves 20K+ tokens
74 protect_turns: int = 2 # Protect last 2 turns
75 protected_tools: list[str] = field(
76 default_factory=lambda: ["skill_execution", "subtask_invocation"]
77 )
79 # Importance scoring (Phase 1.2)
80 use_importance_scoring: bool = True # Enable smart pruning
81 recency_weight: float = 1.0 # Weight for recency score
82 size_weight: float = -0.5 # Weight for size penalty (negative)
83 semantic_weight: float = 2.0 # Weight for semantic importance
84 tool_priority_weight: float = 1.5 # Weight for tool type priority
85 recency_decay: float = 0.1 # Exponential decay rate per turn
87 # Tool importance map (higher = more important)
88 tool_importance: dict[str, float] = field(
89 default_factory=lambda: {
90 "read": 50.0,
91 "write": 100.0, # Code changes are important
92 "edit": 100.0,
93 "grep": 30.0, # Search results often repetitive
94 "glob": 30.0,
95 "bash": 70.0,
96 "skill_execution": 150.0, # Complex operations
97 "subtask_invocation": 150.0,
98 "error": 200.0, # Critical for debugging
99 }
100 )
102 # Semantic boost scores
103 error_boost: float = 500.0 # Boost for outputs with errors
104 warning_boost: float = 200.0 # Boost for warnings
105 user_requested_boost: float = 300.0 # User explicitly asked for this
107 # Deduplication (Phase 1.3)
108 detect_duplicates: bool = True # Enable duplicate detection
109 similarity_threshold: float = 0.8 # Similarity threshold (0.0-1.0)
110 duplicate_lookback: int = 20 # Check last N outputs for duplicates
112 def validate(self) -> None:
113 """Validate pruning configuration.
115 Raises:
116 ConfigurationError: If configuration is invalid
117 """
118 from harnessutils.exceptions import ConfigurationError
120 if self.prune_minimum >= self.prune_protect:
121 raise ConfigurationError(
122 f"prune_minimum ({self.prune_minimum}) must be < "
123 f"prune_protect ({self.prune_protect})",
124 "Set prune_minimum to a value less than prune_protect. "
125 "Example: prune_minimum=20000, prune_protect=40000",
126 )
128 if self.protect_turns < 0:
129 raise ConfigurationError(
130 f"protect_turns must be >= 0, got {self.protect_turns}",
131 "Set protect_turns to 0 or more (e.g., 2 to protect last 2 turns)",
132 )
134 if self.similarity_threshold < 0.0 or self.similarity_threshold > 1.0:
135 raise ConfigurationError(
136 f"similarity_threshold must be between 0.0 and 1.0, "
137 f"got {self.similarity_threshold}",
138 "Set similarity_threshold to a value between 0.0 (loose matching) "
139 "and 1.0 (exact matching). Try 0.8 for most cases.",
140 )
142 if self.duplicate_lookback <= 0:
143 raise ConfigurationError(
144 f"duplicate_lookback must be positive, got {self.duplicate_lookback}",
145 "Set duplicate_lookback to a positive integer (e.g., 20)",
146 )
148 # Check weights are non-negative
149 for weight_name in [
150 "recency_weight",
151 "semantic_weight",
152 "tool_priority_weight",
153 "error_boost",
154 "warning_boost",
155 "user_requested_boost",
156 ]:
157 weight = getattr(self, weight_name)
158 if weight < 0:
159 raise ConfigurationError(
160 f"{weight_name} must be >= 0, got {weight}",
161 f"Set {weight_name} to a non-negative value",
162 )
165@dataclass
166class TokenConfig:
167 """Configuration for token estimation."""
169 chars_per_token: int = 4
172@dataclass
173class ModelLimitsConfig:
174 """Configuration for model limits."""
176 default_context_limit: int = 200_000
177 default_output_limit: int = 8_192
179 def validate(self) -> None:
180 """Validate model limits configuration.
182 Raises:
183 ConfigurationError: If configuration is invalid
184 """
185 from harnessutils.exceptions import ConfigurationError
187 if self.default_context_limit <= 0:
188 raise ConfigurationError(
189 f"default_context_limit must be positive, got {self.default_context_limit}",
190 "Set default_context_limit to your model's context window size (e.g., 200000)",
191 )
193 if self.default_output_limit <= 0:
194 raise ConfigurationError(
195 f"default_output_limit must be positive, got {self.default_output_limit}",
196 "Set default_output_limit to your model's max output tokens (e.g., 8192)",
197 )
199 if self.default_output_limit >= self.default_context_limit:
200 raise ConfigurationError(
201 f"default_output_limit ({self.default_output_limit}) must be < "
202 f"default_context_limit ({self.default_context_limit})",
203 "Set default_output_limit to less than default_context_limit. "
204 "Example: context_limit=200000, output_limit=8192",
205 )
208@dataclass
209class StorageConfig:
210 """Configuration for storage layer."""
212 base_path: Path = field(default_factory=lambda: Path("data"))
213 retention_days: int = 7 # For truncated outputs
216@dataclass
217class SummarizationConfig:
218 """Configuration for Tier 3: Summarization.
220 Note: Model selection is user-controlled. Set differential_model and full_model
221 to match your LLM provider, or pass model parameter directly to compact().
222 If not set, your LLMClient.invoke() will receive None and should use its own default.
223 """
225 mode: str = "differential" # "differential" or "full"
226 differential_model: str | None = None # Optional: model for differential mode
227 full_model: str | None = None # Optional: model for full mode
228 max_messages_since_summary: int = 30 # Force full if exceeded
229 summarization_prompt: str | None = None # None → use SUMMARIZATION_PROMPT constant
230 differential_prompt: str | None = None # None → use DIFFERENTIAL_SUMMARIZATION_PROMPT
231 include_tool_outputs: bool = True # Include tool outputs in summarization input
232 tool_output_max_tokens: int = 300 # Max tokens per tool output in summarization
234 def validate(self) -> None:
235 """Validate summarization configuration.
237 Raises:
238 ConfigurationError: If configuration is invalid
239 """
240 from harnessutils.exceptions import ConfigurationError
242 if self.mode not in ["differential", "full"]:
243 raise ConfigurationError(
244 f"mode must be 'differential' or 'full', got '{self.mode}'",
245 "Set mode to either 'differential' (incremental) or 'full' "
246 "(complete re-summarization)",
247 )
249 if self.max_messages_since_summary <= 0:
250 raise ConfigurationError(
251 f"max_messages_since_summary must be positive, "
252 f"got {self.max_messages_since_summary}",
253 "Set max_messages_since_summary to a positive integer (e.g., 30)",
254 )
257@dataclass
258class CompactionConfig:
259 """Configuration for context compaction."""
261 auto: bool = True # Enable auto-summarization
262 prune: bool = True # Enable pruning
264 # Phase 2: Predictive overflow detection
265 use_predictive: bool = True # Enable predictive overflow detection
266 predictive_lookahead: int = 5 # Predict N turns ahead
267 predictive_safety_margin: float = 0.8 # Trigger at 80% of limit
269 def validate(self) -> None:
270 """Validate compaction configuration.
272 Raises:
273 ConfigurationError: If configuration is invalid
274 """
275 from harnessutils.exceptions import ConfigurationError
277 if self.predictive_lookahead <= 0:
278 raise ConfigurationError(
279 f"predictive_lookahead must be positive, got {self.predictive_lookahead}",
280 "Set predictive_lookahead to a positive integer (e.g., 5 turns ahead)",
281 )
283 if self.predictive_safety_margin <= 0.0 or self.predictive_safety_margin >= 1.0:
284 raise ConfigurationError(
285 f"predictive_safety_margin must be between 0.0 and 1.0 "
286 f"(exclusive), got {self.predictive_safety_margin}",
287 "Set predictive_safety_margin to a value like 0.8 "
288 "(trigger at 80% of limit)",
289 )
292@dataclass
293class HarnessConfig:
294 """Main configuration for harness-utils.
296 Provides all configuration parameters for context window management
297 with sensible defaults from the CTXWINARCH.md specification.
298 """
300 truncation: TruncationConfig = field(default_factory=TruncationConfig)
301 pruning: PruningConfig = field(default_factory=PruningConfig)
302 tokens: TokenConfig = field(default_factory=TokenConfig)
303 model_limits: ModelLimitsConfig = field(default_factory=ModelLimitsConfig)
304 storage: StorageConfig = field(default_factory=StorageConfig)
305 compaction: CompactionConfig = field(default_factory=CompactionConfig)
306 summarization: SummarizationConfig = field(default_factory=SummarizationConfig)
308 def validate(self) -> None:
309 """Validate entire configuration.
311 Checks all sub-configs and cross-config constraints.
313 Raises:
314 ConfigurationError: If any configuration is invalid
315 """
316 from harnessutils.exceptions import ConfigurationError
318 # Validate all sub-configs
319 self.truncation.validate()
320 self.pruning.validate()
321 self.model_limits.validate()
322 self.compaction.validate()
323 self.summarization.validate()
325 # Cross-config validations
326 # 1. prune_protect must be less than context_limit
327 usable_limit = (
328 self.model_limits.default_context_limit
329 - self.model_limits.default_output_limit
330 )
331 if self.pruning.prune_protect >= usable_limit:
332 raise ConfigurationError(
333 f"prune_protect ({self.pruning.prune_protect}) must be < "
334 f"usable context ({usable_limit} = "
335 f"{self.model_limits.default_context_limit} - "
336 f"{self.model_limits.default_output_limit})",
337 f"Set prune_protect to less than {usable_limit}. "
338 f"Recommended: {int(usable_limit * 0.2)} (20% of usable context)",
339 )
341 @classmethod
342 def from_dict(cls, data: dict[str, Any]) -> "HarnessConfig":
343 """Create configuration from dictionary.
345 Args:
346 data: Configuration dictionary
348 Returns:
349 HarnessConfig instance
350 """
351 config = cls()
353 if "truncation" in data:
354 config.truncation = TruncationConfig(**data["truncation"])
355 if "pruning" in data:
356 protected = data["pruning"].get("protected_tools")
357 pruning_data = {k: v for k, v in data["pruning"].items() if k != "protected_tools"}
358 if protected:
359 pruning_data["protected_tools"] = protected
360 config.pruning = PruningConfig(**pruning_data)
361 if "tokens" in data:
362 config.tokens = TokenConfig(**data["tokens"])
363 if "model_limits" in data:
364 config.model_limits = ModelLimitsConfig(**data["model_limits"])
365 if "storage" in data:
366 storage_data = data["storage"].copy()
367 if "base_path" in storage_data:
368 storage_data["base_path"] = Path(storage_data["base_path"])
369 config.storage = StorageConfig(**storage_data)
370 if "compaction" in data:
371 config.compaction = CompactionConfig(**data["compaction"])
372 if "summarization" in data:
373 config.summarization = SummarizationConfig(**data["summarization"])
375 return config
377 @classmethod
378 def from_toml(cls, path: Path) -> "HarnessConfig":
379 """Load configuration from TOML file.
381 Args:
382 path: Path to TOML configuration file
384 Returns:
385 HarnessConfig instance
386 """
387 import tomllib
389 with open(path, "rb") as f:
390 data = tomllib.load(f)
392 return cls.from_dict(data)
394 @classmethod
395 def from_json(cls, path: Path) -> "HarnessConfig":
396 """Load configuration from JSON file.
398 Args:
399 path: Path to JSON configuration file
401 Returns:
402 HarnessConfig instance
403 """
404 import json
406 with open(path) as f:
407 data = json.load(f)
409 return cls.from_dict(data)