Coverage for src / agent_contracts / validator.py: 84%
126 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-09 00:42 +0900
1"""ContractValidator - Static validation for node contracts.
3Validates contracts at registration time to catch configuration
4errors before runtime.
5"""
6from __future__ import annotations
8from dataclasses import dataclass, field
9from typing import TYPE_CHECKING
11from agent_contracts.utils.logging import get_logger
13if TYPE_CHECKING:
14 from agent_contracts.registry import NodeRegistry
16logger = get_logger("agent_contracts.validator")
19# =============================================================================
20# Validation Result
21# =============================================================================
23@dataclass
24class ValidationResult:
25 """Result of contract validation.
27 Attributes:
28 errors: Fatal issues that prevent execution (unknown slices, etc.)
29 warnings: Potential issues that deserve attention
30 info: Informational messages (shared writers, etc.)
31 """
32 errors: list[str] = field(default_factory=list)
33 warnings: list[str] = field(default_factory=list)
34 info: list[str] = field(default_factory=list)
36 @property
37 def has_errors(self) -> bool:
38 """Check if there are any errors."""
39 return len(self.errors) > 0
41 @property
42 def has_warnings(self) -> bool:
43 """Check if there are any warnings."""
44 return len(self.warnings) > 0
46 @property
47 def is_valid(self) -> bool:
48 """Check if validation passed (no errors)."""
49 return not self.has_errors
51 def __str__(self) -> str:
52 """Human-readable format."""
53 lines = []
55 if self.errors:
56 lines.append("ERRORS:")
57 for error in self.errors:
58 lines.append(f" - {error}")
60 if self.warnings:
61 lines.append("WARNINGS:")
62 for warning in self.warnings:
63 lines.append(f" - {warning}")
65 if self.info:
66 lines.append("INFO:")
67 for info_msg in self.info:
68 lines.append(f" - {info_msg}")
70 if not lines:
71 lines.append("✅ All validations passed")
73 return "\n".join(lines)
76# =============================================================================
77# Contract Validator
78# =============================================================================
80class ContractValidator:
81 """Validator for node contracts.
83 Performs static analysis on registered contracts to detect:
84 - Unknown slice names in reads/writes
85 - Missing service dependencies
86 - Orphan/unreachable nodes
87 - Shared writers (informational)
89 Example:
90 from agent_contracts import ContractValidator, get_node_registry
92 registry = get_node_registry()
93 # ... register nodes ...
95 validator = ContractValidator(registry)
96 result = validator.validate()
98 if result.has_errors:
99 print(result)
100 sys.exit(1)
101 """
103 def __init__(
104 self,
105 registry: "NodeRegistry",
106 known_services: set[str] | None = None,
107 ):
108 """Initialize validator.
110 Args:
111 registry: Node registry to validate
112 known_services: Set of known service names for validation.
113 If None, service validation is skipped.
114 """
115 self._registry = registry
116 self._known_services = known_services
118 def validate(self) -> ValidationResult:
119 """Run all validations.
121 Returns:
122 ValidationResult with errors, warnings, and info
123 """
124 result = ValidationResult()
126 # Run validations
127 self._validate_slices(result)
128 self._validate_services(result)
129 self._validate_reachability(result)
130 self._report_shared_writers(result)
132 # Log summary
133 if result.has_errors:
134 logger.error(f"Contract validation failed: {len(result.errors)} errors")
135 elif result.has_warnings:
136 logger.warning(f"Contract validation passed with {len(result.warnings)} warnings")
137 else:
138 logger.info("Contract validation passed")
140 return result
142 def _validate_slices(self, result: ValidationResult) -> None:
143 """Validate that all slice names are known."""
144 valid_slices = self._registry._valid_slices
146 for name in self._registry.get_all_nodes():
147 contract = self._registry.get_contract(name)
148 if not contract:
149 continue
151 # Check reads
152 for slice_name in contract.reads:
153 if slice_name not in valid_slices:
154 result.errors.append(
155 f"Unknown slice '{slice_name}' in node '{name}' reads"
156 )
158 # Check writes
159 for slice_name in contract.writes:
160 if slice_name not in valid_slices:
161 result.errors.append(
162 f"Unknown slice '{slice_name}' in node '{name}' writes"
163 )
165 def _validate_services(self, result: ValidationResult) -> None:
166 """Validate that required services are known."""
167 if self._known_services is None:
168 return # Skip if no known services provided
170 for name in self._registry.get_all_nodes():
171 contract = self._registry.get_contract(name)
172 if not contract:
173 continue
175 for service_name in contract.services:
176 if service_name not in self._known_services:
177 result.warnings.append(
178 f"Unknown service '{service_name}' required by node '{name}'"
179 )
181 def _validate_reachability(self, result: ValidationResult) -> None:
182 """Check for orphan/unreachable nodes."""
183 for name in self._registry.get_all_nodes():
184 contract = self._registry.get_contract(name)
185 if not contract:
186 continue
188 # Check for orphan (no supervisor)
189 if not contract.supervisor:
190 result.warnings.append(
191 f"Node '{name}' has no supervisor (orphan)"
192 )
193 continue
195 # Check for unreachable (no trigger conditions)
196 if not contract.trigger_conditions:
197 result.warnings.append(
198 f"Node '{name}' has no trigger conditions (may be unreachable)"
199 )
201 def _report_shared_writers(self, result: ValidationResult) -> None:
202 """Report slices with multiple writers (informational)."""
203 shared_writers = self.get_shared_writers()
205 for slice_name, writers in shared_writers.items():
206 if len(writers) > 1:
207 writers_str = ", ".join(sorted(writers))
208 result.info.append(
209 f"Shared writers for '{slice_name}': {writers_str}"
210 )
212 def get_shared_writers(self) -> dict[str, list[str]]:
213 """Get all slices and their writers.
215 Returns:
216 {slice_name: [node_names that write to it]}
217 """
218 writers: dict[str, list[str]] = {}
220 for name in self._registry.get_all_nodes():
221 contract = self._registry.get_contract(name)
222 if not contract:
223 continue
225 for slice_name in contract.writes:
226 if slice_name not in writers:
227 writers[slice_name] = []
228 writers[slice_name].append(name)
230 return writers
232 def get_slice_readers(self) -> dict[str, list[str]]:
233 """Get all slices and their readers.
235 Returns:
236 {slice_name: [node_names that read from it]}
237 """
238 readers: dict[str, list[str]] = {}
240 for name in self._registry.get_all_nodes():
241 contract = self._registry.get_contract(name)
242 if not contract:
243 continue
245 for slice_name in contract.reads:
246 if slice_name not in readers:
247 readers[slice_name] = []
248 readers[slice_name].append(name)
250 return readers
252 def get_unused_slices(self) -> dict[str, str]:
253 """Find slices that are written but never read, or vice versa.
255 Returns:
256 {slice_name: "write_only" | "read_only"}
257 """
258 writers = self.get_shared_writers()
259 readers = self.get_slice_readers()
261 all_slices = set(writers.keys()) | set(readers.keys())
262 unused: dict[str, str] = {}
264 for slice_name in all_slices:
265 has_writers = slice_name in writers and len(writers[slice_name]) > 0
266 has_readers = slice_name in readers and len(readers[slice_name]) > 0
268 if has_writers and not has_readers:
269 unused[slice_name] = "write_only"
270 elif has_readers and not has_writers:
271 # Read-only is only a concern for non-input slices
272 if slice_name != "request":
273 unused[slice_name] = "read_only"
275 return unused