Add map/reduce dry-run claim-ledger contracts (#286)
* analysis: add map reduce dry-run contracts Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix dry-run contradiction preservation Reject unresolved contradictory map/reduce claims and preserve deterministic audit records in the editorial plan and sidecar. Expand dry-run tests for malformed ledgers, failed coverage, contradiction handling, and citationless findings. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Juan Manuel Servera committed
Jun 6, 2026 at 22:52 UTC
181aef1bd647deebee984ab217e1d3d067fd0f9b
4 files changed
+1068
-2
.github/workflows/crawl-and-publish.yml
+45
-1
@@ -35,6 +35,14 @@ on:
35
required: false
36
default: ''
37
type: string
38
+ analysis_path:
39
+ description: 'Analysis path. map-reduce-dry-run is allowed only with dry-run/candidate-only and never promotes content.'
40
+ required: false
41
+ default: single-pass
42
+ type: choice
43
+ options:
44
+ - single-pass
45
+ - map-reduce-dry-run
46
47
permissions:
48
contents: read
@@ -489,6 +497,7 @@ jobs:
497
echo "preflight_report_md=$PREFLIGHT_MD" >> "$GITHUB_OUTPUT"
498
499
- name: Install Copilot CLI
500
+ if: ${{ inputs.analysis_path != 'map-reduce-dry-run' }}
501
id: install-copilot
502
run: npm install -g @github/copilot
503
@@ -507,6 +516,8 @@ jobs:
516
MANIFEST_FILE="${{ steps.analysis-context.outputs.publish_manifest_file }}"
517
PUBLISHED_SUMMARY="${{ steps.analysis-context.outputs.published_output_file }}"
518
PRESS_FILE="${{ steps.press-context.outputs.press_file }}"
519
+ ANALYSIS_PATH="${{ inputs.analysis_path || 'single-pass' }}"
520
+ RUN_MODE="${{ steps.analysis-context.outputs.run_mode }}"
521
ANALYSIS_STARTED=$(date +%s)
522
DIAGNOSTICS_DIR="$(dirname "$OUTPUT_FILE")/diagnostics"
523
mkdir -p data/metrics "$(dirname "$OUTPUT_FILE")" "$DIAGNOSTICS_DIR"
@@ -527,6 +538,37 @@ jobs:
538
--report-json "$3"
539
}
540
541
+ if [ "$ANALYSIS_PATH" = "map-reduce-dry-run" ]; then
542
+ if [ "$RUN_MODE" != "dry-run" ] && [ "$RUN_MODE" != "candidate-only" ]; then
543
+ echo "::error::map-reduce-dry-run requires run_mode=dry-run or candidate-only and cannot run in a publishing mode."
544
+ exit 1
545
+ fi
546
+ MAP_REDUCE_DIR="$(dirname "$OUTPUT_FILE")/map-reduce"
547
+ python3 scripts/map_reduce_dry_run.py \
548
+ --raw-json "$WEEK_FILE" \
549
+ --press-context "$PRESS_FILE" \
550
+ --output-dir "$MAP_REDUCE_DIR" \
551
+ --current-datetime "$CURRENT_DATETIME" \
552
+ --run-id "${GITHUB_RUN_ID:-local}" \
553
+ --baseline-summary "$PUBLISHED_SUMMARY"
554
+ cp "$MAP_REDUCE_DIR/${WEEK}-map-reduce-candidate.md" "$OUTPUT_FILE"
555
+ ANALYSIS_SOURCE="map-reduce-dry-run"
556
+ ANALYSIS_MODEL="local-deterministic"
557
+ python3 scripts/track_token_usage.py \
558
+ --stage analysis \
559
+ --source "$ANALYSIS_SOURCE" \
560
+ --model "$ANALYSIS_MODEL" \
561
+ --current-datetime "$CURRENT_DATETIME" \
562
+ --week "$WEEK" \
563
+ --prompt-file "$PROMPT_FILE" \
564
+ --output-file "$OUTPUT_FILE"
565
+ echo "::notice::Map/reduce dry-run candidate written to $OUTPUT_FILE with sidecars under $MAP_REDUCE_DIR. It is not publish eligible."
566
+ rm -f "$PROMPT_FILE"
567
+ echo "analysis_source=$ANALYSIS_SOURCE" >> "$GITHUB_OUTPUT"
568
+ echo "analysis_model=$ANALYSIS_MODEL" >> "$GITHUB_OUTPUT"
569
+ exit 0
570
+ fi
571
+
572
# Retry loop: LLM output can be non-deterministically truncated,
573
# so retry up to 3 attempts if the quality gate rejects the article.
574
MAX_RETRIES=2
@@ -735,7 +777,9 @@ jobs:
777
"external_news=data/raw/${WEEK}-external-news.json" \
778
"techcrunch_news=data/raw/${WEEK}-techcrunch.json" \
779
"correlations=data/analyzed/${WEEK}-correlations.json" \
738
- "press_context=data/analyzed/${WEEK}-press-context.md"
780
+ "press_context=data/analyzed/${WEEK}-press-context.md" \
781
+ "map_reduce_manifest=data/candidates/${WEEK}/${RUN_ID}/map-reduce/manifest.json" \
782
+ "map_reduce_qa=data/candidates/${WEEK}/${RUN_ID}/map-reduce/qa-comparison-report.json"
783
do
784
path="${candidate#*=}"
785
[ -f "$path" ] && ARTIFACT_ARGS+=(--artifact "$candidate")
docs/PRD-matrix-crawl-map-reduce-analysis.md
+12
-1
@@ -2,7 +2,7 @@
2
3
**Author:** Leela (Lead/Architect)
4
**Date:** 2026-06-05
5
-**Status:** Draft
5
+**Status:** Phase 2 dry-run scaffolding implemented for issue #258
6
**Type:** Product / Design Requirements Document
7
**Depends on:** docs/analysis-spec.md, docs/pipeline-validation.md, .squad/decisions.md, scripts/analysis_gate.py
8
**Inputs synthesized:** Bender matrix crawl findings, Farnsworth map/reduce analysis findings, Fry QA gates
@@ -772,6 +772,17 @@ The map/reduce analysis path may become publishable only when:
772
10. Reruns on identical input are stable: same top repo or documented reason for change, and at least 70% overlap in selected key references.
773
11. Existing Copilot -> GitHub Models -> no-AI fallback path remains available until map/reduce beats current path on gate pass rate, citation coverage, and editorial review.
774
775
+### Issue #258 dry-run implementation notes
776
+
777
+The first implementation is intentionally candidate-only:
778
+
779
+- `scripts/map_reduce_dry_run.py` performs deterministic local mappers for `new_repos`, `trending_repos`, `press_correlations`, and `prior_continuity`.
780
+- Each mapper writes an `analysis_map_v1` claim ledger with stable claim IDs, evidence refs, confidence/uncertainty, coverage counts, and artifact checksums.
781
+- The reducer writes `analysis_editorial_plan_v1`, plus `sidecars/rejected-claims.json` and `sidecars/contradictions.json`.
782
+- The final writer emits one reader-facing markdown candidate under `data/candidates/.../map-reduce/`; it is marked not publish eligible.
783
+- `qa-comparison-report.json` records structural analysis-gate status, evidence/editorial gate status, expected publish-provenance failure, blockers, and baseline comparison metadata.
784
+- Workflow dispatch exposes `analysis_path=map-reduce-dry-run`, but the workflow rejects it unless `run_mode` is `dry-run` or `candidate-only`. It never runs generate/deploy/notify/promotion paths.
785
+
786
---
787
788
## Open Questions
scripts/map_reduce_dry_run.py
new
+811
@@ -0,0 +1,811 @@
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
+from dataclasses import asdict, dataclass
17
+from datetime import UTC, datetime
18
+from pathlib import Path
19
+from typing import Any
20
+
21
+try:
22
+ from scripts.analyze_fallback import find_previous_summary
23
+ from scripts.analysis_gate import validate_analysis, validate_publish_quality
24
+ from scripts.render_press_context import estimate_tokens
25
+ from scripts.sanitize_repo_content import sanitize_repo_payload
26
+except ModuleNotFoundError: # pragma: no cover - script execution path
27
+ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
28
+ from scripts.analyze_fallback import find_previous_summary
29
+ from scripts.analysis_gate import validate_analysis, validate_publish_quality
30
+ from scripts.render_press_context import estimate_tokens
31
+ from scripts.sanitize_repo_content import sanitize_repo_payload
32
+
33
+ROOT = Path(__file__).resolve().parent.parent
34
+MAP_SCHEMA = "analysis_map_v1"
35
+PLAN_SCHEMA = "analysis_editorial_plan_v1"
36
+QA_SCHEMA = "analysis_map_reduce_qa_v1"
37
+CANDIDATE_DISCLAIMER = "Map/reduce dry-run candidate only; not publish eligible."
38
+MAPPER_IDS = ("new_repos", "trending_repos", "press_correlations", "prior_continuity")
39
+SECTION_ORDER = [
40
+ "This Week's Trends",
41
+ "Where Industry Meets Code",
42
+ "Signal & Noise",
43
+ "Blind Spots",
44
+ "The Week Ahead",
45
+]
46
+
47
+
48
+@dataclass(frozen=True)
49
+class ArtifactRef:
50
+ path: str
51
+ sha256: str | None
52
+ bytes: int
53
+
54
+
55
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
56
+ parser = argparse.ArgumentParser(description="Create deterministic candidate-only map/reduce analysis artifacts.")
57
+ parser.add_argument("--raw-json", required=True, type=Path, help="Canonical weekly raw GitHub crawl payload.")
58
+ parser.add_argument("--output-dir", required=True, type=Path, help="Candidate artifact output directory.")
59
+ parser.add_argument("--current-datetime", required=True, help="ISO-8601 timestamp for the dry run.")
60
+ parser.add_argument("--run-id", default="local", help="Stable run id to include in contracts.")
61
+ parser.add_argument("--press-context", type=Path, help="Rendered press context markdown, if available.")
62
+ parser.add_argument("--analyzed-dir", type=Path, default=ROOT / "data" / "analyzed")
63
+ parser.add_argument("--baseline-summary", type=Path, help="Optional current single-pass summary for QA comparison.")
64
+ parser.add_argument("--max-repos-per-ledger", type=int, default=10)
65
+ parser.add_argument("--analysis-source", default="map-reduce-dry-run")
66
+ parser.add_argument("--analysis-model", default="local-deterministic")
67
+ return parser.parse_args(argv)
68
+
69
+
70
+def load_json(path: Path) -> dict[str, Any]:
71
+ payload = json.loads(path.read_text(encoding="utf-8"))
72
+ if not isinstance(payload, dict):
73
+ raise ValueError(f"JSON payload must be an object: {path}")
74
+ return payload
75
+
76
+
77
+def sha256_bytes(data: bytes) -> str:
78
+ return hashlib.sha256(data).hexdigest()
79
+
80
+
81
+def stable_json(payload: Any) -> str:
82
+ return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
83
+
84
+
85
+def write_json(path: Path, payload: Any) -> None:
86
+ path.parent.mkdir(parents=True, exist_ok=True)
87
+ path.write_text(stable_json(payload), encoding="utf-8")
88
+
89
+
90
+def file_ref(path: Path | None) -> ArtifactRef | None:
91
+ if path is None or not path.exists() or not path.is_file():
92
+ return None
93
+ data = path.read_bytes()
94
+ return ArtifactRef(path=path.as_posix(), sha256=sha256_bytes(data), bytes=len(data))
95
+
96
+
97
+def normalize_repo_name(repo: dict[str, Any]) -> str:
98
+ full_name = str(repo.get("full_name") or "").strip()
99
+ if full_name:
100
+ return full_name
101
+ owner = str(repo.get("owner") or "").strip()
102
+ name = str(repo.get("name") or "").strip()
103
+ return f"{owner}/{name}" if owner and name else name
104
+
105
+
106
+def repo_url(repo: dict[str, Any], full_name: str) -> str:
107
+ return str(repo.get("url") or repo.get("html_url") or f"https://github.com/{full_name}")
108
+
109
+
110
+def repo_description(repo: dict[str, Any]) -> str:
111
+ desc = str(repo.get("description") or "No description provided").strip()
112
+ return re.sub(r"\s+", " ", desc)[:220]
113
+
114
+
115
+def repo_claim_key(repo: dict[str, Any], *, mapper: str) -> str:
116
+ full_name = normalize_repo_name(repo)
117
+ base = f"{mapper}:{full_name}:{repo.get('stars', 0)}:{repo.get('stars_gained', repo.get('gained', 0))}"
118
+ return hashlib.sha256(base.encode("utf-8")).hexdigest()[:12]
119
+
120
+
121
+def sorted_repos(repos: list[dict[str, Any]], *, mode: str) -> list[dict[str, Any]]:
122
+ if mode == "trending":
123
+ return sorted(
124
+ repos,
125
+ key=lambda r: (int(r.get("stars_gained") or r.get("gained") or 0), int(r.get("stars") or 0), normalize_repo_name(r)),
126
+ reverse=True,
127
+ )
128
+ return sorted(repos, key=lambda r: (int(r.get("stars") or 0), normalize_repo_name(r)), reverse=True)
129
+
130
+
131
+def coverage_for_repos(repos: list[dict[str, Any]], input_count: int, *, omitted_reason: str) -> dict[str, Any]:
132
+ seen = [normalize_repo_name(repo) for repo in repos if normalize_repo_name(repo)]
133
+ omitted = max(0, input_count - len(seen))
134
+ return {
135
+ "repo_ids_seen": seen,
136
+ "article_urls_seen": [],
137
+ "repo_count_input": input_count,
138
+ "repo_count_mapped": len(seen),
139
+ "article_count_input": 0,
140
+ "article_count_mapped": 0,
141
+ "excluded_reason_counts": {omitted_reason: omitted} if omitted else {},
142
+ }
143
+
144
+
145
+def make_repo_finding(repo: dict[str, Any], *, mapper: str, category: str, role: str) -> dict[str, Any]:
146
+ full_name = normalize_repo_name(repo)
147
+ stars = int(repo.get("stars") or 0)
148
+ gained = int(repo.get("stars_gained") or repo.get("gained") or 0)
149
+ language = repo.get("language") or "unknown language"
150
+ topics = repo.get("topics") if isinstance(repo.get("topics"), list) else []
151
+ topic_note = f" with topics {', '.join(str(t) for t in topics[:3])}" if topics else ""
152
+ metric_note = f"{stars:,} stars" + (f", {gained:,} gained" if gained else "")
153
+ claim = f"{full_name} is a {role} {language} signal this week ({metric_note}){topic_note}: {repo_description(repo)}"
154
+ confidence = 0.74 if mapper == "trending_repos" and gained else 0.68
155
+ return {
156
+ "claim_id": f"{mapper}-{repo_claim_key(repo, mapper=mapper)}",
157
+ "claim": claim,
158
+ "category": category,
159
+ "source_type": "github",
160
+ "evidence_refs": [
161
+ {
162
+ "type": "repo",
163
+ "ref": full_name,
164
+ "url": repo_url(repo, full_name),
165
+ "role": "anchor",
166
+ "evidence_note": f"Crawler metrics show {metric_note}.",
167
+ }
168
+ ],
169
+ "repo_full_name": full_name,
170
+ "news_url": None,
171
+ "confidence": confidence,
172
+ "contra_refs": [],
173
+ "uncertainties": [] if gained else ["stars_gained unavailable or zero in raw payload"],
174
+ "quality_flags": ["dry_run_local_mapper"],
175
+ }
176
+
177
+
178
+def base_map_payload(*, run_id: str, week: str, shard_id: str, input_refs: list[str], repo_count: int, article_count: int, token_estimate: int) -> dict[str, Any]:
179
+ return {
180
+ "schema_version": MAP_SCHEMA,
181
+ "run_id": run_id,
182
+ "week": week,
183
+ "shard_id": f"signal-type:{shard_id}",
184
+ "slice": {
185
+ "strategy": "signal_type",
186
+ "input_refs": input_refs,
187
+ "input_token_estimate": token_estimate,
188
+ "repo_count": repo_count,
189
+ "article_count": article_count,
190
+ },
191
+ "coverage": {},
192
+ "findings": [],
193
+ "citations": [],
194
+ "reference_candidates": {"notable_projects": [], "press_articles": []},
195
+ "token_estimate": 0,
196
+ "model": "none",
197
+ "status": "success",
198
+ "errors": [],
199
+ "provenance": {},
200
+ }
201
+
202
+
203
+def map_repositories(
204
+ *,
205
+ run_id: str,
206
+ week: str,
207
+ raw_path: Path,
208
+ raw_ref: ArtifactRef,
209
+ shard_id: str,
210
+ repos: list[dict[str, Any]],
211
+ mode: str,
212
+ max_repos: int,
213
+) -> dict[str, Any]:
214
+ selected = sorted_repos(repos, mode=mode)[:max_repos]
215
+ payload = base_map_payload(
216
+ run_id=run_id,
217
+ week=week,
218
+ shard_id=shard_id,
219
+ input_refs=[f"{raw_path.as_posix()}#{shard_id}[0:{len(repos)}]"],
220
+ repo_count=len(repos),
221
+ article_count=0,
222
+ token_estimate=estimate_tokens(stable_json(repos)),
223
+ )
224
+ category = "trend" if shard_id == "new_repos" else "signal"
225
+ role = "new-repository" if shard_id == "new_repos" else "momentum"
226
+ findings = [make_repo_finding(repo, mapper=shard_id, category=category, role=role) for repo in selected]
227
+ payload["findings"] = findings
228
+ payload["coverage"] = coverage_for_repos(selected, len(repos), omitted_reason="outside_dry_run_top_repo_limit")
229
+ payload["citations"] = [
230
+ {"type": "repo", "url": item["evidence_refs"][0]["url"], "title": item["repo_full_name"]} for item in findings
231
+ ]
232
+ payload["reference_candidates"] = {
233
+ "notable_projects": [item["repo_full_name"] for item in findings],
234
+ "press_articles": [],
235
+ }
236
+ payload["token_estimate"] = estimate_tokens(stable_json(payload))
237
+ payload["provenance"] = {"raw_json": asdict(raw_ref), "deterministic_mapper": True}
238
+ return payload
239
+
240
+
241
+def extract_press_articles(press_context: str) -> list[dict[str, str]]:
242
+ urls = []
243
+ for match in re.finditer(r"https?://[^\s)\]]+", press_context):
244
+ url = match.group(0).rstrip(".,")
245
+ if url not in urls:
246
+ urls.append(url)
247
+ articles: list[dict[str, str]] = []
248
+ lines = [line.strip(" -*") for line in press_context.splitlines() if line.strip()]
249
+ for url in urls[:10]:
250
+ title = next((line[:120] for line in lines if url in line), url)
251
+ articles.append({"url": url, "title": title})
252
+ return articles
253
+
254
+
255
+def map_press(
256
+ *, run_id: str, week: str, press_path: Path | None, press_ref: ArtifactRef | None, raw_ref: ArtifactRef
257
+) -> dict[str, Any]:
258
+ content = press_path.read_text(encoding="utf-8") if press_path and press_path.exists() else ""
259
+ articles = extract_press_articles(content)
260
+ payload = base_map_payload(
261
+ run_id=run_id,
262
+ week=week,
263
+ shard_id="press_correlations",
264
+ input_refs=[press_path.as_posix() if press_path else "press_context:none"],
265
+ repo_count=0,
266
+ article_count=len(articles),
267
+ token_estimate=estimate_tokens(content),
268
+ )
269
+ findings = []
270
+ for index, article in enumerate(articles[:5], start=1):
271
+ claim_id = hashlib.sha256(f"press:{article['url']}".encode("utf-8")).hexdigest()[:12]
272
+ findings.append(
273
+ {
274
+ "claim_id": f"press_correlations-{claim_id}",
275
+ "claim": f"Retained press context cites {article['title']} as industry evidence to compare against repository activity.",
276
+ "category": "press_correlation",
277
+ "source_type": "news",
278
+ "evidence_refs": [
279
+ {
280
+ "type": "article",
281
+ "ref": article["url"],
282
+ "url": article["url"],
283
+ "role": "supporting",
284
+ "evidence_note": "URL was retained in rendered press context.",
285
+ }
286
+ ],
287
+ "repo_full_name": None,
288
+ "news_url": article["url"],
289
+ "confidence": 0.66,
290
+ "contra_refs": [],
291
+ "uncertainties": ["dry-run mapper does not infer unstated press sentiment"],
292
+ "quality_flags": ["dry_run_local_mapper"],
293
+ }
294
+ )
295
+ if not findings:
296
+ payload["status"] = "partial"
297
+ payload["errors"] = ["No press URLs were available; mapper emitted coverage-only ledger."]
298
+ payload["findings"] = findings
299
+ payload["coverage"] = {
300
+ "repo_ids_seen": [],
301
+ "article_urls_seen": [article["url"] for article in articles],
302
+ "repo_count_input": 0,
303
+ "repo_count_mapped": 0,
304
+ "article_count_input": len(articles),
305
+ "article_count_mapped": len(articles[:5]),
306
+ "excluded_reason_counts": {"outside_dry_run_article_limit": max(0, len(articles) - 5)} if len(articles) > 5 else {},
307
+ }
308
+ payload["citations"] = [{"type": "article", "url": item["news_url"], "title": item["claim"][:80]} for item in findings]
309
+ payload["reference_candidates"] = {"notable_projects": [], "press_articles": [item["news_url"] for item in findings]}
310
+ payload["token_estimate"] = estimate_tokens(stable_json(payload))
311
+ payload["provenance"] = {"raw_json": asdict(raw_ref), "press_context": asdict(press_ref) if press_ref else None, "deterministic_mapper": True}
312
+ return payload
313
+
314
+
315
+def map_prior(
316
+ *, run_id: str, week: str, previous_summary: Path | None, previous_ref: ArtifactRef | None, raw_ref: ArtifactRef
317
+) -> dict[str, Any]:
318
+ content = previous_summary.read_text(encoding="utf-8") if previous_summary and previous_summary.exists() else ""
319
+ payload = base_map_payload(
320
+ run_id=run_id,
321
+ week=week,
322
+ shard_id="prior_continuity",
323
+ input_refs=[previous_summary.as_posix() if previous_summary else "prior_summary:none"],
324
+ repo_count=0,
325
+ article_count=0,
326
+ token_estimate=estimate_tokens(content),
327
+ )
328
+ finding = {
329
+ "claim_id": f"prior_continuity-{hashlib.sha256((previous_summary.as_posix() if previous_summary else 'none').encode()).hexdigest()[:12]}",
330
+ "claim": (
331
+ "Prior weekly analysis is available for continuity checks; reducer should compare carried-forward claims against this week's evidence."
332
+ if content
333
+ else "No prior weekly analysis was available, so continuity claims should be treated as open blind spots."
334
+ ),
335
+ "category": "continuity",
336
+ "source_type": "prior_summary",
337
+ "evidence_refs": [
338
+ {
339
+ "type": "prior_summary",
340
+ "ref": previous_summary.as_posix() if previous_summary else "none",
341
+ "url": previous_summary.as_posix() if previous_summary else "none",
342
+ "role": "supporting",
343
+ "evidence_note": "Deterministic local continuity marker.",
344
+ }
345
+ ],
346
+ "repo_full_name": None,
347
+ "news_url": None,
348
+ "confidence": 0.55 if content else 0.35,
349
+ "contra_refs": [],
350
+ "uncertainties": [] if content else ["no prior summary artifact found"],
351
+ "quality_flags": ["dry_run_local_mapper"],
352
+ }
353
+ payload["findings"] = [finding]
354
+ payload["coverage"] = {
355
+ "repo_ids_seen": [],
356
+ "article_urls_seen": [],
357
+ "repo_count_input": 0,
358
+ "repo_count_mapped": 0,
359
+ "article_count_input": 0,
360
+ "article_count_mapped": 0,
361
+ "excluded_reason_counts": {},
362
+ "prior_summary_present": bool(content),
363
+ }
364
+ payload["citations"] = [{"type": "prior_summary", "url": finding["evidence_refs"][0]["url"], "title": "prior weekly summary"}]
365
+ payload["reference_candidates"] = {"notable_projects": [], "press_articles": []}
366
+ payload["token_estimate"] = estimate_tokens(stable_json(payload))
367
+ payload["provenance"] = {"raw_json": asdict(raw_ref), "prior_summary": asdict(previous_ref) if previous_ref else None, "deterministic_mapper": True}
368
+ return payload
369
+
370
+
371
+def validate_map(payload: dict[str, Any]) -> list[str]:
372
+ errors: list[str] = []
373
+ if payload.get("schema_version") != MAP_SCHEMA:
374
+ errors.append("mapper schema_version mismatch")
375
+ for field in ("run_id", "week", "shard_id", "slice", "coverage", "findings", "citations", "reference_candidates", "provenance"):
376
+ if field not in payload:
377
+ errors.append(f"mapper missing {field}")
378
+ if "findings" in payload and not isinstance(payload.get("findings"), list):
379
+ errors.append("findings must be a list")
380
+ findings = payload.get("findings") if isinstance(payload.get("findings"), list) else []
381
+ for index, finding in enumerate(findings):
382
+ if not isinstance(finding, dict):
383
+ errors.append(f"finding {index} must be an object")
384
+ continue
385
+ for field in ("claim_id", "claim", "category", "source_type", "evidence_refs", "confidence", "contra_refs", "uncertainties"):
386
+ if field not in finding:
387
+ errors.append(f"finding {index} missing {field}")
388
+ refs = finding.get("evidence_refs")
389
+ if not isinstance(refs, list) or not refs:
390
+ errors.append(f"finding {index} has no evidence refs")
391
+ else:
392
+ for ref in refs:
393
+ if not isinstance(ref, dict) or not ref.get("type") or not ref.get("ref") or not ref.get("url"):
394
+ errors.append(f"finding {index} has malformed evidence ref")
395
+ confidence = finding.get("confidence")
396
+ if not isinstance(confidence, (int, float)) or not (0 <= float(confidence) <= 1):
397
+ errors.append(f"finding {index} confidence out of range")
398
+ if "contra_refs" in finding and not isinstance(finding.get("contra_refs"), list):
399
+ errors.append(f"finding {index} contra_refs must be a list")
400
+ coverage = payload.get("coverage")
401
+ if not isinstance(coverage, dict):
402
+ errors.append("coverage must be an object")
403
+ else:
404
+ for key in ("repo_ids_seen", "article_urls_seen", "excluded_reason_counts"):
405
+ if key not in coverage:
406
+ errors.append(f"coverage missing {key}")
407
+ if "repo_ids_seen" in coverage and not isinstance(coverage.get("repo_ids_seen"), list):
408
+ errors.append("coverage repo_ids_seen must be a list")
409
+ if "article_urls_seen" in coverage and not isinstance(coverage.get("article_urls_seen"), list):
410
+ errors.append("coverage article_urls_seen must be a list")
411
+ excluded = coverage.get("excluded_reason_counts")
412
+ if "excluded_reason_counts" in coverage and not isinstance(excluded, dict):
413
+ errors.append("coverage excluded_reason_counts must be an object")
414
+ elif isinstance(excluded, dict):
415
+ for reason, count in excluded.items():
416
+ if not reason or not isinstance(count, int) or count < 0:
417
+ errors.append("coverage excluded_reason_counts must contain non-negative integer counts")
418
+ break
419
+ for prefix in ("repo", "article"):
420
+ input_key = f"{prefix}_count_input"
421
+ mapped_key = f"{prefix}_count_mapped"
422
+ if input_key in coverage or mapped_key in coverage:
423
+ input_count = coverage.get(input_key)
424
+ mapped_count = coverage.get(mapped_key)
425
+ if not isinstance(input_count, int) or not isinstance(mapped_count, int) or input_count < 0 or mapped_count < 0:
426
+ errors.append(f"coverage {prefix} counts must be non-negative integers")
427
+ continue
428
+ if mapped_count > input_count:
429
+ errors.append(f"coverage {mapped_key} exceeds {input_key}")
430
+ if mapped_count < input_count and not coverage.get("excluded_reason_counts"):
431
+ errors.append(f"coverage {mapped_key} below {input_key} without excluded reasons")
432
+ status = payload.get("status")
433
+ if status == "failed":
434
+ errors.append("mapper status failed")
435
+ return errors
436
+
437
+
438
+def normalized_claim_key(finding: dict[str, Any]) -> str:
439
+ repo = finding.get("repo_full_name") or ""
440
+ article = finding.get("news_url") or ""
441
+ claim = str(finding.get("claim") or "").lower()
442
+ words = "-".join(re.findall(r"[a-z0-9]+", claim)[:8])
443
+ return f"{finding.get('category')}:{repo or article or words}"
444
+
445
+
446
+def contra_ref_targets(contra_refs: Any) -> set[str]:
447
+ targets: set[str] = set()
448
+ if not isinstance(contra_refs, list):
449
+ return targets
450
+ for ref in contra_refs:
451
+ if isinstance(ref, str) and ref:
452
+ targets.add(ref)
453
+ elif isinstance(ref, dict):
454
+ for key in ("claim_id", "ref", "url"):
455
+ value = ref.get(key)
456
+ if value:
457
+ targets.add(str(value))
458
+ return targets
459
+
460
+
461
+def contradiction_record(
462
+ finding: dict[str, Any],
463
+ ledger: dict[str, Any],
464
+ *,
465
+ contradicted_by: list[str],
466
+) -> dict[str, Any]:
467
+ contra_refs = finding.get("contra_refs") if isinstance(finding.get("contra_refs"), list) else []
468
+ return {
469
+ "claim_id": finding.get("claim_id"),
470
+ "claim": finding.get("claim"),
471
+ "source_shard": ledger.get("shard_id"),
472
+ "normalized_claim_key": normalized_claim_key(finding),
473
+ "evidence_refs": finding.get("evidence_refs") if isinstance(finding.get("evidence_refs"), list) else [],
474
+ "contra_refs": contra_refs,
475
+ "contradicted_by": sorted(set(contradicted_by)),
476
+ "resolution": "rejected_unresolved",
477
+ "reason": "Unresolved contradiction refs are preserved for audit and excluded from selected editorial material.",
478
+ }
479
+
480
+
481
+def reduce_ledgers(ledgers: list[dict[str, Any]], *, raw_payload: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
482
+ selected: list[dict[str, Any]] = []
483
+ rejected: list[dict[str, Any]] = []
484
+ contradictions: list[dict[str, Any]] = []
485
+ seen_keys: dict[str, dict[str, Any]] = {}
486
+ section_by_category = {
487
+ "trend": "This Week's Trends",
488
+ "signal": "Signal & Noise",
489
+ "noise": "Signal & Noise",
490
+ "gap": "Blind Spots",
491
+ "press_correlation": "Where Industry Meets Code",
492
+ "press_divergence": "Where Industry Meets Code",
493
+ "continuity": "The Week Ahead",
494
+ }
495
+ inbound_contradictions: dict[str, list[str]] = {}
496
+ for ledger in ledgers:
497
+ for finding in ledger.get("findings", []) if isinstance(ledger.get("findings"), list) else []:
498
+ if not isinstance(finding, dict):
499
+ continue
500
+ source_claim_id = finding.get("claim_id")
501
+ for target in contra_ref_targets(finding.get("contra_refs")):
502
+ inbound_contradictions.setdefault(target, []).append(str(source_claim_id))
503
+ for ledger in ledgers:
504
+ ledger_findings = ledger.get("findings", []) if isinstance(ledger.get("findings"), list) else []
505
+ for finding in ledger_findings:
506
+ if not isinstance(finding, dict):
507
+ rejected.append({"claim_id": None, "reason": "malformed_finding", "source_shard": ledger.get("shard_id")})
508
+ continue
509
+ refs = finding.get("evidence_refs") if isinstance(finding.get("evidence_refs"), list) else []
510
+ contra_refs = finding.get("contra_refs") if isinstance(finding.get("contra_refs"), list) else []
511
+ contradicted_by = inbound_contradictions.get(str(finding.get("claim_id")), [])
512
+ if contra_refs or contradicted_by:
513
+ contradictions.append(contradiction_record(finding, ledger, contradicted_by=contradicted_by))
514
+ rejected.append({"claim_id": finding.get("claim_id"), "reason": "unresolved_contradiction", "source_shard": ledger.get("shard_id")})
515
+ continue
516
+ if not refs:
517
+ rejected.append({"claim_id": finding.get("claim_id"), "reason": "weak_citation", "source_shard": ledger.get("shard_id")})
518
+ continue
519
+ key = normalized_claim_key(finding)
520
+ if key in seen_keys:
521
+ existing = seen_keys[key]
522
+ existing["merged_from"].append(finding["claim_id"])
523
+ existing["citation_bindings"]["repos"].extend([r.get("ref") for r in refs if r.get("type") == "repo"])
524
+ existing["citation_bindings"]["articles"].extend([r.get("url") for r in refs if r.get("type") == "article"])
525
+ rejected.append({"claim_id": finding.get("claim_id"), "reason": "duplicate", "source_shard": ledger.get("shard_id")})
526
+ continue
527
+ reduced = {
528
+ "claim_id": f"reduce-{hashlib.sha256(key.encode('utf-8')).hexdigest()[:12]}",
529
+ "section": section_by_category.get(finding.get("category"), "Signal & Noise"),
530
+ "merged_from": [finding["claim_id"]],
531
+ "normalized_claim_key": key,
532
+ "claim": finding["claim"],
533
+ "citation_bindings": {
534
+ "repos": [r.get("ref") for r in refs if r.get("type") == "repo"],
535
+ "articles": [r.get("url") for r in refs if r.get("type") == "article"],
536
+ },
537
+ "confidence": finding.get("confidence", 0),
538
+ "rationale": "Selected by deterministic dry-run reducer because it has explicit evidence references and unique normalized key.",
539
+ }
540
+ seen_keys[key] = reduced
541
+ selected.append(reduced)
542
+ for claim in selected:
543
+ claim["citation_bindings"]["repos"] = sorted(set(filter(None, claim["citation_bindings"]["repos"])))
544
+ claim["citation_bindings"]["articles"] = sorted(set(filter(None, claim["citation_bindings"]["articles"])))
545
+ contradictions = sorted(contradictions, key=lambda c: (str(c.get("source_shard")), str(c.get("claim_id"))))
546
+ selected = sorted(selected, key=lambda c: (SECTION_ORDER.index(c["section"]) if c["section"] in SECTION_ORDER else 99, -float(c["confidence"]), c["claim_id"]))[:16]
547
+ all_repos = raw_payload.get("new_repos", []) + raw_payload.get("trending_repos", [])
548
+ top_repo = normalize_repo_name(sorted_repos(all_repos, mode="new")[:1][0]) if all_repos else "unknown/unknown"
549
+ topics = raw_payload.get("signals", {}).get("top_topics", []) if isinstance(raw_payload.get("signals"), dict) else []
550
+ tags = []
551
+ for topic in topics:
552
+ value = topic.get("topic") if isinstance(topic, dict) else topic
553
+ if value:
554
+ tags.append(str(value))
555
+ tags = tags[:5] or ["open-source", "developer-tools", "automation"]
556
+ notable = sorted({repo for claim in selected for repo in claim["citation_bindings"]["repos"]})
557
+ articles = sorted({url for claim in selected for url in claim["citation_bindings"]["articles"]})
558
+ plan = {
559
+ "schema_version": PLAN_SCHEMA,
560
+ "title": f"{top_repo.split('/')[-1]} and the Week's Candidate Repo Signals",
561
+ "summary": "Deterministic map/reduce dry-run candidate built from validated claim ledgers; not publish eligible.",
562
+ "top_repo": top_repo,
563
+ "tags": tags,
564
+ "selected_claims": selected,
565
+ "key_references": {"notable_projects": notable[:10], "press_articles": articles[:10]},
566
+ "rejected_claims": rejected,
567
+ "contradictions": contradictions,
568
+ "quality_notes": [CANDIDATE_DISCLAIMER, "Reducer consumed only validated analysis_map_v1 ledgers."],
569
+ }
570
+ return plan, rejected, contradictions
571
+
572
+
573
+def section_claims(plan: dict[str, Any], section: str) -> list[dict[str, Any]]:
574
+ return [claim for claim in plan.get("selected_claims", []) if claim.get("section") == section]
575
+
576
+
577
+def repo_link(repo: str) -> str:
578
+ return f"[{repo}](https://github.com/{repo})"
579
+
580
+
581
+def yaml_quote(value: str) -> str:
582
+ return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
583
+
584
+
585
+def render_claim_sentence(claim: dict[str, Any]) -> str:
586
+ repos = [repo_link(repo) for repo in claim.get("citation_bindings", {}).get("repos", [])]
587
+ articles = claim.get("citation_bindings", {}).get("articles", [])
588
+ evidence = ""
589
+ if repos:
590
+ evidence = f" Evidence anchor: {', '.join(repos[:3])}."
591
+ if articles:
592
+ evidence += f" Press reference: {articles[0]}."
593
+ 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."
594
+
595
+
596
+def render_section(plan: dict[str, Any], section: str, fallback: str) -> str:
597
+ claims = section_claims(plan, section)
598
+ sentences = [render_claim_sentence(claim) for claim in claims[:4]]
599
+ if not sentences:
600
+ sentences = [fallback]
601
+ # Add deterministic editorial context so candidate exercises analysis_gate-like structural and evidence checks.
602
+ context = {
603
+ "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.",
604
+ "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.",
605
+ "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.",
606
+ "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.",
607
+ "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.",
608
+ }[section]
609
+ return "\n\n".join(sentences + [context])
610
+
611
+
612
+def render_candidate(plan: dict[str, Any], raw_payload: dict[str, Any], current_datetime: str) -> str:
613
+ week = raw_payload["week"]
614
+ year = int(week.split("-W", 1)[0])
615
+ repos_featured = len(raw_payload.get("new_repos", [])) + len(raw_payload.get("trending_repos", []))
616
+ stars_tracked = sum(int(repo.get("stars") or 0) for repo in raw_payload.get("new_repos", []) + raw_payload.get("trending_repos", []))
617
+ tags = ", ".join(yaml_quote(str(tag)) for tag in plan["tags"])
618
+ frontmatter = f'''---
619
+title: {yaml_quote(str(plan['title']))}
620
+date: {current_datetime}
621
+week: "{week}"
622
+year: {year}
623
+tags: [{tags}]
624
+categories: [weekly]
625
+repos_featured: {repos_featured}
626
+stars_tracked: {stars_tracked}
627
+top_repo: {yaml_quote(str(plan['top_repo']))}
628
+quality_score: 60
629
+summary: {yaml_quote(str(plan['summary']))}
630
+---'''
631
+ notable = plan.get("key_references", {}).get("notable_projects", []) or [plan["top_repo"]]
632
+ notable_lines = "\n".join(f"- {repo_link(repo)}" for repo in notable[:10])
633
+ articles = plan.get("key_references", {}).get("press_articles", [])
634
+ press_lines = "\n".join(f"- {url}" for url in articles[:10]) if articles else "- No retained press URLs were selected by the dry-run reducer."
635
+ return (
636
+ frontmatter
637
+ + f"\n\n> {CANDIDATE_DISCLAIMER}\n\n"
638
+ + "## This Week's Trends\n\n"
639
+ + render_section(plan, "This Week's Trends", f"The leading dry-run trend is anchored by {repo_link(plan['top_repo'])}, but the reducer requires future human/model QA before publication.")
640
+ + "\n\n## Where Industry Meets Code\n\n"
641
+ + render_section(plan, "Where Industry Meets Code", "No strong press correlation survived this deterministic dry run; the absence is surfaced as uncertainty rather than converted into a publishable claim.")
642
+ + "\n\n## Signal & Noise\n\n"
643
+ + render_section(plan, "Signal & Noise", f"The clearest candidate signal is repository-backed momentum around {repo_link(plan['top_repo'])}, while uncited or duplicate findings stay in rejected sidecars.")
644
+ + "\n\n## Blind Spots\n\n"
645
+ + render_section(plan, "Blind Spots", "The reducer exposes blind spots instead of filling them with prose: omitted repos, missing press URLs, and absent prior continuity remain QA findings.")
646
+ + "\n\n## The Week Ahead\n\n"
647
+ + render_section(plan, "The Week Ahead", "Before any promotion, QA must show no regression against the current single-pass path and the candidate must remain blocked from publish workflows.")
648
+ + "\n\n## Key References\n\n### Notable Projects\n\n"
649
+ + notable_lines
650
+ + "\n\n### Press & Industry\n\n"
651
+ + press_lines
652
+ + "\n"
653
+ )
654
+
655
+
656
+def build_qa_report(
657
+ *,
658
+ candidate_path: Path,
659
+ candidate_text: str,
660
+ raw_payload: dict[str, Any],
661
+ current_datetime: str,
662
+ plan: dict[str, Any],
663
+ map_errors: dict[str, list[str]],
664
+ baseline_summary: Path | None,
665
+ source: str,
666
+ model: str,
667
+) -> dict[str, Any]:
668
+ structural_errors, word_count = validate_analysis(candidate_text, raw_payload, current_datetime)
669
+ publish_errors, gates = validate_publish_quality(candidate_text, raw_payload, source=source, model=model)
670
+ non_provenance_errors = [error for error in publish_errors if not error.startswith("AI provenance")]
671
+ baseline_ref = file_ref(baseline_summary)
672
+ selected_refs = set(plan.get("key_references", {}).get("notable_projects", [])) | set(plan.get("key_references", {}).get("press_articles", []))
673
+ report = {
674
+ "schema_version": QA_SCHEMA,
675
+ "candidate": asdict(file_ref(candidate_path)) if file_ref(candidate_path) else None,
676
+ "baseline_summary": asdict(baseline_ref) if baseline_ref else None,
677
+ "status": "passed" if not structural_errors and not non_provenance_errors and not any(map_errors.values()) else "failed",
678
+ "publish_eligible": False,
679
+ "promotion_blockers": [
680
+ CANDIDATE_DISCLAIMER,
681
+ "analysis source/model are local deterministic dry-run values, not publishable AI provenance.",
682
+ "No workflow path promotes map/reduce dry-run output to content/weekly or data/analyzed.",
683
+ ],
684
+ "regressions": [],
685
+ "checks": {
686
+ "mapper_contracts": {"passed": not any(map_errors.values()), "errors_by_mapper": map_errors},
687
+ "structural_analysis_gate": {"passed": not structural_errors, "errors": structural_errors, "word_count": word_count},
688
+ "evidence_and_editorial_gates": {"passed": not non_provenance_errors, "errors": non_provenance_errors, "gate_details": gates},
689
+ "publish_provenance_gate": {
690
+ "passed": False,
691
+ "expected_failure": True,
692
+ "errors": [error for error in publish_errors if error.startswith("AI provenance")],
693
+ },
694
+ "sidecars_present": {
695
+ "passed": isinstance(plan.get("rejected_claims"), list) and isinstance(plan.get("contradictions"), list),
696
+ "rejected_count": len(plan.get("rejected_claims", [])),
697
+ "contradiction_count": len(plan.get("contradictions", [])),
698
+ },
699
+ "reference_count": {"selected": len(selected_refs), "notable_projects": len(plan.get("key_references", {}).get("notable_projects", [])), "press_articles": len(plan.get("key_references", {}).get("press_articles", []))},
700
+ },
701
+ }
702
+ if baseline_summary and not baseline_summary.exists():
703
+ report["regressions"].append(f"baseline summary not found: {baseline_summary}")
704
+ return report
705
+
706
+
707
+def run(args: argparse.Namespace) -> dict[str, Path]:
708
+ raw_payload = sanitize_repo_payload(load_json(args.raw_json))
709
+ week = raw_payload["week"]
710
+ raw_ref = file_ref(args.raw_json)
711
+ if raw_ref is None:
712
+ raise ValueError(f"raw JSON not found: {args.raw_json}")
713
+ press_ref = file_ref(args.press_context)
714
+ previous_summary = find_previous_summary(week, args.analyzed_dir)
715
+ previous_ref = file_ref(previous_summary)
716
+
717
+ maps = {
718
+ "new_repos": map_repositories(
719
+ run_id=args.run_id,
720
+ week=week,
721
+ raw_path=args.raw_json,
722
+ raw_ref=raw_ref,
723
+ shard_id="new_repos",
724
+ repos=raw_payload.get("new_repos", []),
725
+ mode="new",
726
+ max_repos=args.max_repos_per_ledger,
727
+ ),
728
+ "trending_repos": map_repositories(
729
+ run_id=args.run_id,
730
+ week=week,
731
+ raw_path=args.raw_json,
732
+ raw_ref=raw_ref,
733
+ shard_id="trending_repos",
734
+ repos=raw_payload.get("trending_repos", []),
735
+ mode="trending",
736
+ max_repos=args.max_repos_per_ledger,
737
+ ),
738
+ "press_correlations": map_press(run_id=args.run_id, week=week, press_path=args.press_context, press_ref=press_ref, raw_ref=raw_ref),
739
+ "prior_continuity": map_prior(run_id=args.run_id, week=week, previous_summary=previous_summary, previous_ref=previous_ref, raw_ref=raw_ref),
740
+ }
741
+ map_errors = {name: validate_map(payload) for name, payload in maps.items()}
742
+ if any(map_errors.values()):
743
+ for name, errors in map_errors.items():
744
+ if errors:
745
+ maps[name]["status"] = "failed"
746
+ maps[name]["errors"] = errors
747
+ plan, rejected, contradictions = reduce_ledgers(list(maps.values()), raw_payload=raw_payload)
748
+ candidate_text = render_candidate(plan, raw_payload, args.current_datetime)
749
+
750
+ out = args.output_dir
751
+ maps_dir = out / "maps"
752
+ sidecars_dir = out / "sidecars"
753
+ for name, payload in maps.items():
754
+ write_json(maps_dir / f"{name}.json", payload)
755
+ write_json(out / "editorial-plan.json", plan)
756
+ write_json(sidecars_dir / "rejected-claims.json", {"schema_version": "analysis_rejected_claims_v1", "week": week, "rejected_claims": rejected})
757
+ write_json(sidecars_dir / "contradictions.json", {"schema_version": "analysis_contradictions_v1", "week": week, "contradictions": contradictions})
758
+ candidate_path = out / f"{week}-map-reduce-candidate.md"
759
+ candidate_path.write_text(candidate_text, encoding="utf-8")
760
+ qa = build_qa_report(
761
+ candidate_path=candidate_path,
762
+ candidate_text=candidate_text,
763
+ raw_payload=raw_payload,
764
+ current_datetime=args.current_datetime,
765
+ plan=plan,
766
+ map_errors=map_errors,
767
+ baseline_summary=args.baseline_summary,
768
+ source=args.analysis_source,
769
+ model=args.analysis_model,
770
+ )
771
+ write_json(out / "qa-comparison-report.json", qa)
772
+ manifest = {
773
+ "schema_version": "analysis_map_reduce_dry_run_manifest_v1",
774
+ "week": week,
775
+ "run_id": args.run_id,
776
+ "created_at": args.current_datetime,
777
+ "publish_eligible": False,
778
+ "candidate_only": True,
779
+ "artifacts": {
780
+ "maps": {name: (maps_dir / f"{name}.json").as_posix() for name in MAPPER_IDS},
781
+ "editorial_plan": (out / "editorial-plan.json").as_posix(),
782
+ "rejected_claims": (sidecars_dir / "rejected-claims.json").as_posix(),
783
+ "contradictions": (sidecars_dir / "contradictions.json").as_posix(),
784
+ "candidate": candidate_path.as_posix(),
785
+ "qa_report": (out / "qa-comparison-report.json").as_posix(),
786
+ },
787
+ "promotion_policy": "blocked: dry-run/candidate-only map/reduce output must not write data/analyzed, content/weekly, deploy, notify, or satisfy publish eligibility.",
788
+ }
789
+ write_json(out / "manifest.json", manifest)
790
+ return {
791
+ "manifest": out / "manifest.json",
792
+ "qa_report": out / "qa-comparison-report.json",
793
+ "candidate": candidate_path,
794
+ }
795
+
796
+
797
+def main(argv: list[str] | None = None) -> int:
798
+ args = parse_args(argv)
799
+ try:
800
+ artifacts = run(args)
801
+ except Exception as exc: # pragma: no cover - CLI guard
802
+ print(f"map/reduce dry-run failed: {exc}", file=sys.stderr)
803
+ return 1
804
+ print(f"Map/reduce dry-run artifacts written to {args.output_dir}")
805
+ for name, path in artifacts.items():
806
+ print(f"{name}={path.as_posix()}")
807
+ return 0
808
+
809
+
810
+if __name__ == "__main__":
811
+ raise SystemExit(main())
tests/test_map_reduce_dry_run.py
new
+200
@@ -0,0 +1,200 @@
1
+from __future__ import annotations
2
+
3
+import json
4
+import tempfile
5
+from pathlib import Path
6
+
7
+from scripts import map_reduce_dry_run as dry_run
8
+
9
+
10
+def make_repo(owner: str, name: str, stars: int, gained: int = 0) -> dict[str, object]:
11
+ return {
12
+ "name": name,
13
+ "owner": owner,
14
+ "full_name": f"{owner}/{name}",
15
+ "description": f"{name} provides evidence-backed developer infrastructure for testing map reduce analysis contracts.",
16
+ "language": "Python",
17
+ "stars": stars,
18
+ "stars_gained": gained,
19
+ "created_at": "2026-05-18T10:00:00Z",
20
+ "topics": ["ai", "developer-tools"],
21
+ "url": f"https://github.com/{owner}/{name}",
22
+ }
23
+
24
+
25
+def test_dry_run_emits_valid_contract_artifacts() -> None:
26
+ tests_root = Path(__file__).resolve().parent
27
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
28
+ base = Path(tmpdir)
29
+ raw_path = base / "data" / "raw" / "2026-W21.json"
30
+ press_path = base / "data" / "analyzed" / "2026-W21-press-context.md"
31
+ output_dir = base / "data" / "candidates" / "2026-W21" / "local" / "map-reduce"
32
+ raw_path.parent.mkdir(parents=True)
33
+ press_path.parent.mkdir(parents=True)
34
+ raw_payload = {
35
+ "week": "2026-W21",
36
+ "crawled_at": "2026-05-20T12:00:00Z",
37
+ "new_repos": [make_repo("octo", "alpha", 1200), make_repo("octo", "beta", 900)],
38
+ "trending_repos": [make_repo("tools", "gamma", 5000, 450), make_repo("tools", "delta", 3000, 250)],
39
+ "signals": {"top_topics": ["ai", "developer-tools", "testing"]},
40
+ }
41
+ raw_path.write_text(json.dumps(raw_payload), encoding="utf-8")
42
+ press_path.write_text(
43
+ "### Correlation Summary\n- Industry article: https://example.com/ai-tooling links repo momentum to developer tools.\n",
44
+ encoding="utf-8",
45
+ )
46
+
47
+ rc = dry_run.main(
48
+ [
49
+ "--raw-json",
50
+ raw_path.as_posix(),
51
+ "--press-context",
52
+ press_path.as_posix(),
53
+ "--output-dir",
54
+ output_dir.as_posix(),
55
+ "--current-datetime",
56
+ "2026-05-20T12:00:00Z",
57
+ "--run-id",
58
+ "local",
59
+ ]
60
+ )
61
+
62
+ assert rc == 0
63
+ manifest = json.loads((output_dir / "manifest.json").read_text(encoding="utf-8"))
64
+ assert manifest["publish_eligible"] is False
65
+ assert manifest["candidate_only"] is True
66
+ for mapper in dry_run.MAPPER_IDS:
67
+ ledger = json.loads((output_dir / "maps" / f"{mapper}.json").read_text(encoding="utf-8"))
68
+ assert ledger["schema_version"] == "analysis_map_v1"
69
+ assert ledger["coverage"]["excluded_reason_counts"] == {}
70
+ assert dry_run.validate_map(ledger) == []
71
+ plan = json.loads((output_dir / "editorial-plan.json").read_text(encoding="utf-8"))
72
+ assert plan["schema_version"] == "analysis_editorial_plan_v1"
73
+ assert (output_dir / "sidecars" / "rejected-claims.json").exists()
74
+ assert (output_dir / "sidecars" / "contradictions.json").exists()
75
+ candidate = (output_dir / "2026-W21-map-reduce-candidate.md").read_text(encoding="utf-8")
76
+ assert "Map/reduce dry-run candidate only" in candidate
77
+ assert "[tools/gamma](https://github.com/tools/gamma)" in candidate
78
+ qa = json.loads((output_dir / "qa-comparison-report.json").read_text(encoding="utf-8"))
79
+ assert qa["status"] == "passed"
80
+ assert qa["publish_eligible"] is False
81
+ assert qa["checks"]["structural_analysis_gate"]["passed"] is True
82
+ assert qa["checks"]["evidence_and_editorial_gates"]["passed"] is True
83
+ assert qa["checks"]["publish_provenance_gate"]["expected_failure"] is True
84
+
85
+
86
+def test_validate_map_rejects_citationless_findings() -> None:
87
+ payload = {
88
+ "schema_version": "analysis_map_v1",
89
+ "run_id": "local",
90
+ "week": "2026-W21",
91
+ "shard_id": "signal-type:new_repos",
92
+ "slice": {},
93
+ "coverage": {"repo_ids_seen": [], "article_urls_seen": [], "excluded_reason_counts": {}},
94
+ "findings": [
95
+ {
96
+ "claim_id": "bad",
97
+ "claim": "unsupported",
98
+ "category": "trend",
99
+ "source_type": "github",
100
+ "evidence_refs": [],
101
+ "confidence": 0.5,
102
+ "contra_refs": [],
103
+ "uncertainties": [],
104
+ }
105
+ ],
106
+ "citations": [],
107
+ "reference_candidates": {"notable_projects": [], "press_articles": []},
108
+ "provenance": {},
109
+ }
110
+
111
+ assert "finding 0 has no evidence refs" in dry_run.validate_map(payload)
112
+
113
+
114
+def valid_ledger(findings: list[dict[str, object]] | None = None) -> dict[str, object]:
115
+ return {
116
+ "schema_version": "analysis_map_v1",
117
+ "run_id": "local",
118
+ "week": "2026-W21",
119
+ "shard_id": "signal-type:test",
120
+ "slice": {},
121
+ "coverage": {
122
+ "repo_ids_seen": ["octo/alpha"],
123
+ "article_urls_seen": [],
124
+ "repo_count_input": 1,
125
+ "repo_count_mapped": 1,
126
+ "article_count_input": 0,
127
+ "article_count_mapped": 0,
128
+ "excluded_reason_counts": {},
129
+ },
130
+ "findings": findings
131
+ if findings is not None
132
+ else [
133
+ {
134
+ "claim_id": "claim-a",
135
+ "claim": "octo/alpha is supported by direct repository evidence.",
136
+ "category": "trend",
137
+ "source_type": "github",
138
+ "evidence_refs": [{"type": "repo", "ref": "octo/alpha", "url": "https://github.com/octo/alpha"}],
139
+ "repo_full_name": "octo/alpha",
140
+ "news_url": None,
141
+ "confidence": 0.8,
142
+ "contra_refs": [],
143
+ "uncertainties": [],
144
+ }
145
+ ],
146
+ "citations": [],
147
+ "reference_candidates": {"notable_projects": ["octo/alpha"], "press_articles": []},
148
+ "provenance": {},
149
+ }
150
+
151
+
152
+def test_validate_map_rejects_malformed_ledger() -> None:
153
+ payload = valid_ledger()
154
+ payload.pop("coverage")
155
+ payload["findings"] = "not-a-list"
156
+
157
+ errors = dry_run.validate_map(payload)
158
+
159
+ assert "mapper missing coverage" in errors
160
+ assert "findings must be a list" in errors
161
+ assert "coverage must be an object" in errors
162
+
163
+
164
+def test_validate_map_rejects_failed_or_low_coverage() -> None:
165
+ payload = valid_ledger()
166
+ payload["coverage"] = {
167
+ "repo_ids_seen": ["octo/alpha"],
168
+ "article_urls_seen": [],
169
+ "repo_count_input": 3,
170
+ "repo_count_mapped": 1,
171
+ "article_count_input": 0,
172
+ "article_count_mapped": 1,
173
+ "excluded_reason_counts": {},
174
+ }
175
+ payload["status"] = "failed"
176
+
177
+ errors = dry_run.validate_map(payload)
178
+
179
+ assert "coverage repo_count_mapped below repo_count_input without excluded reasons" in errors
180
+ assert "coverage article_count_mapped exceeds article_count_input" in errors
181
+ assert "mapper status failed" in errors
182
+
183
+
184
+def test_reduce_rejects_and_preserves_contradictory_claims() -> None:
185
+ supported = valid_ledger()["findings"][0]
186
+ contradictory = {
187
+ **supported,
188
+ "claim_id": "claim-b",
189
+ "claim": "octo/alpha evidence is contradicted by another retained source.",
190
+ "contra_refs": ["claim-a"],
191
+ "confidence": 0.9,
192
+ }
193
+ ledger = valid_ledger([supported, contradictory])
194
+
195
+ plan, rejected, contradictions = dry_run.reduce_ledgers([ledger], raw_payload={"week": "2026-W21", "new_repos": [make_repo("octo", "alpha", 1200)], "trending_repos": []})
196
+
197
+ assert plan["selected_claims"] == []
198
+ assert [item["claim_id"] for item in contradictions] == ["claim-a", "claim-b"]
199
+ assert {item["claim_id"] for item in rejected if item["reason"] == "unresolved_contradiction"} == {"claim-a", "claim-b"}
200
+ assert plan["contradictions"] == contradictions