| 1 | #!/usr/bin/env python3 |
| 2 | """Fan-in validation contract for crawl matrix artifacts. |
| 3 | |
| 4 | This module implements the non-publishing fan-in validator that ensures |
| 5 | crawl artifacts (whether from monolithic or matrix legs) meet the |
| 6 | consistency requirements before canonical output is produced. |
| 7 | |
| 8 | Validation checks: |
| 9 | - Schema/version consistency across all legs |
| 10 | - Checksum integrity (content checksums match declared values) |
| 11 | - Window consistency (all legs use the same since/until) |
| 12 | - Deterministic ordering (repos by full_name, articles by source+url) |
| 13 | - Duplicate URL/repo handling (dedup with documented priority rules) |
| 14 | - Stale cache rejection (artifacts older than configured max age) |
| 15 | - Source status metadata (required vs optional failure behavior) |
| 16 | - Byte-stable output verification (same inputs → same canonical output) |
| 17 | |
| 18 | References: |
| 19 | - Issue #333: Define crawl matrix readiness and fan-in validation path |
| 20 | - docs/matrix-crawl-fan-in-contracts.md: Full contract specification |
| 21 | """ |
| 22 | |
| 23 | from __future__ import annotations |
| 24 | |
| 25 | import hashlib |
| 26 | import json |
| 27 | from dataclasses import dataclass, field |
| 28 | from datetime import UTC, datetime, timedelta |
| 29 | from typing import Any |
| 30 | |
| 31 | from scripts.run_context import RunContext |
| 32 | |
| 33 | |
| 34 | class FanInContractError(Exception): |
| 35 | """Raised when a fan-in contract violation is detected.""" |
| 36 | |
| 37 | pass |
| 38 | |
| 39 | |
| 40 | @dataclass(slots=True) |
| 41 | class ValidationResult: |
| 42 | """Result of fan-in validation.""" |
| 43 | |
| 44 | valid: bool |
| 45 | errors: list[str] = field(default_factory=list) |
| 46 | warnings: list[str] = field(default_factory=list) |
| 47 | artifact_count: int = 0 |
| 48 | sources_present: list[str] = field(default_factory=list) |
| 49 | sources_missing_required: list[str] = field(default_factory=list) |
| 50 | sources_missing_optional: list[str] = field(default_factory=list) |
| 51 | duplicate_urls: list[str] = field(default_factory=list) |
| 52 | duplicate_repos: list[str] = field(default_factory=list) |
| 53 | stale_artifacts: list[str] = field(default_factory=list) |
| 54 | |
| 55 | def to_dict(self) -> dict[str, Any]: |
| 56 | return { |
| 57 | "valid": self.valid, |
| 58 | "errors": self.errors, |
| 59 | "warnings": self.warnings, |
| 60 | "artifact_count": self.artifact_count, |
| 61 | "sources_present": self.sources_present, |
| 62 | "sources_missing_required": self.sources_missing_required, |
| 63 | "sources_missing_optional": self.sources_missing_optional, |
| 64 | "duplicate_urls": self.duplicate_urls, |
| 65 | "duplicate_repos": self.duplicate_repos, |
| 66 | "stale_artifacts": self.stale_artifacts, |
| 67 | } |
| 68 | |
| 69 | |
| 70 | # Maximum age of a per-source artifact before it's considered stale |
| 71 | DEFAULT_MAX_ARTIFACT_AGE = timedelta(hours=24) |
| 72 | |
| 73 | # Minimum percentage of required sources that must succeed |
| 74 | MINIMUM_SOURCE_SUCCESS_RATIO = 0.6 |
| 75 | |
| 76 | |
| 77 | def validate_artifact_schema( |
| 78 | artifact: dict[str, Any], |
| 79 | expected_schema_version: str, |
| 80 | ) -> list[str]: |
| 81 | """Validate artifact schema structure. Returns list of errors.""" |
| 82 | errors: list[str] = [] |
| 83 | |
| 84 | if not isinstance(artifact, dict): |
| 85 | return ["artifact must be a JSON object"] |
| 86 | |
| 87 | sv = artifact.get("schema_version") or artifact.get("source_artifact_schema_version") |
| 88 | if sv is None: |
| 89 | errors.append("missing schema_version field") |
| 90 | elif str(sv) != str(expected_schema_version): |
| 91 | errors.append(f"schema_version mismatch: expected '{expected_schema_version}', got '{sv}'") |
| 92 | |
| 93 | return errors |
| 94 | |
| 95 | |
| 96 | def validate_checksum_integrity(artifact: dict[str, Any]) -> list[str]: |
| 97 | """Verify that declared checksums match computed values.""" |
| 98 | errors: list[str] = [] |
| 99 | |
| 100 | # Check artifact_checksum if present |
| 101 | if "artifact_checksum" in artifact and "articles" in artifact: |
| 102 | payload = { |
| 103 | "source_id": artifact.get("source_id", ""), |
| 104 | "run_context": artifact.get("run_context", {}), |
| 105 | "articles": artifact.get("articles", []), |
| 106 | } |
| 107 | serialized = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
| 108 | computed = hashlib.sha256(serialized.encode("utf-8")).hexdigest() |
| 109 | if artifact["artifact_checksum"] != computed: |
| 110 | errors.append( |
| 111 | f"artifact_checksum mismatch for source '{artifact.get('source_id', '?')}': " |
| 112 | f"declared={artifact['artifact_checksum'][:16]}..., computed={computed[:16]}..." |
| 113 | ) |
| 114 | |
| 115 | # Check content_checksum in metrics if present |
| 116 | if "checksum" in artifact: |
| 117 | # For GitHub shard artifacts: checksum covers repositories array |
| 118 | if "repositories" in artifact: |
| 119 | content = json.dumps( |
| 120 | artifact["repositories"], sort_keys=True, separators=(",", ":"), ensure_ascii=False |
| 121 | ) |
| 122 | computed = hashlib.sha256(content.encode("utf-8")).hexdigest() |
| 123 | if artifact["checksum"] != computed: |
| 124 | errors.append(f"checksum mismatch for shard '{artifact.get('shard_id', '?')}'") |
| 125 | |
| 126 | return errors |
| 127 | |
| 128 | |
| 129 | def validate_window_consistency( |
| 130 | artifacts: list[dict[str, Any]], |
| 131 | run_context: RunContext | dict[str, Any], |
| 132 | ) -> list[str]: |
| 133 | """Ensure all artifacts share the same crawl window as the run context.""" |
| 134 | errors: list[str] = [] |
| 135 | |
| 136 | if isinstance(run_context, RunContext): |
| 137 | expected_since = run_context.since |
| 138 | expected_until = run_context.until |
| 139 | expected_week = run_context.week |
| 140 | else: |
| 141 | expected_since = run_context.get("since") or run_context.get("crawl_window", {}).get( |
| 142 | "since" |
| 143 | ) |
| 144 | expected_until = run_context.get("until") or run_context.get("crawl_window", {}).get( |
| 145 | "until" |
| 146 | ) |
| 147 | expected_week = run_context.get("week", "") |
| 148 | |
| 149 | for i, artifact in enumerate(artifacts): |
| 150 | ctx = artifact.get("run_context", {}) |
| 151 | source_id = artifact.get("source_id") or artifact.get("shard_id") or f"artifact[{i}]" |
| 152 | |
| 153 | # Check window |
| 154 | art_since = ctx.get("since") or ctx.get("crawl_window", {}).get("since") |
| 155 | art_until = ctx.get("until") or ctx.get("crawl_window", {}).get("until") |
| 156 | art_week = ctx.get("week", "") |
| 157 | |
| 158 | if art_week and art_week != expected_week: |
| 159 | errors.append(f"{source_id}: week mismatch ({art_week} vs {expected_week})") |
| 160 | if art_since and art_since != expected_since: |
| 161 | errors.append(f"{source_id}: since mismatch ({art_since} vs {expected_since})") |
| 162 | if art_until and art_until != expected_until: |
| 163 | errors.append(f"{source_id}: until mismatch ({art_until} vs {expected_until})") |
| 164 | |
| 165 | return errors |
| 166 | |
| 167 | |
| 168 | def validate_deterministic_ordering(articles: list[dict[str, Any]]) -> list[str]: |
| 169 | """Verify articles are in deterministic order (source_id, url).""" |
| 170 | errors: list[str] = [] |
| 171 | |
| 172 | for i in range(len(articles) - 1): |
| 173 | key_a = (articles[i].get("source", ""), articles[i].get("url", "")) |
| 174 | key_b = (articles[i + 1].get("source", ""), articles[i + 1].get("url", "")) |
| 175 | if key_a > key_b: |
| 176 | errors.append(f"non-deterministic ordering at index {i}: {key_a} > {key_b}") |
| 177 | break # One violation is enough to flag |
| 178 | |
| 179 | return errors |
| 180 | |
| 181 | |
| 182 | def detect_duplicate_urls(articles: list[dict[str, Any]]) -> list[str]: |
| 183 | """Find duplicate URLs across all articles.""" |
| 184 | seen: dict[str, int] = {} |
| 185 | duplicates: list[str] = [] |
| 186 | |
| 187 | for article in articles: |
| 188 | url = _normalize_url(article.get("url", "")) |
| 189 | if url in seen: |
| 190 | duplicates.append(url) |
| 191 | else: |
| 192 | seen[url] = 1 |
| 193 | |
| 194 | return duplicates |
| 195 | |
| 196 | |
| 197 | def detect_duplicate_repos(repositories: list[dict[str, Any]]) -> list[str]: |
| 198 | """Find duplicate repository full_names.""" |
| 199 | seen: set[str] = set() |
| 200 | duplicates: list[str] = [] |
| 201 | |
| 202 | for repo in repositories: |
| 203 | name = repo.get("full_name", "") |
| 204 | if name in seen: |
| 205 | duplicates.append(name) |
| 206 | else: |
| 207 | seen.add(name) |
| 208 | |
| 209 | return duplicates |
| 210 | |
| 211 | |
| 212 | def validate_stale_artifacts( |
| 213 | artifacts: list[dict[str, Any]], |
| 214 | reference_time: datetime | None = None, |
| 215 | max_age: timedelta = DEFAULT_MAX_ARTIFACT_AGE, |
| 216 | ) -> list[str]: |
| 217 | """Reject artifacts older than max_age from reference time.""" |
| 218 | stale: list[str] = [] |
| 219 | now = reference_time or datetime.now(UTC) |
| 220 | |
| 221 | for artifact in artifacts: |
| 222 | crawled_at = artifact.get("crawled_at") or artifact.get("created_at") |
| 223 | if not crawled_at: |
| 224 | continue |
| 225 | |
| 226 | try: |
| 227 | ts = datetime.fromisoformat(crawled_at.replace("Z", "+00:00")) |
| 228 | if (now - ts) > max_age: |
| 229 | source_id = artifact.get("source_id") or artifact.get("shard_id") or "unknown" |
| 230 | stale.append(f"{source_id}: artifact age {now - ts} exceeds max {max_age}") |
| 231 | except (ValueError, TypeError): |
| 232 | pass |
| 233 | |
| 234 | return stale |
| 235 | |
| 236 | |
| 237 | def validate_source_status( |
| 238 | artifacts: list[dict[str, Any]], |
| 239 | required_sources: list[str], |
| 240 | optional_sources: list[str] | None = None, |
| 241 | ) -> tuple[list[str], list[str]]: |
| 242 | """Validate source status metadata. |
| 243 | |
| 244 | Returns (errors, warnings): |
| 245 | - Errors for required sources that are missing or failed |
| 246 | - Warnings for optional sources that are missing or failed |
| 247 | """ |
| 248 | errors: list[str] = [] |
| 249 | warnings: list[str] = [] |
| 250 | optional = set(optional_sources or []) |
| 251 | |
| 252 | present_sources: dict[str, dict[str, Any]] = {} |
| 253 | for artifact in artifacts: |
| 254 | source_id = artifact.get("source_id") or artifact.get("shard_id", "") |
| 255 | present_sources[source_id] = artifact |
| 256 | |
| 257 | # Check required sources |
| 258 | for source in required_sources: |
| 259 | if source not in present_sources: |
| 260 | errors.append(f"required source '{source}' missing") |
| 261 | else: |
| 262 | status = present_sources[source].get("status", {}) |
| 263 | if isinstance(status, dict) and not status.get("success", True): |
| 264 | errors.append( |
| 265 | f"required source '{source}' failed: " |
| 266 | f"{status.get('error_message', 'unknown error')}" |
| 267 | ) |
| 268 | |
| 269 | # Check optional sources |
| 270 | for source in optional: |
| 271 | if source not in present_sources: |
| 272 | warnings.append(f"optional source '{source}' missing") |
| 273 | else: |
| 274 | status = present_sources[source].get("status", {}) |
| 275 | if isinstance(status, dict) and not status.get("success", True): |
| 276 | warnings.append( |
| 277 | f"optional source '{source}' degraded: {status.get('error_message', 'unknown')}" |
| 278 | ) |
| 279 | |
| 280 | return errors, warnings |
| 281 | |
| 282 | |
| 283 | def verify_byte_stability( |
| 284 | canonical_output: dict[str, Any], |
| 285 | reference_output: dict[str, Any], |
| 286 | exclude_fields: list[str] | None = None, |
| 287 | ) -> list[str]: |
| 288 | """Verify that canonical output is byte-stable compared to reference. |
| 289 | |
| 290 | Excludes documented timestamp fields from comparison. |
| 291 | """ |
| 292 | errors: list[str] = [] |
| 293 | exclude = set(exclude_fields or ["merged_at", "crawled_at", "created_at"]) |
| 294 | |
| 295 | def _strip_excluded(obj: Any) -> Any: |
| 296 | if isinstance(obj, dict): |
| 297 | return {k: _strip_excluded(v) for k, v in obj.items() if k not in exclude} |
| 298 | if isinstance(obj, list): |
| 299 | return [_strip_excluded(item) for item in obj] |
| 300 | return obj |
| 301 | |
| 302 | stripped_canonical = _strip_excluded(canonical_output) |
| 303 | stripped_reference = _strip_excluded(reference_output) |
| 304 | |
| 305 | canonical_json = json.dumps(stripped_canonical, sort_keys=True, separators=(",", ":")) |
| 306 | reference_json = json.dumps(stripped_reference, sort_keys=True, separators=(",", ":")) |
| 307 | |
| 308 | if canonical_json != reference_json: |
| 309 | errors.append("byte-stability violation: outputs differ (excluding timestamp fields)") |
| 310 | |
| 311 | return errors |
| 312 | |
| 313 | |
| 314 | def run_full_validation( |
| 315 | artifacts: list[dict[str, Any]], |
| 316 | run_context: RunContext | dict[str, Any], |
| 317 | *, |
| 318 | required_sources: list[str] | None = None, |
| 319 | optional_sources: list[str] | None = None, |
| 320 | expected_schema_version: str = "1", |
| 321 | max_artifact_age: timedelta = DEFAULT_MAX_ARTIFACT_AGE, |
| 322 | reference_time: datetime | None = None, |
| 323 | ) -> ValidationResult: |
| 324 | """Run the complete fan-in validation contract. |
| 325 | |
| 326 | This is the primary entry point for validating a set of crawl artifacts |
| 327 | before producing canonical merged output. |
| 328 | """ |
| 329 | result = ValidationResult(valid=True, artifact_count=len(artifacts)) |
| 330 | |
| 331 | if not artifacts: |
| 332 | result.valid = False |
| 333 | result.errors.append("no artifacts provided") |
| 334 | return result |
| 335 | |
| 336 | # 1. Schema validation |
| 337 | for artifact in artifacts: |
| 338 | schema_errors = validate_artifact_schema(artifact, expected_schema_version) |
| 339 | result.errors.extend(schema_errors) |
| 340 | |
| 341 | # 2. Checksum integrity |
| 342 | for artifact in artifacts: |
| 343 | checksum_errors = validate_checksum_integrity(artifact) |
| 344 | result.errors.extend(checksum_errors) |
| 345 | |
| 346 | # 3. Window consistency |
| 347 | window_errors = validate_window_consistency(artifacts, run_context) |
| 348 | result.errors.extend(window_errors) |
| 349 | |
| 350 | # 4. Stale cache rejection |
| 351 | stale = validate_stale_artifacts(artifacts, reference_time, max_artifact_age) |
| 352 | result.stale_artifacts = stale |
| 353 | result.errors.extend(stale) |
| 354 | |
| 355 | # 5. Source status metadata |
| 356 | req_sources = required_sources or [] |
| 357 | opt_sources = optional_sources or [] |
| 358 | source_errors, source_warnings = validate_source_status(artifacts, req_sources, opt_sources) |
| 359 | result.errors.extend(source_errors) |
| 360 | result.warnings.extend(source_warnings) |
| 361 | |
| 362 | # 6. Track present/missing sources |
| 363 | result.sources_present = [ |
| 364 | a.get("source_id") or a.get("shard_id") or "unknown" for a in artifacts |
| 365 | ] |
| 366 | result.sources_missing_required = [s for s in req_sources if s not in result.sources_present] |
| 367 | result.sources_missing_optional = [s for s in opt_sources if s not in result.sources_present] |
| 368 | |
| 369 | # 7. Duplicate detection |
| 370 | all_articles = [] |
| 371 | all_repos = [] |
| 372 | for artifact in artifacts: |
| 373 | all_articles.extend(artifact.get("articles", [])) |
| 374 | all_repos.extend(artifact.get("repositories", [])) |
| 375 | |
| 376 | if all_articles: |
| 377 | result.duplicate_urls = detect_duplicate_urls(all_articles) |
| 378 | if result.duplicate_urls: |
| 379 | result.warnings.append( |
| 380 | f"duplicate URLs detected ({len(result.duplicate_urls)}): deduplication will apply" |
| 381 | ) |
| 382 | |
| 383 | if all_repos: |
| 384 | result.duplicate_repos = detect_duplicate_repos(all_repos) |
| 385 | if result.duplicate_repos: |
| 386 | result.warnings.append( |
| 387 | f"duplicate repos detected ({len(result.duplicate_repos)}): " |
| 388 | f"deduplication will apply" |
| 389 | ) |
| 390 | |
| 391 | # Final verdict |
| 392 | result.valid = len(result.errors) == 0 |
| 393 | return result |
| 394 | |
| 395 | |
| 396 | def _normalize_url(url: str) -> str: |
| 397 | """Normalize URL for deduplication: scheme + host + path (strip query).""" |
| 398 | from urllib.parse import urlparse |
| 399 | |
| 400 | parsed = urlparse(url) |
| 401 | return f"{parsed.scheme}://{parsed.netloc}{parsed.path}".lower() |