| 1 | #!/usr/bin/env python3 |
| 2 | """Deterministic map/reduce analysis dry-run scaffolding. |
| 3 | |
| 4 | This module intentionally performs no live AI calls and never writes to published |
| 5 | content paths. It emits candidate-only artifacts under an explicit output |
| 6 | folder so the contracts can be validated before any future promotion work. |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | import argparse |
| 12 | import hashlib |
| 13 | import json |
| 14 | import re |
| 15 | import sys |
| 16 | import time |
| 17 | from dataclasses import asdict, dataclass |
| 18 | from pathlib import Path |
| 19 | from typing import Any |
| 20 | |
| 21 | try: |
| 22 | from scripts.analysis_gate import validate_analysis, validate_publish_quality |
| 23 | from scripts.analyze_fallback import find_previous_summary |
| 24 | from scripts.model_pricing import estimate_cost_usd |
| 25 | from scripts.observability_metrics import ( |
| 26 | DEFAULT_OBSERVABILITY_DIR, |
| 27 | METRICS_SCHEMA_VERSION, |
| 28 | AnalysisMetrics, |
| 29 | MapReduceMetrics, |
| 30 | ObservabilityLedger, |
| 31 | emit_ledger, |
| 32 | ) |
| 33 | from scripts.render_press_context import estimate_tokens |
| 34 | from scripts.sanitize_repo_content import sanitize_repo_payload |
| 35 | except ModuleNotFoundError: # pragma: no cover - script execution path |
| 36 | sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) |
| 37 | from scripts.analysis_gate import validate_analysis, validate_publish_quality |
| 38 | from scripts.analyze_fallback import find_previous_summary |
| 39 | from scripts.model_pricing import estimate_cost_usd |
| 40 | from scripts.observability_metrics import ( |
| 41 | DEFAULT_OBSERVABILITY_DIR, |
| 42 | METRICS_SCHEMA_VERSION, |
| 43 | AnalysisMetrics, |
| 44 | MapReduceMetrics, |
| 45 | ObservabilityLedger, |
| 46 | emit_ledger, |
| 47 | ) |
| 48 | from scripts.render_press_context import estimate_tokens |
| 49 | from scripts.sanitize_repo_content import sanitize_repo_payload |
| 50 | |
| 51 | ROOT = Path(__file__).resolve().parent.parent |
| 52 | MAP_SCHEMA = "analysis_map_v1" |
| 53 | PLAN_SCHEMA = "analysis_editorial_plan_v1" |
| 54 | QA_SCHEMA = "analysis_map_reduce_qa_v1" |
| 55 | CANDIDATE_DISCLAIMER = "Map/reduce dry-run candidate only; not publish eligible." |
| 56 | MAPPER_IDS = ("new_repos", "trending_repos", "press_correlations", "prior_continuity") |
| 57 | TOKEN_ESTIMATE_KEY = "_".join(("token", "estimate")) |
| 58 | SECTION_ORDER = [ |
| 59 | "This Week's Trends", |
| 60 | "Where Industry Meets Code", |
| 61 | "Signal & Noise", |
| 62 | "Blind Spots", |
| 63 | "The Week Ahead", |
| 64 | ] |
| 65 | |
| 66 | |
| 67 | @dataclass(frozen=True) |
| 68 | class ArtifactRef: |
| 69 | path: str |
| 70 | sha256: str | None |
| 71 | bytes: int |
| 72 | |
| 73 | |
| 74 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 75 | parser = argparse.ArgumentParser( |
| 76 | description="Create deterministic candidate-only map/reduce analysis artifacts." |
| 77 | ) |
| 78 | parser.add_argument( |
| 79 | "--raw-json", required=True, type=Path, help="Canonical weekly raw GitHub crawl payload." |
| 80 | ) |
| 81 | parser.add_argument( |
| 82 | "--output-dir", required=True, type=Path, help="Candidate artifact output directory." |
| 83 | ) |
| 84 | parser.add_argument( |
| 85 | "--current-datetime", required=True, help="ISO-8601 timestamp for the dry run." |
| 86 | ) |
| 87 | parser.add_argument("--run-id", default="local", help="Stable run id to include in contracts.") |
| 88 | parser.add_argument( |
| 89 | "--press-context", type=Path, help="Rendered press context markdown, if available." |
| 90 | ) |
| 91 | parser.add_argument("--analyzed-dir", type=Path, default=ROOT / "data" / "analyzed") |
| 92 | parser.add_argument( |
| 93 | "--baseline-summary", |
| 94 | type=Path, |
| 95 | help="Optional current single-pass summary for QA comparison.", |
| 96 | ) |
| 97 | parser.add_argument("--max-repos-per-ledger", type=int, default=10) |
| 98 | parser.add_argument("--analysis-source", default="map-reduce-dry-run") |
| 99 | parser.add_argument("--analysis-model", default="local-deterministic") |
| 100 | return parser.parse_args(argv) |
| 101 | |
| 102 | |
| 103 | def load_json(path: Path) -> dict[str, Any]: |
| 104 | payload = json.loads(path.read_text(encoding="utf-8")) |
| 105 | if not isinstance(payload, dict): |
| 106 | raise ValueError(f"JSON payload must be an object: {path}") |
| 107 | return payload |
| 108 | |
| 109 | |
| 110 | def sha256_bytes(data: bytes) -> str: |
| 111 | return hashlib.sha256(data).hexdigest() |
| 112 | |
| 113 | |
| 114 | def stable_json(payload: Any) -> str: |
| 115 | return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n" |
| 116 | |
| 117 | |
| 118 | def write_json(path: Path, payload: Any) -> None: |
| 119 | path.parent.mkdir(parents=True, exist_ok=True) |
| 120 | path.write_text(stable_json(payload), encoding="utf-8") |
| 121 | |
| 122 | |
| 123 | def metric_cost_usd(model: str, input_tokens: int, output_tokens: int) -> float: |
| 124 | return estimate_cost_usd(model, input_tokens, output_tokens) or 0.0 |
| 125 | |
| 126 | |
| 127 | def file_ref(path: Path | None) -> ArtifactRef | None: |
| 128 | if path is None or not path.exists() or not path.is_file(): |
| 129 | return None |
| 130 | data = path.read_bytes() |
| 131 | return ArtifactRef(path=path.as_posix(), sha256=sha256_bytes(data), bytes=len(data)) |
| 132 | |
| 133 | |
| 134 | def collect_gate_failure_reasons(qa_report: dict[str, Any]) -> list[str]: |
| 135 | reasons: list[str] = [] |
| 136 | checks = qa_report.get("checks", {}) if isinstance(qa_report.get("checks"), dict) else {} |
| 137 | mapper_contracts = checks.get("mapper_contracts", {}) |
| 138 | if isinstance(mapper_contracts, dict): |
| 139 | errors_by_mapper = mapper_contracts.get("errors_by_mapper", {}) |
| 140 | if isinstance(errors_by_mapper, dict): |
| 141 | for mapper, errors in sorted(errors_by_mapper.items()): |
| 142 | if isinstance(errors, list): |
| 143 | reasons.extend(f"{mapper}: {error}" for error in errors) |
| 144 | for key in ( |
| 145 | "structural_analysis_gate", |
| 146 | "evidence_and_editorial_gates", |
| 147 | "publish_provenance_gate", |
| 148 | ): |
| 149 | check = checks.get(key, {}) |
| 150 | if ( |
| 151 | isinstance(check, dict) |
| 152 | and check.get("expected_failure") is not True |
| 153 | and isinstance(check.get("errors"), list) |
| 154 | ): |
| 155 | reasons.extend(str(error) for error in check["errors"] if error) |
| 156 | if isinstance(qa_report.get("regressions"), list): |
| 157 | reasons.extend(str(error) for error in qa_report["regressions"] if error) |
| 158 | sidecars = checks.get("sidecars_present", {}) |
| 159 | if isinstance(sidecars, dict) and sidecars.get("passed") is False: |
| 160 | reasons.append("sidecars_present failed") |
| 161 | seen: set[str] = set() |
| 162 | ordered: list[str] = [] |
| 163 | for reason in reasons: |
| 164 | normalized = reason.strip() |
| 165 | if normalized and normalized not in seen: |
| 166 | seen.add(normalized) |
| 167 | ordered.append(normalized) |
| 168 | return ordered |
| 169 | |
| 170 | |
| 171 | def normalize_repo_name(repo: dict[str, Any]) -> str: |
| 172 | full_name = str(repo.get("full_name") or "").strip() |
| 173 | if full_name: |
| 174 | return full_name |
| 175 | owner = str(repo.get("owner") or "").strip() |
| 176 | name = str(repo.get("name") or "").strip() |
| 177 | return f"{owner}/{name}" if owner and name else name |
| 178 | |
| 179 | |
| 180 | def repo_url(repo: dict[str, Any], full_name: str) -> str: |
| 181 | return str(repo.get("url") or repo.get("html_url") or f"https://github.com/{full_name}") |
| 182 | |
| 183 | |
| 184 | def repo_description(repo: dict[str, Any]) -> str: |
| 185 | desc = str(repo.get("description") or "No description provided").strip() |
| 186 | return re.sub(r"\s+", " ", desc)[:220] |
| 187 | |
| 188 | |
| 189 | def repo_claim_key(repo: dict[str, Any], *, mapper: str) -> str: |
| 190 | full_name = normalize_repo_name(repo) |
| 191 | base = f"{mapper}:{full_name}:{repo.get('stars', 0)}:{repo.get('stars_gained', repo.get('gained', 0))}" |
| 192 | return hashlib.sha256(base.encode("utf-8")).hexdigest()[:12] |
| 193 | |
| 194 | |
| 195 | def sorted_repos(repos: list[dict[str, Any]], *, mode: str) -> list[dict[str, Any]]: |
| 196 | if mode == "trending": |
| 197 | return sorted( |
| 198 | repos, |
| 199 | key=lambda r: ( |
| 200 | int(r.get("stars_gained") or r.get("gained") or 0), |
| 201 | int(r.get("stars") or 0), |
| 202 | normalize_repo_name(r), |
| 203 | ), |
| 204 | reverse=True, |
| 205 | ) |
| 206 | return sorted( |
| 207 | repos, key=lambda r: (int(r.get("stars") or 0), normalize_repo_name(r)), reverse=True |
| 208 | ) |
| 209 | |
| 210 | |
| 211 | def coverage_for_repos( |
| 212 | repos: list[dict[str, Any]], input_count: int, *, omitted_reason: str |
| 213 | ) -> dict[str, Any]: |
| 214 | seen = [normalize_repo_name(repo) for repo in repos if normalize_repo_name(repo)] |
| 215 | omitted = max(0, input_count - len(seen)) |
| 216 | return { |
| 217 | "repo_ids_seen": seen, |
| 218 | "article_urls_seen": [], |
| 219 | "repo_count_input": input_count, |
| 220 | "repo_count_mapped": len(seen), |
| 221 | "article_count_input": 0, |
| 222 | "article_count_mapped": 0, |
| 223 | "excluded_reason_counts": {omitted_reason: omitted} if omitted else {}, |
| 224 | } |
| 225 | |
| 226 | |
| 227 | def make_repo_finding( |
| 228 | repo: dict[str, Any], *, mapper: str, category: str, role: str |
| 229 | ) -> dict[str, Any]: |
| 230 | full_name = normalize_repo_name(repo) |
| 231 | stars = int(repo.get("stars") or 0) |
| 232 | gained = int(repo.get("stars_gained") or repo.get("gained") or 0) |
| 233 | language = repo.get("language") or "unknown language" |
| 234 | topics = repo.get("topics") if isinstance(repo.get("topics"), list) else [] |
| 235 | topic_note = f" with topics {', '.join(str(t) for t in topics[:3])}" if topics else "" |
| 236 | metric_note = f"{stars:,} stars" + (f", {gained:,} gained" if gained else "") |
| 237 | claim = f"{full_name} is a {role} {language} signal this week ({metric_note}){topic_note}: {repo_description(repo)}" |
| 238 | confidence = 0.74 if mapper == "trending_repos" and gained else 0.68 |
| 239 | return { |
| 240 | "claim_id": f"{mapper}-{repo_claim_key(repo, mapper=mapper)}", |
| 241 | "claim": claim, |
| 242 | "category": category, |
| 243 | "source_type": "github", |
| 244 | "evidence_refs": [ |
| 245 | { |
| 246 | "type": "repo", |
| 247 | "ref": full_name, |
| 248 | "url": repo_url(repo, full_name), |
| 249 | "full_name": full_name, |
| 250 | "description": repo.get("description"), |
| 251 | "language": repo.get("language"), |
| 252 | "topics": topics, |
| 253 | "stars": stars, |
| 254 | "stars_gained": gained, |
| 255 | "created_at": repo.get("created_at"), |
| 256 | "role": "anchor", |
| 257 | "evidence_note": f"Crawler metrics show {metric_note}.", |
| 258 | } |
| 259 | ], |
| 260 | "repo_full_name": full_name, |
| 261 | "news_url": None, |
| 262 | "confidence": confidence, |
| 263 | "contra_refs": [], |
| 264 | "uncertainties": [] if gained else ["stars_gained unavailable or zero in raw payload"], |
| 265 | "quality_flags": ["dry_run_local_mapper"], |
| 266 | } |
| 267 | |
| 268 | |
| 269 | def base_map_payload( |
| 270 | *, |
| 271 | run_id: str, |
| 272 | week: str, |
| 273 | shard_id: str, |
| 274 | input_refs: list[str], |
| 275 | repo_count: int, |
| 276 | article_count: int, |
| 277 | token_estimate: int, |
| 278 | ) -> dict[str, Any]: |
| 279 | return { |
| 280 | "schema_version": MAP_SCHEMA, |
| 281 | "run_id": run_id, |
| 282 | "week": week, |
| 283 | "shard_id": f"signal-type:{shard_id}", |
| 284 | "slice": { |
| 285 | "strategy": "signal_type", |
| 286 | "input_refs": input_refs, |
| 287 | "input_token_estimate": token_estimate, |
| 288 | "repo_count": repo_count, |
| 289 | "article_count": article_count, |
| 290 | }, |
| 291 | "coverage": {}, |
| 292 | "findings": [], |
| 293 | "citations": [], |
| 294 | "reference_candidates": {"notable_projects": [], "press_articles": []}, |
| 295 | TOKEN_ESTIMATE_KEY: 0, |
| 296 | "model": "none", |
| 297 | "status": "success", |
| 298 | "errors": [], |
| 299 | "provenance": {}, |
| 300 | } |
| 301 | |
| 302 | |
| 303 | def map_repositories( |
| 304 | *, |
| 305 | run_id: str, |
| 306 | week: str, |
| 307 | raw_path: Path, |
| 308 | raw_ref: ArtifactRef, |
| 309 | shard_id: str, |
| 310 | repos: list[dict[str, Any]], |
| 311 | mode: str, |
| 312 | max_repos: int, |
| 313 | ) -> dict[str, Any]: |
| 314 | selected = sorted_repos(repos, mode=mode)[:max_repos] |
| 315 | payload = base_map_payload( |
| 316 | run_id=run_id, |
| 317 | week=week, |
| 318 | shard_id=shard_id, |
| 319 | input_refs=[f"{raw_path.as_posix()}#{shard_id}[0:{len(repos)}]"], |
| 320 | repo_count=len(repos), |
| 321 | article_count=0, |
| 322 | token_estimate=estimate_tokens(stable_json(repos)), |
| 323 | ) |
| 324 | category = "trend" if shard_id == "new_repos" else "signal" |
| 325 | role = "new-repository" if shard_id == "new_repos" else "momentum" |
| 326 | findings = [ |
| 327 | make_repo_finding(repo, mapper=shard_id, category=category, role=role) for repo in selected |
| 328 | ] |
| 329 | payload["findings"] = findings |
| 330 | payload["coverage"] = coverage_for_repos( |
| 331 | selected, len(repos), omitted_reason="outside_dry_run_top_repo_limit" |
| 332 | ) |
| 333 | payload["citations"] = [ |
| 334 | {"type": "repo", "url": item["evidence_refs"][0]["url"], "title": item["repo_full_name"]} |
| 335 | for item in findings |
| 336 | ] |
| 337 | payload["reference_candidates"] = { |
| 338 | "notable_projects": [item["repo_full_name"] for item in findings], |
| 339 | "press_articles": [], |
| 340 | } |
| 341 | payload["token_estimate"] = estimate_tokens(stable_json(payload)) |
| 342 | payload["provenance"] = {"raw_json": asdict(raw_ref), "deterministic_mapper": True} |
| 343 | return payload |
| 344 | |
| 345 | |
| 346 | def extract_press_articles(press_context: str) -> list[dict[str, str]]: |
| 347 | urls = [] |
| 348 | for match in re.finditer(r"https?://[^\s)\]]+", press_context): |
| 349 | url = match.group(0).rstrip(".,") |
| 350 | if url not in urls: |
| 351 | urls.append(url) |
| 352 | articles: list[dict[str, str]] = [] |
| 353 | lines = [line.strip(" -*") for line in press_context.splitlines() if line.strip()] |
| 354 | for url in urls[:10]: |
| 355 | title = next((line[:120] for line in lines if url in line), url) |
| 356 | articles.append({"url": url, "title": title}) |
| 357 | return articles |
| 358 | |
| 359 | |
| 360 | def map_press( |
| 361 | *, |
| 362 | run_id: str, |
| 363 | week: str, |
| 364 | press_path: Path | None, |
| 365 | press_ref: ArtifactRef | None, |
| 366 | raw_ref: ArtifactRef, |
| 367 | ) -> dict[str, Any]: |
| 368 | content = press_path.read_text(encoding="utf-8") if press_path and press_path.exists() else "" |
| 369 | articles = extract_press_articles(content) |
| 370 | payload = base_map_payload( |
| 371 | run_id=run_id, |
| 372 | week=week, |
| 373 | shard_id="press_correlations", |
| 374 | input_refs=[press_path.as_posix() if press_path else "press_context:none"], |
| 375 | repo_count=0, |
| 376 | article_count=len(articles), |
| 377 | token_estimate=estimate_tokens(content), |
| 378 | ) |
| 379 | findings = [] |
| 380 | for index, article in enumerate(articles[:5], start=1): |
| 381 | claim_id = hashlib.sha256(f"press:{article['url']}".encode("utf-8")).hexdigest()[:12] |
| 382 | findings.append( |
| 383 | { |
| 384 | "claim_id": f"press_correlations-{claim_id}", |
| 385 | "claim": f"Retained press context cites {article['title']} as industry evidence to compare against repository activity.", |
| 386 | "category": "press_correlation", |
| 387 | "source_type": "news", |
| 388 | "evidence_refs": [ |
| 389 | { |
| 390 | "type": "article", |
| 391 | "ref": article["url"], |
| 392 | "url": article["url"], |
| 393 | "role": "supporting", |
| 394 | "evidence_note": "URL was retained in rendered press context.", |
| 395 | } |
| 396 | ], |
| 397 | "repo_full_name": None, |
| 398 | "news_url": article["url"], |
| 399 | "confidence": 0.66, |
| 400 | "contra_refs": [], |
| 401 | "uncertainties": ["dry-run mapper does not infer unstated press sentiment"], |
| 402 | "quality_flags": ["dry_run_local_mapper"], |
| 403 | } |
| 404 | ) |
| 405 | if not findings: |
| 406 | payload["status"] = "partial" |
| 407 | payload["errors"] = ["No press URLs were available; mapper emitted coverage-only ledger."] |
| 408 | payload["findings"] = findings |
| 409 | payload["coverage"] = { |
| 410 | "repo_ids_seen": [], |
| 411 | "article_urls_seen": [article["url"] for article in articles], |
| 412 | "repo_count_input": 0, |
| 413 | "repo_count_mapped": 0, |
| 414 | "article_count_input": len(articles), |
| 415 | "article_count_mapped": len(articles[:5]), |
| 416 | "excluded_reason_counts": {"outside_dry_run_article_limit": max(0, len(articles) - 5)} |
| 417 | if len(articles) > 5 |
| 418 | else {}, |
| 419 | } |
| 420 | payload["citations"] = [ |
| 421 | {"type": "article", "url": item["news_url"], "title": item["claim"][:80]} |
| 422 | for item in findings |
| 423 | ] |
| 424 | payload["reference_candidates"] = { |
| 425 | "notable_projects": [], |
| 426 | "press_articles": [item["news_url"] for item in findings], |
| 427 | } |
| 428 | payload["token_estimate"] = estimate_tokens(stable_json(payload)) |
| 429 | payload["provenance"] = { |
| 430 | "raw_json": asdict(raw_ref), |
| 431 | "press_context": asdict(press_ref) if press_ref else None, |
| 432 | "deterministic_mapper": True, |
| 433 | } |
| 434 | return payload |
| 435 | |
| 436 | |
| 437 | def map_prior( |
| 438 | *, |
| 439 | run_id: str, |
| 440 | week: str, |
| 441 | previous_summary: Path | None, |
| 442 | previous_ref: ArtifactRef | None, |
| 443 | raw_ref: ArtifactRef, |
| 444 | ) -> dict[str, Any]: |
| 445 | content = ( |
| 446 | previous_summary.read_text(encoding="utf-8") |
| 447 | if previous_summary and previous_summary.exists() |
| 448 | else "" |
| 449 | ) |
| 450 | payload = base_map_payload( |
| 451 | run_id=run_id, |
| 452 | week=week, |
| 453 | shard_id="prior_continuity", |
| 454 | input_refs=[previous_summary.as_posix() if previous_summary else "prior_summary:none"], |
| 455 | repo_count=0, |
| 456 | article_count=0, |
| 457 | token_estimate=estimate_tokens(content), |
| 458 | ) |
| 459 | finding = { |
| 460 | "claim_id": f"prior_continuity-{hashlib.sha256((previous_summary.as_posix() if previous_summary else 'none').encode()).hexdigest()[:12]}", |
| 461 | "claim": ( |
| 462 | "Prior weekly analysis is available for continuity checks; reducer should compare carried-forward claims against this week's evidence." |
| 463 | if content |
| 464 | else "No prior weekly analysis was available, so continuity claims should be treated as open blind spots." |
| 465 | ), |
| 466 | "category": "continuity", |
| 467 | "source_type": "prior_summary", |
| 468 | "evidence_refs": [ |
| 469 | { |
| 470 | "type": "prior_summary", |
| 471 | "ref": previous_summary.as_posix() if previous_summary else "none", |
| 472 | "url": previous_summary.as_posix() if previous_summary else "none", |
| 473 | "role": "supporting", |
| 474 | "evidence_note": "Deterministic local continuity marker.", |
| 475 | } |
| 476 | ], |
| 477 | "repo_full_name": None, |
| 478 | "news_url": None, |
| 479 | "confidence": 0.55 if content else 0.35, |
| 480 | "contra_refs": [], |
| 481 | "uncertainties": [] if content else ["no prior summary artifact found"], |
| 482 | "quality_flags": ["dry_run_local_mapper"], |
| 483 | } |
| 484 | payload["findings"] = [finding] |
| 485 | payload["coverage"] = { |
| 486 | "repo_ids_seen": [], |
| 487 | "article_urls_seen": [], |
| 488 | "repo_count_input": 0, |
| 489 | "repo_count_mapped": 0, |
| 490 | "article_count_input": 0, |
| 491 | "article_count_mapped": 0, |
| 492 | "excluded_reason_counts": {}, |
| 493 | "prior_summary_present": bool(content), |
| 494 | } |
| 495 | payload["citations"] = [ |
| 496 | { |
| 497 | "type": "prior_summary", |
| 498 | "url": finding["evidence_refs"][0]["url"], |
| 499 | "title": "prior weekly summary", |
| 500 | } |
| 501 | ] |
| 502 | payload["reference_candidates"] = {"notable_projects": [], "press_articles": []} |
| 503 | payload["token_estimate"] = estimate_tokens(stable_json(payload)) |
| 504 | payload["provenance"] = { |
| 505 | "raw_json": asdict(raw_ref), |
| 506 | "prior_summary": asdict(previous_ref) if previous_ref else None, |
| 507 | "deterministic_mapper": True, |
| 508 | } |
| 509 | return payload |
| 510 | |
| 511 | |
| 512 | def validate_map(payload: dict[str, Any]) -> list[str]: |
| 513 | errors: list[str] = [] |
| 514 | if payload.get("schema_version") != MAP_SCHEMA: |
| 515 | errors.append("mapper schema_version mismatch") |
| 516 | for field in ( |
| 517 | "run_id", |
| 518 | "week", |
| 519 | "shard_id", |
| 520 | "slice", |
| 521 | "coverage", |
| 522 | "findings", |
| 523 | "citations", |
| 524 | "reference_candidates", |
| 525 | "provenance", |
| 526 | ): |
| 527 | if field not in payload: |
| 528 | errors.append(f"mapper missing {field}") |
| 529 | if "findings" in payload and not isinstance(payload.get("findings"), list): |
| 530 | errors.append("findings must be a list") |
| 531 | findings = payload.get("findings") if isinstance(payload.get("findings"), list) else [] |
| 532 | for index, finding in enumerate(findings): |
| 533 | if not isinstance(finding, dict): |
| 534 | errors.append(f"finding {index} must be an object") |
| 535 | continue |
| 536 | for field in ( |
| 537 | "claim_id", |
| 538 | "claim", |
| 539 | "category", |
| 540 | "source_type", |
| 541 | "evidence_refs", |
| 542 | "confidence", |
| 543 | "contra_refs", |
| 544 | "uncertainties", |
| 545 | ): |
| 546 | if field not in finding: |
| 547 | errors.append(f"finding {index} missing {field}") |
| 548 | refs = finding.get("evidence_refs") |
| 549 | if not isinstance(refs, list) or not refs: |
| 550 | errors.append(f"finding {index} has no evidence refs") |
| 551 | else: |
| 552 | for ref in refs: |
| 553 | if ( |
| 554 | not isinstance(ref, dict) |
| 555 | or not ref.get("type") |
| 556 | or not ref.get("ref") |
| 557 | or not ref.get("url") |
| 558 | ): |
| 559 | errors.append(f"finding {index} has malformed evidence ref") |
| 560 | confidence = finding.get("confidence") |
| 561 | if not isinstance(confidence, (int, float)) or not (0 <= float(confidence) <= 1): |
| 562 | errors.append(f"finding {index} confidence out of range") |
| 563 | if "contra_refs" in finding and not isinstance(finding.get("contra_refs"), list): |
| 564 | errors.append(f"finding {index} contra_refs must be a list") |
| 565 | coverage = payload.get("coverage") |
| 566 | if not isinstance(coverage, dict): |
| 567 | errors.append("coverage must be an object") |
| 568 | else: |
| 569 | for key in ("repo_ids_seen", "article_urls_seen", "excluded_reason_counts"): |
| 570 | if key not in coverage: |
| 571 | errors.append(f"coverage missing {key}") |
| 572 | if "repo_ids_seen" in coverage and not isinstance(coverage.get("repo_ids_seen"), list): |
| 573 | errors.append("coverage repo_ids_seen must be a list") |
| 574 | if "article_urls_seen" in coverage and not isinstance( |
| 575 | coverage.get("article_urls_seen"), list |
| 576 | ): |
| 577 | errors.append("coverage article_urls_seen must be a list") |
| 578 | excluded = coverage.get("excluded_reason_counts") |
| 579 | if "excluded_reason_counts" in coverage and not isinstance(excluded, dict): |
| 580 | errors.append("coverage excluded_reason_counts must be an object") |
| 581 | elif isinstance(excluded, dict): |
| 582 | for reason, count in excluded.items(): |
| 583 | if not reason or not isinstance(count, int) or count < 0: |
| 584 | errors.append( |
| 585 | "coverage excluded_reason_counts must contain non-negative integer counts" |
| 586 | ) |
| 587 | break |
| 588 | for prefix in ("repo", "article"): |
| 589 | input_key = f"{prefix}_count_input" |
| 590 | mapped_key = f"{prefix}_count_mapped" |
| 591 | if input_key in coverage or mapped_key in coverage: |
| 592 | input_count = coverage.get(input_key) |
| 593 | mapped_count = coverage.get(mapped_key) |
| 594 | if ( |
| 595 | not isinstance(input_count, int) |
| 596 | or not isinstance(mapped_count, int) |
| 597 | or input_count < 0 |
| 598 | or mapped_count < 0 |
| 599 | ): |
| 600 | errors.append(f"coverage {prefix} counts must be non-negative integers") |
| 601 | continue |
| 602 | if mapped_count > input_count: |
| 603 | errors.append(f"coverage {mapped_key} exceeds {input_key}") |
| 604 | if mapped_count < input_count and not coverage.get("excluded_reason_counts"): |
| 605 | errors.append( |
| 606 | f"coverage {mapped_key} below {input_key} without excluded reasons" |
| 607 | ) |
| 608 | status = payload.get("status") |
| 609 | if status == "failed": |
| 610 | errors.append("mapper status failed") |
| 611 | return errors |
| 612 | |
| 613 | |
| 614 | def normalized_claim_key(finding: dict[str, Any]) -> str: |
| 615 | repo = finding.get("repo_full_name") or "" |
| 616 | article = finding.get("news_url") or "" |
| 617 | claim = str(finding.get("claim") or "").lower() |
| 618 | words = "-".join(re.findall(r"[a-z0-9]+", claim)[:8]) |
| 619 | return f"{finding.get('category')}:{repo or article or words}" |
| 620 | |
| 621 | |
| 622 | def contra_ref_targets(contra_refs: Any) -> set[str]: |
| 623 | targets: set[str] = set() |
| 624 | if not isinstance(contra_refs, list): |
| 625 | return targets |
| 626 | for ref in contra_refs: |
| 627 | if isinstance(ref, str) and ref: |
| 628 | targets.add(ref) |
| 629 | elif isinstance(ref, dict): |
| 630 | for key in ("claim_id", "ref", "url"): |
| 631 | value = ref.get(key) |
| 632 | if value: |
| 633 | targets.add(str(value)) |
| 634 | return targets |
| 635 | |
| 636 | |
| 637 | def contradiction_record( |
| 638 | finding: dict[str, Any], |
| 639 | ledger: dict[str, Any], |
| 640 | *, |
| 641 | contradicted_by: list[str], |
| 642 | ) -> dict[str, Any]: |
| 643 | contra_refs = finding.get("contra_refs") if isinstance(finding.get("contra_refs"), list) else [] |
| 644 | return { |
| 645 | "claim_id": finding.get("claim_id"), |
| 646 | "claim": finding.get("claim"), |
| 647 | "source_shard": ledger.get("shard_id"), |
| 648 | "normalized_claim_key": normalized_claim_key(finding), |
| 649 | "evidence_refs": finding.get("evidence_refs") |
| 650 | if isinstance(finding.get("evidence_refs"), list) |
| 651 | else [], |
| 652 | "contra_refs": contra_refs, |
| 653 | "contradicted_by": sorted(set(contradicted_by)), |
| 654 | "resolution": "rejected_unresolved", |
| 655 | "reason": "Unresolved contradiction refs are preserved for audit and excluded from selected editorial material.", |
| 656 | } |
| 657 | |
| 658 | |
| 659 | def reduce_ledgers( |
| 660 | ledgers: list[dict[str, Any]], *, raw_payload: dict[str, Any] |
| 661 | ) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]: |
| 662 | selected: list[dict[str, Any]] = [] |
| 663 | rejected: list[dict[str, Any]] = [] |
| 664 | contradictions: list[dict[str, Any]] = [] |
| 665 | seen_keys: dict[str, dict[str, Any]] = {} |
| 666 | section_by_category = { |
| 667 | "trend": "This Week's Trends", |
| 668 | "signal": "Signal & Noise", |
| 669 | "noise": "Signal & Noise", |
| 670 | "gap": "Blind Spots", |
| 671 | "press_correlation": "Where Industry Meets Code", |
| 672 | "press_divergence": "Where Industry Meets Code", |
| 673 | "continuity": "The Week Ahead", |
| 674 | } |
| 675 | inbound_contradictions: dict[str, list[str]] = {} |
| 676 | for ledger in ledgers: |
| 677 | for finding in ( |
| 678 | ledger.get("findings", []) if isinstance(ledger.get("findings"), list) else [] |
| 679 | ): |
| 680 | if not isinstance(finding, dict): |
| 681 | continue |
| 682 | source_claim_id = finding.get("claim_id") |
| 683 | for target in contra_ref_targets(finding.get("contra_refs")): |
| 684 | inbound_contradictions.setdefault(target, []).append(str(source_claim_id)) |
| 685 | for ledger in ledgers: |
| 686 | ledger_findings = ( |
| 687 | ledger.get("findings", []) if isinstance(ledger.get("findings"), list) else [] |
| 688 | ) |
| 689 | for finding in ledger_findings: |
| 690 | if not isinstance(finding, dict): |
| 691 | rejected.append( |
| 692 | { |
| 693 | "claim_id": None, |
| 694 | "reason": "malformed_finding", |
| 695 | "source_shard": ledger.get("shard_id"), |
| 696 | } |
| 697 | ) |
| 698 | continue |
| 699 | refs = ( |
| 700 | finding.get("evidence_refs") |
| 701 | if isinstance(finding.get("evidence_refs"), list) |
| 702 | else [] |
| 703 | ) |
| 704 | contra_refs = ( |
| 705 | finding.get("contra_refs") if isinstance(finding.get("contra_refs"), list) else [] |
| 706 | ) |
| 707 | contradicted_by = inbound_contradictions.get(str(finding.get("claim_id")), []) |
| 708 | if contra_refs or contradicted_by: |
| 709 | contradictions.append( |
| 710 | contradiction_record(finding, ledger, contradicted_by=contradicted_by) |
| 711 | ) |
| 712 | rejected.append( |
| 713 | { |
| 714 | "claim_id": finding.get("claim_id"), |
| 715 | "reason": "unresolved_contradiction", |
| 716 | "source_shard": ledger.get("shard_id"), |
| 717 | } |
| 718 | ) |
| 719 | continue |
| 720 | if not refs: |
| 721 | rejected.append( |
| 722 | { |
| 723 | "claim_id": finding.get("claim_id"), |
| 724 | "reason": "weak_citation", |
| 725 | "source_shard": ledger.get("shard_id"), |
| 726 | } |
| 727 | ) |
| 728 | continue |
| 729 | key = normalized_claim_key(finding) |
| 730 | if key in seen_keys: |
| 731 | existing = seen_keys[key] |
| 732 | existing["merged_from"].append(finding["claim_id"]) |
| 733 | existing["citation_bindings"]["repos"].extend( |
| 734 | [r.get("ref") for r in refs if r.get("type") == "repo"] |
| 735 | ) |
| 736 | existing["citation_bindings"]["articles"].extend( |
| 737 | [r.get("url") for r in refs if r.get("type") == "article"] |
| 738 | ) |
| 739 | rejected.append( |
| 740 | { |
| 741 | "claim_id": finding.get("claim_id"), |
| 742 | "reason": "duplicate", |
| 743 | "source_shard": ledger.get("shard_id"), |
| 744 | } |
| 745 | ) |
| 746 | continue |
| 747 | reduced = { |
| 748 | "claim_id": f"reduce-{hashlib.sha256(key.encode('utf-8')).hexdigest()[:12]}", |
| 749 | "section": section_by_category.get(finding.get("category"), "Signal & Noise"), |
| 750 | "merged_from": [finding["claim_id"]], |
| 751 | "normalized_claim_key": key, |
| 752 | "claim": finding["claim"], |
| 753 | "citation_bindings": { |
| 754 | "repos": [r.get("ref") for r in refs if r.get("type") == "repo"], |
| 755 | "articles": [r.get("url") for r in refs if r.get("type") == "article"], |
| 756 | }, |
| 757 | "confidence": finding.get("confidence", 0), |
| 758 | "rationale": "Selected by deterministic dry-run reducer because it has explicit evidence references and unique normalized key.", |
| 759 | } |
| 760 | seen_keys[key] = reduced |
| 761 | selected.append(reduced) |
| 762 | for claim in selected: |
| 763 | claim["citation_bindings"]["repos"] = sorted( |
| 764 | set(filter(None, claim["citation_bindings"]["repos"])) |
| 765 | ) |
| 766 | claim["citation_bindings"]["articles"] = sorted( |
| 767 | set(filter(None, claim["citation_bindings"]["articles"])) |
| 768 | ) |
| 769 | contradictions = sorted( |
| 770 | contradictions, key=lambda c: (str(c.get("source_shard")), str(c.get("claim_id"))) |
| 771 | ) |
| 772 | selected = sorted( |
| 773 | selected, |
| 774 | key=lambda c: ( |
| 775 | SECTION_ORDER.index(c["section"]) if c["section"] in SECTION_ORDER else 99, |
| 776 | -float(c["confidence"]), |
| 777 | c["claim_id"], |
| 778 | ), |
| 779 | )[:16] |
| 780 | all_repos = raw_payload.get("new_repos", []) + raw_payload.get("trending_repos", []) |
| 781 | top_repo = ( |
| 782 | normalize_repo_name(sorted_repos(all_repos, mode="new")[:1][0]) |
| 783 | if all_repos |
| 784 | else "unknown/unknown" |
| 785 | ) |
| 786 | topics = ( |
| 787 | raw_payload.get("signals", {}).get("top_topics", []) |
| 788 | if isinstance(raw_payload.get("signals"), dict) |
| 789 | else [] |
| 790 | ) |
| 791 | tags = [] |
| 792 | for topic in topics: |
| 793 | value = topic.get("topic") if isinstance(topic, dict) else topic |
| 794 | if value: |
| 795 | tags.append(str(value)) |
| 796 | tags = tags[:5] or ["open-source", "developer-tools", "automation"] |
| 797 | notable = sorted({repo for claim in selected for repo in claim["citation_bindings"]["repos"]}) |
| 798 | articles = sorted({url for claim in selected for url in claim["citation_bindings"]["articles"]}) |
| 799 | plan = { |
| 800 | "schema_version": PLAN_SCHEMA, |
| 801 | "title": f"{top_repo.split('/')[-1]} and the Week's Candidate Repo Signals", |
| 802 | "summary": "Deterministic map/reduce dry-run candidate built from validated claim ledgers; not publish eligible.", |
| 803 | "top_repo": top_repo, |
| 804 | "tags": tags, |
| 805 | "selected_claims": selected, |
| 806 | "key_references": {"notable_projects": notable[:10], "press_articles": articles[:10]}, |
| 807 | "rejected_claims": rejected, |
| 808 | "contradictions": contradictions, |
| 809 | "quality_notes": [ |
| 810 | CANDIDATE_DISCLAIMER, |
| 811 | "Reducer consumed only validated analysis_map_v1 ledgers.", |
| 812 | ], |
| 813 | } |
| 814 | return plan, rejected, contradictions |
| 815 | |
| 816 | |
| 817 | def section_claims(plan: dict[str, Any], section: str) -> list[dict[str, Any]]: |
| 818 | return [claim for claim in plan.get("selected_claims", []) if claim.get("section") == section] |
| 819 | |
| 820 | |
| 821 | def repo_link(repo: str) -> str: |
| 822 | return f"[{repo}](https://github.com/{repo})" |
| 823 | |
| 824 | |
| 825 | def yaml_quote(value: str) -> str: |
| 826 | return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' |
| 827 | |
| 828 | |
| 829 | def render_claim_sentence(claim: dict[str, Any]) -> str: |
| 830 | repos = [repo_link(repo) for repo in claim.get("citation_bindings", {}).get("repos", [])] |
| 831 | articles = claim.get("citation_bindings", {}).get("articles", []) |
| 832 | evidence = "" |
| 833 | if repos: |
| 834 | evidence = f" Evidence anchor: {', '.join(repos[:3])}." |
| 835 | if articles: |
| 836 | evidence += f" Press reference: {articles[0]}." |
| 837 | return f"{claim.get('claim')} {evidence} Confidence is {float(claim.get('confidence', 0)):.2f}; this remains a candidate signal because the dry-run reducer has not been promoted." |
| 838 | |
| 839 | |
| 840 | def render_section(plan: dict[str, Any], section: str, fallback: str) -> str: |
| 841 | claims = section_claims(plan, section) |
| 842 | sentences = [render_claim_sentence(claim) for claim in claims[:4]] |
| 843 | if not sentences: |
| 844 | sentences = [fallback] |
| 845 | # Add deterministic editorial context so candidate exercises analysis_gate-like structural and evidence checks. |
| 846 | context = { |
| 847 | "This Week's Trends": "The durable trend test is whether repeated repository evidence points to reusable developer infrastructure rather than isolated launches. These candidate claims are useful for QA because each one is bound to ledger provenance, explicit confidence, and source coverage counts.", |
| 848 | "Where Industry Meets Code": "The industry comparison stays cautious: retained press URLs can support context, but the reducer rejects unstated sentiment and keeps weak correlations out of the final plan. This matters because press excitement and repository adoption often move at different speeds.", |
| 849 | "Signal & Noise": "The signal/noise split favors claims with direct repository citations, measurable stars or momentum, and clear uncertainty notes. Noise remains possible where descriptions are thin, stars are early, or a project resembles a promotional launch rather than durable ecosystem work.", |
| 850 | "Blind Spots": "The main blind spots are deterministic: this dry run cannot make live model judgments, cannot infer sentiment beyond supplied artifacts, and cannot publish. Coverage ledgers expose omitted repositories and missing prior context so future QA can decide whether human review is required.", |
| 851 | "The Week Ahead": "Future eligibility depends on the same candidate passing structural analysis gates, evidence gates, and comparison QA while still staying non-publishing until an explicit promotion policy exists. The next run should compare selected references, contradictions, and rejected claims against the current single-pass path.", |
| 852 | }[section] |
| 853 | return "\n\n".join(sentences + [context]) |
| 854 | |
| 855 | |
| 856 | def render_candidate( |
| 857 | plan: dict[str, Any], raw_payload: dict[str, Any], current_datetime: str |
| 858 | ) -> str: |
| 859 | week = raw_payload["week"] |
| 860 | year = int(week.split("-W", 1)[0]) |
| 861 | repos_featured = len(raw_payload.get("new_repos", [])) + len( |
| 862 | raw_payload.get("trending_repos", []) |
| 863 | ) |
| 864 | stars_tracked = sum( |
| 865 | int(repo.get("stars") or 0) |
| 866 | for repo in raw_payload.get("new_repos", []) + raw_payload.get("trending_repos", []) |
| 867 | ) |
| 868 | tags = ", ".join(yaml_quote(str(tag)) for tag in plan["tags"]) |
| 869 | frontmatter = f'''--- |
| 870 | title: {yaml_quote(str(plan["title"]))} |
| 871 | date: {current_datetime} |
| 872 | week: "{week}" |
| 873 | year: {year} |
| 874 | tags: [{tags}] |
| 875 | categories: [weekly] |
| 876 | repos_featured: {repos_featured} |
| 877 | stars_tracked: {stars_tracked} |
| 878 | top_repo: {yaml_quote(str(plan["top_repo"]))} |
| 879 | quality_score: 60 |
| 880 | summary: {yaml_quote(str(plan["summary"]))} |
| 881 | ---''' |
| 882 | notable = plan.get("key_references", {}).get("notable_projects", []) or [plan["top_repo"]] |
| 883 | notable_lines = "\n".join(f"- {repo_link(repo)}" for repo in notable[:10]) |
| 884 | articles = plan.get("key_references", {}).get("press_articles", []) |
| 885 | press_lines = ( |
| 886 | "\n".join(f"- {url}" for url in articles[:10]) |
| 887 | if articles |
| 888 | else "- No retained press URLs were selected by the dry-run reducer." |
| 889 | ) |
| 890 | return ( |
| 891 | frontmatter |
| 892 | + f"\n\n> {CANDIDATE_DISCLAIMER}\n\n" |
| 893 | + "## This Week's Trends\n\n" |
| 894 | + render_section( |
| 895 | plan, |
| 896 | "This Week's Trends", |
| 897 | f"The leading dry-run trend is anchored by {repo_link(plan['top_repo'])}, but the reducer requires future human/model QA before publication.", |
| 898 | ) |
| 899 | + "\n\n## Where Industry Meets Code\n\n" |
| 900 | + render_section( |
| 901 | plan, |
| 902 | "Where Industry Meets Code", |
| 903 | "No strong press correlation survived this deterministic dry run; the absence is surfaced as uncertainty rather than converted into a publishable claim.", |
| 904 | ) |
| 905 | + "\n\n## Signal & Noise\n\n" |
| 906 | + render_section( |
| 907 | plan, |
| 908 | "Signal & Noise", |
| 909 | f"The clearest candidate signal is repository-backed momentum around {repo_link(plan['top_repo'])}, while uncited or duplicate findings stay in rejected sidecars.", |
| 910 | ) |
| 911 | + "\n\n## Blind Spots\n\n" |
| 912 | + render_section( |
| 913 | plan, |
| 914 | "Blind Spots", |
| 915 | "The reducer exposes blind spots instead of filling them with prose: omitted repos, missing press URLs, and absent prior continuity remain QA findings.", |
| 916 | ) |
| 917 | + "\n\n## The Week Ahead\n\n" |
| 918 | + render_section( |
| 919 | plan, |
| 920 | "The Week Ahead", |
| 921 | "Before any promotion, QA must show no regression against the current single-pass path and the candidate must remain blocked from publish workflows.", |
| 922 | ) |
| 923 | + "\n\n## Key References\n\n### Notable Projects\n\n" |
| 924 | + notable_lines |
| 925 | + "\n\n### Press & Industry\n\n" |
| 926 | + press_lines |
| 927 | + "\n" |
| 928 | ) |
| 929 | |
| 930 | |
| 931 | def build_qa_report( |
| 932 | *, |
| 933 | candidate_path: Path, |
| 934 | candidate_text: str, |
| 935 | raw_payload: dict[str, Any], |
| 936 | current_datetime: str, |
| 937 | plan: dict[str, Any], |
| 938 | map_errors: dict[str, list[str]], |
| 939 | baseline_summary: Path | None, |
| 940 | source: str, |
| 941 | model: str, |
| 942 | ) -> dict[str, Any]: |
| 943 | structural_errors, word_count = validate_analysis(candidate_text, raw_payload, current_datetime) |
| 944 | publish_errors, gates = validate_publish_quality( |
| 945 | candidate_text, raw_payload, source=source, model=model |
| 946 | ) |
| 947 | non_provenance_errors = [ |
| 948 | error for error in publish_errors if not error.startswith("AI provenance") |
| 949 | ] |
| 950 | baseline_ref = file_ref(baseline_summary) |
| 951 | selected_refs = set(plan.get("key_references", {}).get("notable_projects", [])) | set( |
| 952 | plan.get("key_references", {}).get("press_articles", []) |
| 953 | ) |
| 954 | report = { |
| 955 | "schema_version": QA_SCHEMA, |
| 956 | "candidate": asdict(file_ref(candidate_path)) if file_ref(candidate_path) else None, |
| 957 | "baseline_summary": asdict(baseline_ref) if baseline_ref else None, |
| 958 | "status": "passed" |
| 959 | if not structural_errors and not non_provenance_errors and not any(map_errors.values()) |
| 960 | else "failed", |
| 961 | "publish_eligible": False, |
| 962 | "promotion_blockers": [ |
| 963 | CANDIDATE_DISCLAIMER, |
| 964 | "analysis source/model are local deterministic dry-run values, not publishable AI provenance.", |
| 965 | "No workflow path promotes map/reduce dry-run output to content/weekly or data/analyzed.", |
| 966 | ], |
| 967 | "regressions": [], |
| 968 | "checks": { |
| 969 | "mapper_contracts": { |
| 970 | "passed": not any(map_errors.values()), |
| 971 | "errors_by_mapper": map_errors, |
| 972 | }, |
| 973 | "structural_analysis_gate": { |
| 974 | "passed": not structural_errors, |
| 975 | "errors": structural_errors, |
| 976 | "word_count": word_count, |
| 977 | }, |
| 978 | "evidence_and_editorial_gates": { |
| 979 | "passed": not non_provenance_errors, |
| 980 | "errors": non_provenance_errors, |
| 981 | "gate_details": gates, |
| 982 | }, |
| 983 | "publish_provenance_gate": { |
| 984 | "passed": False, |
| 985 | "expected_failure": True, |
| 986 | "errors": [error for error in publish_errors if error.startswith("AI provenance")], |
| 987 | }, |
| 988 | "sidecars_present": { |
| 989 | "passed": isinstance(plan.get("rejected_claims"), list) |
| 990 | and isinstance(plan.get("contradictions"), list), |
| 991 | "rejected_count": len(plan.get("rejected_claims", [])), |
| 992 | "contradiction_count": len(plan.get("contradictions", [])), |
| 993 | }, |
| 994 | "reference_count": { |
| 995 | "selected": len(selected_refs), |
| 996 | "notable_projects": len(plan.get("key_references", {}).get("notable_projects", [])), |
| 997 | "press_articles": len(plan.get("key_references", {}).get("press_articles", [])), |
| 998 | }, |
| 999 | }, |
| 1000 | } |
| 1001 | if baseline_summary and not baseline_summary.exists(): |
| 1002 | report["regressions"].append(f"baseline summary not found: {baseline_summary}") |
| 1003 | return report |
| 1004 | |
| 1005 | |
| 1006 | def run(args: argparse.Namespace) -> dict[str, Path]: |
| 1007 | analysis_started = time.monotonic() |
| 1008 | raw_payload = sanitize_repo_payload(load_json(args.raw_json)) |
| 1009 | week = raw_payload["week"] |
| 1010 | raw_ref = file_ref(args.raw_json) |
| 1011 | if raw_ref is None: |
| 1012 | raise ValueError(f"raw JSON not found: {args.raw_json}") |
| 1013 | press_ref = file_ref(args.press_context) |
| 1014 | previous_summary = find_previous_summary(week, args.analyzed_dir) |
| 1015 | previous_ref = file_ref(previous_summary) |
| 1016 | |
| 1017 | map_stage_metrics: list[MapReduceMetrics] = [] |
| 1018 | maps: dict[str, dict[str, Any]] = {} |
| 1019 | map_builders = { |
| 1020 | "new_repos": lambda: map_repositories( |
| 1021 | run_id=args.run_id, |
| 1022 | week=week, |
| 1023 | raw_path=args.raw_json, |
| 1024 | raw_ref=raw_ref, |
| 1025 | shard_id="new_repos", |
| 1026 | repos=raw_payload.get("new_repos", []), |
| 1027 | mode="new", |
| 1028 | max_repos=args.max_repos_per_ledger, |
| 1029 | ), |
| 1030 | "trending_repos": lambda: map_repositories( |
| 1031 | run_id=args.run_id, |
| 1032 | week=week, |
| 1033 | raw_path=args.raw_json, |
| 1034 | raw_ref=raw_ref, |
| 1035 | shard_id="trending_repos", |
| 1036 | repos=raw_payload.get("trending_repos", []), |
| 1037 | mode="trending", |
| 1038 | max_repos=args.max_repos_per_ledger, |
| 1039 | ), |
| 1040 | "press_correlations": lambda: map_press( |
| 1041 | run_id=args.run_id, |
| 1042 | week=week, |
| 1043 | press_path=args.press_context, |
| 1044 | press_ref=press_ref, |
| 1045 | raw_ref=raw_ref, |
| 1046 | ), |
| 1047 | "prior_continuity": lambda: map_prior( |
| 1048 | run_id=args.run_id, |
| 1049 | week=week, |
| 1050 | previous_summary=previous_summary, |
| 1051 | previous_ref=previous_ref, |
| 1052 | raw_ref=raw_ref, |
| 1053 | ), |
| 1054 | } |
| 1055 | for name in MAPPER_IDS: |
| 1056 | stage_started = time.monotonic() |
| 1057 | payload = map_builders[name]() |
| 1058 | maps[name] = payload |
| 1059 | stage_duration = round(time.monotonic() - stage_started, 3) |
| 1060 | input_tokens = int(payload.get("slice", {}).get("input_token_estimate") or 0) |
| 1061 | output_tokens = int(payload.get("token_estimate") or 0) |
| 1062 | map_stage_metrics.append( |
| 1063 | MapReduceMetrics( |
| 1064 | stage=name, |
| 1065 | duration_seconds=stage_duration, |
| 1066 | input_tokens=input_tokens, |
| 1067 | output_tokens=output_tokens, |
| 1068 | cost_usd=metric_cost_usd(args.analysis_model, input_tokens, output_tokens), |
| 1069 | status="pass", |
| 1070 | gate_failure_reasons=[], |
| 1071 | ) |
| 1072 | ) |
| 1073 | map_errors = {name: validate_map(payload) for name, payload in maps.items()} |
| 1074 | if any(map_errors.values()): |
| 1075 | for name, errors in map_errors.items(): |
| 1076 | if errors: |
| 1077 | maps[name]["status"] = "failed" |
| 1078 | maps[name]["errors"] = errors |
| 1079 | map_metrics_by_stage = {metric.stage: metric for metric in map_stage_metrics} |
| 1080 | for name, errors in map_errors.items(): |
| 1081 | if errors: |
| 1082 | metric = map_metrics_by_stage.get(name) |
| 1083 | if metric is not None: |
| 1084 | map_stage_metrics[map_stage_metrics.index(metric)] = MapReduceMetrics( |
| 1085 | stage=metric.stage, |
| 1086 | duration_seconds=metric.duration_seconds, |
| 1087 | input_tokens=metric.input_tokens, |
| 1088 | output_tokens=metric.output_tokens, |
| 1089 | cost_usd=metric.cost_usd, |
| 1090 | status="fail", |
| 1091 | gate_failure_reasons=list(errors), |
| 1092 | ) |
| 1093 | reduce_started = time.monotonic() |
| 1094 | plan, rejected, contradictions = reduce_ledgers(list(maps.values()), raw_payload=raw_payload) |
| 1095 | candidate_text = render_candidate(plan, raw_payload, args.current_datetime) |
| 1096 | |
| 1097 | out = args.output_dir |
| 1098 | maps_dir = out / "maps" |
| 1099 | sidecars_dir = out / "sidecars" |
| 1100 | evidence_slices_dir = sidecars_dir / "evidence-slices" |
| 1101 | evidence_slice_refs: dict[str, dict[str, Any]] = {} |
| 1102 | for name, payload in maps.items(): |
| 1103 | map_path = maps_dir / f"{name}.json" |
| 1104 | write_json(map_path, payload) |
| 1105 | map_ref = file_ref(map_path) |
| 1106 | if map_ref: |
| 1107 | addressed_path = evidence_slices_dir / f"{name}-{map_ref.sha256[:12]}.json" |
| 1108 | write_json(addressed_path, payload) |
| 1109 | addressed_ref = file_ref(addressed_path) |
| 1110 | evidence_slice_refs[name] = asdict(addressed_ref) if addressed_ref else {} |
| 1111 | write_json(out / "editorial-plan.json", plan) |
| 1112 | write_json( |
| 1113 | sidecars_dir / "rejected-claims.json", |
| 1114 | { |
| 1115 | "schema_version": "analysis_rejected_claims_v1", |
| 1116 | "week": week, |
| 1117 | "rejected_claims": rejected, |
| 1118 | }, |
| 1119 | ) |
| 1120 | write_json( |
| 1121 | sidecars_dir / "contradictions.json", |
| 1122 | { |
| 1123 | "schema_version": "analysis_contradictions_v1", |
| 1124 | "week": week, |
| 1125 | "contradictions": contradictions, |
| 1126 | }, |
| 1127 | ) |
| 1128 | candidate_path = out / f"{week}-map-reduce-candidate.md" |
| 1129 | candidate_path.write_text(candidate_text, encoding="utf-8") |
| 1130 | qa = build_qa_report( |
| 1131 | candidate_path=candidate_path, |
| 1132 | candidate_text=candidate_text, |
| 1133 | raw_payload=raw_payload, |
| 1134 | current_datetime=args.current_datetime, |
| 1135 | plan=plan, |
| 1136 | map_errors=map_errors, |
| 1137 | baseline_summary=args.baseline_summary, |
| 1138 | source=args.analysis_source, |
| 1139 | model=args.analysis_model, |
| 1140 | ) |
| 1141 | reduce_duration = round(time.monotonic() - reduce_started, 3) |
| 1142 | reduce_input_tokens = sum(metric.output_tokens for metric in map_stage_metrics) |
| 1143 | reduce_output_tokens = estimate_tokens(candidate_text) |
| 1144 | reduce_failure_reasons = collect_gate_failure_reasons(qa) |
| 1145 | reduce_stage_metric = MapReduceMetrics( |
| 1146 | stage="reduce", |
| 1147 | duration_seconds=reduce_duration, |
| 1148 | input_tokens=reduce_input_tokens, |
| 1149 | output_tokens=reduce_output_tokens, |
| 1150 | cost_usd=metric_cost_usd(args.analysis_model, reduce_input_tokens, reduce_output_tokens), |
| 1151 | status="pass" if qa.get("status") == "passed" else "fail", |
| 1152 | gate_failure_reasons=reduce_failure_reasons, |
| 1153 | ) |
| 1154 | write_json(out / "qa-comparison-report.json", qa) |
| 1155 | raw_component = file_ref(args.raw_json) |
| 1156 | press_component = file_ref(args.press_context) |
| 1157 | template_component = file_ref(ROOT / "prompts" / "analyze-weekly.md") |
| 1158 | prior_component = file_ref(previous_summary) |
| 1159 | slice_components = {name: ref for name, ref in evidence_slice_refs.items()} |
| 1160 | manifest = { |
| 1161 | "schema_version": "analysis_map_reduce_dry_run_manifest_v1", |
| 1162 | "week": week, |
| 1163 | "run_id": args.run_id, |
| 1164 | "created_at": args.current_datetime, |
| 1165 | "publish_eligible": False, |
| 1166 | "candidate_only": True, |
| 1167 | "artifacts": { |
| 1168 | "maps": {name: (maps_dir / f"{name}.json").as_posix() for name in MAPPER_IDS}, |
| 1169 | "evidence_slices": {name: ref.get("path") for name, ref in evidence_slice_refs.items()}, |
| 1170 | "editorial_plan": (out / "editorial-plan.json").as_posix(), |
| 1171 | "rejected_claims": (sidecars_dir / "rejected-claims.json").as_posix(), |
| 1172 | "contradictions": (sidecars_dir / "contradictions.json").as_posix(), |
| 1173 | "candidate": candidate_path.as_posix(), |
| 1174 | "qa_report": (out / "qa-comparison-report.json").as_posix(), |
| 1175 | }, |
| 1176 | "component_estimates": { |
| 1177 | "raw_json": asdict(raw_component) if raw_component else None, |
| 1178 | "press_context": asdict(press_component) if press_component else None, |
| 1179 | "prompt_template": asdict(template_component) if template_component else None, |
| 1180 | "prior_continuity": asdict(prior_component) if prior_component else None, |
| 1181 | "generated_evidence_slices": slice_components, |
| 1182 | "rendered_prompt_estimate": { |
| 1183 | "bytes": len(candidate_text.encode("utf-8")), |
| 1184 | "tokens": estimate_tokens(candidate_text), |
| 1185 | "checksum_sha256": sha256_bytes(candidate_text.encode("utf-8")), |
| 1186 | }, |
| 1187 | }, |
| 1188 | "citation_inventories": { |
| 1189 | "repos": sorted( |
| 1190 | { |
| 1191 | ref.get("ref") |
| 1192 | for payload in maps.values() |
| 1193 | for finding in payload.get("findings", []) |
| 1194 | for ref in finding.get("evidence_refs", []) |
| 1195 | if isinstance(ref, dict) and ref.get("type") == "repo" and ref.get("ref") |
| 1196 | } |
| 1197 | ), |
| 1198 | "press_articles": sorted( |
| 1199 | { |
| 1200 | ref.get("url") |
| 1201 | for payload in maps.values() |
| 1202 | for finding in payload.get("findings", []) |
| 1203 | for ref in finding.get("evidence_refs", []) |
| 1204 | if isinstance(ref, dict) and ref.get("type") == "article" and ref.get("url") |
| 1205 | } |
| 1206 | ), |
| 1207 | }, |
| 1208 | "promotion_policy": "blocked: dry-run/candidate-only map/reduce output must not write data/analyzed, content/weekly, deploy, notify, or satisfy publish eligibility.", |
| 1209 | } |
| 1210 | write_json(out / "manifest.json", manifest) |
| 1211 | total_input_tokens = ( |
| 1212 | sum(metric.input_tokens for metric in map_stage_metrics) + reduce_stage_metric.input_tokens |
| 1213 | ) |
| 1214 | total_output_tokens = ( |
| 1215 | sum(metric.output_tokens for metric in map_stage_metrics) |
| 1216 | + reduce_stage_metric.output_tokens |
| 1217 | ) |
| 1218 | total_cost_usd = round( |
| 1219 | sum(metric.cost_usd for metric in map_stage_metrics) + reduce_stage_metric.cost_usd, 6 |
| 1220 | ) |
| 1221 | analysis_duration = round(time.monotonic() - analysis_started, 3) |
| 1222 | observability_path = DEFAULT_OBSERVABILITY_DIR / f"{week}-map-reduce.json" |
| 1223 | emit_ledger( |
| 1224 | ObservabilityLedger( |
| 1225 | schema_version=METRICS_SCHEMA_VERSION, |
| 1226 | run_id=args.run_id, |
| 1227 | week=week, |
| 1228 | timestamp=args.current_datetime, |
| 1229 | crawl_metrics=[], |
| 1230 | analysis_metrics=AnalysisMetrics( |
| 1231 | duration_seconds=analysis_duration, |
| 1232 | token_ledger={ |
| 1233 | "input_tokens": total_input_tokens, |
| 1234 | "output_tokens": total_output_tokens, |
| 1235 | "total_tokens": total_input_tokens + total_output_tokens, |
| 1236 | "cost_usd": total_cost_usd, |
| 1237 | }, |
| 1238 | map_stages=map_stage_metrics, |
| 1239 | reduce_stage=reduce_stage_metric, |
| 1240 | ), |
| 1241 | environment={ |
| 1242 | "pipeline": "map-reduce-dry-run", |
| 1243 | "analysis_source": args.analysis_source, |
| 1244 | "analysis_model": args.analysis_model, |
| 1245 | "output_dir": out.as_posix(), |
| 1246 | "qa_status": qa.get("status"), |
| 1247 | "publish_eligible": qa.get("publish_eligible"), |
| 1248 | "pass_fail_counts": { |
| 1249 | "map_pass": sum(1 for metric in map_stage_metrics if metric.status == "pass"), |
| 1250 | "map_fail": sum(1 for metric in map_stage_metrics if metric.status == "fail"), |
| 1251 | "reduce_pass": 1 if reduce_stage_metric.status == "pass" else 0, |
| 1252 | "reduce_fail": 1 if reduce_stage_metric.status == "fail" else 0, |
| 1253 | }, |
| 1254 | "gate_failure_reasons": reduce_failure_reasons, |
| 1255 | "artifacts": { |
| 1256 | "manifest": (out / "manifest.json").as_posix(), |
| 1257 | "candidate": candidate_path.as_posix(), |
| 1258 | "qa_report": (out / "qa-comparison-report.json").as_posix(), |
| 1259 | }, |
| 1260 | }, |
| 1261 | ), |
| 1262 | observability_path, |
| 1263 | ) |
| 1264 | return { |
| 1265 | "manifest": out / "manifest.json", |
| 1266 | "qa_report": out / "qa-comparison-report.json", |
| 1267 | "candidate": candidate_path, |
| 1268 | "observability": observability_path, |
| 1269 | } |
| 1270 | |
| 1271 | |
| 1272 | def main(argv: list[str] | None = None) -> int: |
| 1273 | args = parse_args(argv) |
| 1274 | try: |
| 1275 | artifacts = run(args) |
| 1276 | except Exception as exc: # pragma: no cover - CLI guard |
| 1277 | print(f"map/reduce dry-run failed: {exc}", file=sys.stderr) |
| 1278 | return 1 |
| 1279 | print(f"Map/reduce dry-run artifacts written to {args.output_dir}") |
| 1280 | for name, path in artifacts.items(): |
| 1281 | print(f"{name}={path.as_posix()}") |
| 1282 | return 0 |
| 1283 | |
| 1284 | |
| 1285 | if __name__ == "__main__": |
| 1286 | raise SystemExit(main()) |