Coverage for src / harnessutils / config.py: 84%
165 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-04-07 09:04 -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 # Query-conditioned re-scoring
108 query_weight: float = 1.5 # Weight for query relevance boost
110 # Deduplication (Phase 1.3)
111 detect_duplicates: bool = True # Enable duplicate detection
112 similarity_threshold: float = 0.8 # Similarity threshold (0.0-1.0)
113 duplicate_lookback: int = 20 # Check last N outputs for duplicates
115 def validate(self) -> None:
116 """Validate pruning configuration.
118 Raises:
119 ConfigurationError: If configuration is invalid
120 """
121 from harnessutils.exceptions import ConfigurationError
123 if self.prune_minimum >= self.prune_protect:
124 raise ConfigurationError(
125 f"prune_minimum ({self.prune_minimum}) must be < "
126 f"prune_protect ({self.prune_protect})",
127 "Set prune_minimum to a value less than prune_protect. "
128 "Example: prune_minimum=20000, prune_protect=40000",
129 )
131 if self.protect_turns < 0:
132 raise ConfigurationError(
133 f"protect_turns must be >= 0, got {self.protect_turns}",
134 "Set protect_turns to 0 or more (e.g., 2 to protect last 2 turns)",
135 )
137 if self.similarity_threshold < 0.0 or self.similarity_threshold > 1.0:
138 raise ConfigurationError(
139 f"similarity_threshold must be between 0.0 and 1.0, "
140 f"got {self.similarity_threshold}",
141 "Set similarity_threshold to a value between 0.0 (loose matching) "
142 "and 1.0 (exact matching). Try 0.8 for most cases.",
143 )
145 if self.duplicate_lookback <= 0:
146 raise ConfigurationError(
147 f"duplicate_lookback must be positive, got {self.duplicate_lookback}",
148 "Set duplicate_lookback to a positive integer (e.g., 20)",
149 )
151 # Check weights are non-negative
152 for weight_name in [
153 "recency_weight",
154 "semantic_weight",
155 "tool_priority_weight",
156 "error_boost",
157 "warning_boost",
158 "user_requested_boost",
159 ]:
160 weight = getattr(self, weight_name)
161 if weight < 0:
162 raise ConfigurationError(
163 f"{weight_name} must be >= 0, got {weight}",
164 f"Set {weight_name} to a non-negative value",
165 )
168@dataclass
169class TokenConfig:
170 """Configuration for token estimation."""
172 chars_per_token: int = 4
175@dataclass
176class ModelLimitsConfig:
177 """Configuration for model limits."""
179 default_context_limit: int = 200_000
180 default_output_limit: int = 8_192
182 def validate(self) -> None:
183 """Validate model limits configuration.
185 Raises:
186 ConfigurationError: If configuration is invalid
187 """
188 from harnessutils.exceptions import ConfigurationError
190 if self.default_context_limit <= 0:
191 raise ConfigurationError(
192 f"default_context_limit must be positive, got {self.default_context_limit}",
193 "Set default_context_limit to your model's context window size (e.g., 200000)",
194 )
196 if self.default_output_limit <= 0:
197 raise ConfigurationError(
198 f"default_output_limit must be positive, got {self.default_output_limit}",
199 "Set default_output_limit to your model's max output tokens (e.g., 8192)",
200 )
202 if self.default_output_limit >= self.default_context_limit:
203 raise ConfigurationError(
204 f"default_output_limit ({self.default_output_limit}) must be < "
205 f"default_context_limit ({self.default_context_limit})",
206 "Set default_output_limit to less than default_context_limit. "
207 "Example: context_limit=200000, output_limit=8192",
208 )
211@dataclass
212class StorageConfig:
213 """Configuration for storage layer."""
215 base_path: Path = field(default_factory=lambda: Path("data"))
216 retention_days: int = 7 # For truncated outputs
219@dataclass
220class SummarizationConfig:
221 """Configuration for Tier 3: Summarization.
223 Note: Model selection is user-controlled. Set differential_model and full_model
224 to match your LLM provider, or pass model parameter directly to compact().
225 If not set, your LLMClient.invoke() will receive None and should use its own default.
226 """
228 mode: str = "differential" # "differential" or "full"
229 differential_model: str | None = None # Optional: model for differential mode
230 full_model: str | None = None # Optional: model for full mode
231 max_messages_since_summary: int = 30 # Force full if exceeded
232 summarization_prompt: str | None = None # None → use SUMMARIZATION_PROMPT constant
233 differential_prompt: str | None = None # None → use DIFFERENTIAL_SUMMARIZATION_PROMPT
234 include_tool_outputs: bool = True # Include tool outputs in summarization input
235 tool_output_max_tokens: int = 300 # Max tokens per tool output in summarization
237 def validate(self) -> None:
238 """Validate summarization configuration.
240 Raises:
241 ConfigurationError: If configuration is invalid
242 """
243 from harnessutils.exceptions import ConfigurationError
245 if self.mode not in ["differential", "full"]:
246 raise ConfigurationError(
247 f"mode must be 'differential' or 'full', got '{self.mode}'",
248 "Set mode to either 'differential' (incremental) or 'full' "
249 "(complete re-summarization)",
250 )
252 if self.max_messages_since_summary <= 0:
253 raise ConfigurationError(
254 f"max_messages_since_summary must be positive, "
255 f"got {self.max_messages_since_summary}",
256 "Set max_messages_since_summary to a positive integer (e.g., 30)",
257 )
260@dataclass
261class CompactionConfig:
262 """Configuration for context compaction."""
264 auto: bool = True # Enable auto-summarization
265 prune: bool = True # Enable pruning
267 # Phase 2: Predictive overflow detection
268 use_predictive: bool = True # Enable predictive overflow detection
269 predictive_lookahead: int = 5 # Predict N turns ahead
270 predictive_safety_margin: float = 0.8 # Trigger at 80% of limit
272 def validate(self) -> None:
273 """Validate compaction configuration.
275 Raises:
276 ConfigurationError: If configuration is invalid
277 """
278 from harnessutils.exceptions import ConfigurationError
280 if self.predictive_lookahead <= 0:
281 raise ConfigurationError(
282 f"predictive_lookahead must be positive, got {self.predictive_lookahead}",
283 "Set predictive_lookahead to a positive integer (e.g., 5 turns ahead)",
284 )
286 if self.predictive_safety_margin <= 0.0 or self.predictive_safety_margin >= 1.0:
287 raise ConfigurationError(
288 f"predictive_safety_margin must be between 0.0 and 1.0 "
289 f"(exclusive), got {self.predictive_safety_margin}",
290 "Set predictive_safety_margin to a value like 0.8 "
291 "(trigger at 80% of limit)",
292 )
295@dataclass
296class HarnessConfig:
297 """Main configuration for harness-utils.
299 Provides all configuration parameters for context window management
300 with sensible defaults from the CTXWINARCH.md specification.
301 """
303 truncation: TruncationConfig = field(default_factory=TruncationConfig)
304 pruning: PruningConfig = field(default_factory=PruningConfig)
305 tokens: TokenConfig = field(default_factory=TokenConfig)
306 model_limits: ModelLimitsConfig = field(default_factory=ModelLimitsConfig)
307 storage: StorageConfig = field(default_factory=StorageConfig)
308 compaction: CompactionConfig = field(default_factory=CompactionConfig)
309 summarization: SummarizationConfig = field(default_factory=SummarizationConfig)
311 def validate(self) -> None:
312 """Validate entire configuration.
314 Checks all sub-configs and cross-config constraints.
316 Raises:
317 ConfigurationError: If any configuration is invalid
318 """
319 from harnessutils.exceptions import ConfigurationError
321 # Validate all sub-configs
322 self.truncation.validate()
323 self.pruning.validate()
324 self.model_limits.validate()
325 self.compaction.validate()
326 self.summarization.validate()
328 # Cross-config validations
329 # 1. prune_protect must be less than context_limit
330 usable_limit = (
331 self.model_limits.default_context_limit
332 - self.model_limits.default_output_limit
333 )
334 if self.pruning.prune_protect >= usable_limit:
335 raise ConfigurationError(
336 f"prune_protect ({self.pruning.prune_protect}) must be < "
337 f"usable context ({usable_limit} = "
338 f"{self.model_limits.default_context_limit} - "
339 f"{self.model_limits.default_output_limit})",
340 f"Set prune_protect to less than {usable_limit}. "
341 f"Recommended: {int(usable_limit * 0.2)} (20% of usable context)",
342 )
344 @classmethod
345 def from_dict(cls, data: dict[str, Any]) -> "HarnessConfig":
346 """Create configuration from dictionary.
348 Args:
349 data: Configuration dictionary
351 Returns:
352 HarnessConfig instance
353 """
354 config = cls()
356 if "truncation" in data:
357 config.truncation = TruncationConfig(**data["truncation"])
358 if "pruning" in data:
359 protected = data["pruning"].get("protected_tools")
360 pruning_data = {k: v for k, v in data["pruning"].items() if k != "protected_tools"}
361 if protected:
362 pruning_data["protected_tools"] = protected
363 config.pruning = PruningConfig(**pruning_data)
364 if "tokens" in data:
365 config.tokens = TokenConfig(**data["tokens"])
366 if "model_limits" in data:
367 config.model_limits = ModelLimitsConfig(**data["model_limits"])
368 if "storage" in data:
369 storage_data = data["storage"].copy()
370 if "base_path" in storage_data:
371 storage_data["base_path"] = Path(storage_data["base_path"])
372 config.storage = StorageConfig(**storage_data)
373 if "compaction" in data:
374 config.compaction = CompactionConfig(**data["compaction"])
375 if "summarization" in data:
376 config.summarization = SummarizationConfig(**data["summarization"])
378 return config
380 @classmethod
381 def from_toml(cls, path: Path) -> "HarnessConfig":
382 """Load configuration from TOML file.
384 Args:
385 path: Path to TOML configuration file
387 Returns:
388 HarnessConfig instance
389 """
390 import tomllib
392 with open(path, "rb") as f:
393 data = tomllib.load(f)
395 return cls.from_dict(data)
397 @classmethod
398 def from_json(cls, path: Path) -> "HarnessConfig":
399 """Load configuration from JSON file.
401 Args:
402 path: Path to JSON configuration file
404 Returns:
405 HarnessConfig instance
406 """
407 import json
409 with open(path) as f:
410 data = json.load(f)
412 return cls.from_dict(data)