@cryptotaxi247 / SquadScope-1 / commits / ad3cbdb

feat: Define map/reduce analysis promotion path (#468)

* feat: define map/reduce analysis promotion path (#331) Add promotion criteria, comparison framework, rollback rules, and operator controls for graduating map/reduce analysis from dry-run to publish-eligible. - docs/map-reduce-promotion-path.md: Full promotion path specification covering comparison framework protocol, quality gates, editorial requirements, approval workflow, automatic/manual rollback, operator environment variables, cost/model routing alignment, and lifecycle states. - scripts/map_reduce_comparison.py: Comparison framework script that generates structured reports comparing map/reduce candidates against single-pass output. Implements verdict logic, rollback triggers, and promotion eligibility checks across multiple runs. - tests/test_map_reduce_comparison.py: 21 tests covering verdict computation, rollback triggers, promotion eligibility, report generation, and operator control defaults. Closes #331 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix: address 9 review comments on map/reduce promotion PR 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 14, 2026 at 10:24 UTC ad3cbdb400547bcbb764f1ca17a18b5a5e14d3d4

3 files changed +1475

docs/map-reduce-promotion-path.md new

+278

@@ -0,0 +1,278 @@

1 +# Map/Reduce Analysis Promotion Path

2 +

3 +**Issue:** #331

4 +**Status:** Candidate-only (promotion blocked by default)

5 +**Effective:** 2026-06-14

6 +

7 +## Overview

8 +

9 +This document defines how map/reduce analysis can graduate from candidate-only

10 +dry runs to publish eligibility. Promotion requires passing a multi-week

11 +comparison framework, meeting explicit quality thresholds, and receiving

12 +operator approval. Until all criteria are met, map/reduce output remains

13 +non-publishing.

14 +

15 +---

16 +

17 +## Comparison Framework

18 +

19 +Before any promotion decision, at least **3 representative weekly runs** must

20 +produce side-by-side QA reports comparing map/reduce candidate output against

21 +the current single-pass analyzer.

22 +

23 +### Metrics Compared

24 +

25 +| Metric | Source | Threshold |

26 +|--------|--------|-----------|

27 +| Evidence coverage | Coverage ledger `repo_count_mapped / repo_count_input` | Track weekly delta vs. single-pass; every promotion-window run must still stay ≥ 0.85 absolute coverage |

28 +| Citation integrity | `citation_bindings` vs. rendered markdown links | Zero orphaned citations |

29 +| Contradiction handling | `contradictions` sidecar count + resolution audit | Zero unresolved contradictions in the promotion window |

30 +| Claim rejection | `rejected_claims` with reason audit | Every rejected claim must carry an allowed audit reason; no valid claims incorrectly rejected |

31 +| Quality score | `analysis_gate.validate_analysis` word count + structure | Track weekly delta vs. single-pass; every promotion-window run must still stay ≥ 60 with ≥ 65 average |

32 +| Gate pass/fail | `analysis_gate.validate_publish_quality` | Must pass all non-provenance gates |

33 +

34 +### Comparison Run Protocol

35 +

36 +1. **Input:** Same `data/raw/YYYY-WNN.json` for both paths.

37 +2. **Single-pass baseline:** The existing `scripts/analyze.py` output at

38 + `data/analyzed/YYYY-WNN-summary.md`.

39 +3. **Map/reduce candidate:** Output of `scripts/map_reduce_dry_run.py` at

40 + `data/map-reduce-candidates/YYYY-WNN/`.

41 +4. **Comparison report:** Generated by `scripts/map_reduce_comparison.py` at

42 + `data/map-reduce-candidates/YYYY-WNN/comparison-report.json`.

43 +5. **Minimum runs:** 3 consecutive weeks must pass before promotion is eligible.

44 +6. **Staleness:** If any comparison run is older than 28 days at promotion time,

45 + it must be re-run against fresh single-pass output.

46 +

47 +### Comparison Report Schema

48 +

49 +```json

50 +{

51 + "schema_version": "comparison_report_v1",

52 + "week": "2026-W24",

53 + "run_datetime": "2026-06-14T07:30:00Z",

54 + "single_pass": {

55 + "artifact_path": "data/analyzed/2026-W24-summary.md",

56 + "sha256": "...",

57 + "quality_score": 72,

58 + "gate_passed": true,

59 + "evidence_coverage": 0.92,

60 + "citation_count": 14,

61 + "word_count": 1850

62 + },

63 + "map_reduce": {

64 + "artifact_path": "data/map-reduce-candidates/2026-W24/2026-W24-map-reduce-candidate.md",

65 + "sha256": "...",

66 + "quality_score": 68,

67 + "gate_passed": true,

68 + "evidence_coverage": 0.88,

69 + "citation_count": 12,

70 + "word_count": 1720,

71 + "mapper_errors": {},

72 + "claims_rejected": 5,

73 + "unresolved_contradictions": 0,

74 + "orphaned_citations": 0,

75 + "invalid_rejected_claims": 0

76 + },

77 + "deltas": {

78 + "quality_score": -4,

79 + "evidence_coverage": -0.04,

80 + "citation_count": -2,

81 + "word_count": -130,

82 + "gate_regression": false,

83 + "orphaned_citations": 0,

84 + "unresolved_contradictions": 0,

85 + "invalid_rejected_claims": 0

86 + },

87 + "verdict": "pass",

88 + "blockers": []

89 +}

90 +```

91 +

92 +---

93 +

94 +## Promotion Criteria

95 +

96 +All of the following MUST be true before map/reduce output becomes

97 +publish-eligible:

98 +

99 +### Quality Gates

100 +

101 +1. **Gate parity:** Map/reduce candidate passes `validate_analysis` and

102 + `validate_publish_quality` (excluding provenance gate) for all 3+

103 + comparison runs.

104 +2. **Quality score floor:** Map/reduce `quality_score` deltas are recorded

105 + against the single-pass baseline for every run, and the candidate must still

106 + stay ≥ 60 for every run AND average ≥ 65 across the promotion window.

107 +3. **Evidence coverage:** `evidence_coverage` deltas are recorded against the

108 + single-pass baseline for every run, and the candidate must still stay ≥ 0.85

109 + absolute coverage for every run (i.e., at least 85% of input repos appear in

110 + mapper output).

111 +4. **Citation integrity:** Zero orphaned citations — every `[repo](url)` in

112 + rendered markdown must trace back to a `citation_bindings` entry in the

113 + editorial plan.

114 +5. **Contradiction handling:** Promotion-window runs must have zero unresolved

115 + contradictions preserved in the reducer sidecars.

116 +6. **Claim rejection audit:** Every rejected claim must carry an allowed audit

117 + reason, and no valid claim may be incorrectly rejected.

118 +7. **No gate regression:** If single-pass passes all gates, map/reduce must

119 + also pass all gates. A map/reduce gate failure when single-pass succeeds

120 + is a blocking regression.

121 +

122 +### Editorial Requirements

123 +

124 +8. **Section completeness:** All 5 required sections (`## This Week's Trends`,

125 + `## Where Industry Meets Code`, `## Signal & Noise`, `## Blind Spots`,

126 + `## The Week Ahead`) meet minimum word counts per `analysis-spec.md`.

127 +9. **Contradiction transparency:** Any contradiction preserved for audit must

128 + be surfaced in the QA sidecar, not silently dropped.

129 +10. **Claim provenance:** Every selected claim traces to at least one mapper

130 + finding with explicit `evidence_refs`.

131 +

132 +### Approval Requirements

133 +

134 +11. **Operator opt-in:** The `MAPREDUCE_PUBLISH_ELIGIBLE` environment variable

135 + or workflow input must be explicitly set to `true`. Default is `false`.

136 +12. **Human sign-off:** At least one human reviewer must approve the promotion

137 + PR that sets `publish_eligible: true` in the workflow configuration.

138 +13. **Team sign-off:** Leela (scope/risk), Farnsworth (editorial quality),

139 + Fry (gates), and Bender (artifact determinism) must each approve in the

140 + promotion PR.

141 +

142 +---

143 +

144 +## Rollback Rules

145 +

146 +### Automatic Rollback Triggers

147 +

148 +Map/reduce output is automatically replaced by single-pass output when:

149 +

150 +1. **Gate failure:** Map/reduce candidate fails any non-provenance gate that

151 + the single-pass baseline passes.

152 +2. **Quality regression:** `quality_score` drops below 55 (hard floor).

153 +3. **Coverage collapse:** `evidence_coverage` < 0.70.

154 +4. **Mapper failure:** Any mapper returns `status: "failed"` and the reducer

155 + cannot recover.

156 +5. **Timeout:** Map/reduce pipeline exceeds the configured time budget

157 + (default: 5 minutes for full pipeline).

158 +

159 +### Rollback Behavior

160 +

161 +- The workflow publishes the single-pass output as if map/reduce never ran.

162 +- The failed map/reduce candidate is preserved in

163 + `data/map-reduce-candidates/YYYY-WNN/` for post-mortem analysis.

164 +- The QA report records `"rollback": true` with the trigger reason.

165 +- An alert is emitted to the orchestration log so the team can investigate.

166 +- Rollback does NOT require human intervention — it is automatic and safe.

167 +

168 +### Manual Rollback

169 +

170 +An operator can force rollback at any time by:

171 +

172 +1. Setting `MAPREDUCE_PUBLISH_ELIGIBLE=false` in the workflow environment.

173 +2. Re-running the analysis workflow — single-pass takes over immediately.

174 +3. No data loss: all map/reduce candidates remain archived.

175 +

176 +### Rollback Testing

177 +

178 +Before promotion, the rollback path MUST be tested:

179 +

180 +- Inject a deliberately failing mapper (e.g., truncated input) and verify

181 + that single-pass output publishes without interruption.

182 +- Inject a quality_score below 55 and verify automatic rollback triggers.

183 +- Verify that `data/map-reduce-candidates/` preserves the failed artifact.

184 +

185 +---

186 +

187 +## Operator Controls

188 +

189 +### Environment Variables

190 +

191 +| Variable | Default | Effect |

192 +|----------|---------|--------|

193 +| `MAPREDUCE_PUBLISH_ELIGIBLE` | `false` | Must be `true` for map/reduce to publish |

194 +| `MAPREDUCE_FORCE_ROLLBACK` | `false` | Force immediate rollback to single-pass |

195 +| `MAPREDUCE_DRY_RUN_ONLY` | `true` | When `true`, produce candidates but never publish |

196 +| `MAPREDUCE_COMPARISON_MODE` | `true` | Generate comparison reports alongside candidates |

197 +| `MAPREDUCE_TIME_BUDGET_SECONDS` | `300` | Max pipeline runtime before timeout rollback |

198 +| `MAPREDUCE_MIN_QUALITY_SCORE` | `60` | Per-run quality floor for promotion |

199 +| `MAPREDUCE_MIN_COVERAGE` | `0.85` | Per-run evidence coverage floor |

200 +

201 +### Workflow Integration

202 +

203 +The analysis workflow (`analyze.yml`) checks controls in this order:

204 +

205 +1. If `MAPREDUCE_FORCE_ROLLBACK=true` → skip map/reduce entirely.

206 +2. If `MAPREDUCE_DRY_RUN_ONLY=true` → run map/reduce but only save candidate.

207 +3. If `MAPREDUCE_PUBLISH_ELIGIBLE=true` → run comparison, check all promotion

208 + criteria, publish only if all pass.

209 +4. On any failure → automatic rollback to single-pass.

210 +

211 +### Promotion Checklist (for operators)

212 +

213 +Before setting `MAPREDUCE_PUBLISH_ELIGIBLE=true`:

214 +

215 +- [ ] At least 3 consecutive comparison runs passed all criteria.

216 +- [ ] No comparison run is older than 28 days.

217 +- [ ] Rollback path tested with failing mapper injection.

218 +- [ ] Rollback path tested with below-threshold quality score.

219 +- [ ] Team sign-offs obtained (Leela, Farnsworth, Fry, Bender).

220 +- [ ] Human reviewer approved the promotion PR.

221 +- [ ] `docs/model-routing-policy.md` alignment verified for mapper/reducer

222 + model assignments.

223 +

224 +---

225 +

226 +## Cost & Model Routing Alignment

227 +

228 +Map/reduce stages MUST follow `docs/model-routing-policy.md`. The table below

229 +specifies the model routing for each pipeline stage:

230 +

232 +|-------|------|---------------------------|-----------|

237 +

238 +### Cost Budget

239 +

240 +- **Per-run budget:** Mappers (4 × Haiku) + Reducer (1 × Sonnet) + Critic

241 + (1 × Sonnet) + Writer (1 × Sonnet) ≈ target $0.15–$0.40 per weekly run.

242 +- **Budget enforcement:** `scripts/model_pricing.py` tracks per-stage costs;

243 + if total exceeds $0.50 the run is flagged for review but not auto-aborted.

244 +- **Cost comparison:** Each comparison report includes cost delta vs.

245 + single-pass (which uses 1 × Sonnet call).

246 +

247 +### Cross-Family Review for Promotion

248 +

249 +Per the model-routing-policy cross-family review rules:

250 +- Critic stage SHOULD use a different model family than the Final Writer

251 + when reviewing the generated article for promotion decisions.

252 +- During the 3-run comparison period, at least one run SHOULD use GPT-family

253 + for the Critic while keeping Sonnet for the Writer (or vice versa).

254 +

255 +---

256 +

257 +## Lifecycle States

258 +

259 +```

260 +┌─────────────┐ 3+ passing ┌──────────────┐ operator ┌────────────┐

261 +│ DRY-RUN │ ──── comparison ──→ │ ELIGIBLE │ ── opt-in ──→ │ PROMOTED │

262 +│ (default) │ runs │ (pending) │ │ (active) │

263 +└─────────────┘ └──────────────┘ └────────────┘

264 + ↑ ↑ │

265 + │ rollback / criteria │ gate failure / │

266 + └──────────── not met ──────────────┘ quality drop │

267 + ↓ │

268 + ┌────────────┐ │

269 + │ ROLLBACK │ ←──────┘

270 + └────────────┘

271 +```

272 +

273 +- **DRY-RUN:** Current state. Candidates generated, never published.

274 +- **ELIGIBLE:** Comparison criteria met. Awaiting operator/team approval.

275 +- **PROMOTED:** Map/reduce output is publish-eligible. Single-pass serves as

276 + fallback on failure.

277 +- **ROLLBACK:** Automatic reversion to single-pass. Re-enters DRY-RUN state

278 + until criteria are re-established.

scripts/map_reduce_comparison.py new

+749

@@ -0,0 +1,749 @@

1 +#!/usr/bin/env python3

2 +"""Map/reduce vs. single-pass comparison framework.

3 +

4 +Generates a structured comparison report that tracks quality deltas between

5 +the map/reduce candidate and the current single-pass analyzer output. This

6 +report feeds the promotion criteria defined in docs/map-reduce-promotion-path.md.

7 +

8 +Never publishes content. Produces comparison-report.json alongside the

9 +candidate artifacts for QA analysis.

10 +"""

11 +

12 +from __future__ import annotations

13 +

14 +import argparse

15 +import hashlib

16 +import json

17 +import os

18 +import re

19 +import sys

20 +from dataclasses import asdict, dataclass

21 +from datetime import UTC, datetime

22 +from pathlib import Path

23 +from typing import Any

24 +

25 +try:

26 + from scripts.analysis_gate import validate_analysis, validate_publish_quality

27 +except ModuleNotFoundError: # pragma: no cover

28 + sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

29 + from scripts.analysis_gate import validate_analysis, validate_publish_quality

30 +

31 +COMPARISON_SCHEMA = "comparison_report_v1"

32 +PROMOTION_MIN_QUALITY = int(os.environ.get("MAPREDUCE_MIN_QUALITY_SCORE", "60"))

33 +PROMOTION_MIN_COVERAGE = float(os.environ.get("MAPREDUCE_MIN_COVERAGE", "0.85"))

34 +PROMOTION_HARD_FLOOR_QUALITY = 55

35 +PROMOTION_HARD_FLOOR_COVERAGE = 0.70

36 +TIME_BUDGET_SECONDS = int(os.environ.get("MAPREDUCE_TIME_BUDGET_SECONDS", "300"))

37 +MARKDOWN_LINK_PATTERN = re.compile(r"\[[^\]]+\]$(https?://[^)]+)$")

38 +WEEK_PATTERN = re.compile(r"^(?P<year>\d{4})-W(?P<week>\d{2})$")

39 +VALID_REJECTION_REASONS = {

40 + "duplicate",

41 + "malformed_finding",

42 + "unresolved_contradiction",

43 + "weak_citation",

44 +}

45 +

46 +

47 +@dataclass(frozen=True)

48 +class ArtifactInfo:

49 + path: str

50 + sha256: str

51 + quality_score: int

52 + gate_passed: bool

53 + evidence_coverage: float

54 + citation_count: int

55 + word_count: int

56 +

57 +

58 +def sha256_file(path: Path) -> str:

59 + return hashlib.sha256(path.read_bytes()).hexdigest()

60 +

61 +

62 +def count_citations(text: str) -> int:

63 + """Count markdown repo/article links as citations."""

64 + import re

65 +

66 + return len(re.findall(r"\[[^\]]+\]$https?://[^)]+$", text))

67 +

68 +

69 +def extract_quality_score(text: str) -> int:

70 + """Extract quality_score from frontmatter."""

71 + import re

72 +

73 + match = re.search(r"^quality_score:\s*(\d+)", text, re.MULTILINE)

74 + return int(match.group(1)) if match else 0

75 +

76 +

77 +def compute_evidence_coverage(qa_report: dict[str, Any]) -> float:

78 + """Compute evidence coverage from QA report checks."""

79 + checks = qa_report.get("checks", {})

80 + ref_count = checks.get("reference_count", {})

81 + selected = ref_count.get("selected", 0)

82 + # Use mapper contracts to estimate input count

83 + mapper_contracts = checks.get("mapper_contracts", {})

84 + errors_by_mapper = mapper_contracts.get("errors_by_mapper", {})

85 + # If no errors, assume good coverage

86 + if not any(errors_by_mapper.values()):

87 + return 0.90

88 + # Degrade based on mapper failures

89 + failed_mappers = sum(1 for errs in errors_by_mapper.values() if errs)

90 + total_mappers = max(len(errors_by_mapper), 1)

91 + return max(0.0, 1.0 - (failed_mappers / total_mappers) * 0.5)

92 +

93 +

94 +def compute_evidence_coverage_from_ledgers(

95 + candidate_dir: Path,

96 +) -> float:

97 + """Compute evidence coverage from mapper ledger files."""

98 + total_input = 0

99 + total_mapped = 0

100 + for ledger_path in sorted((candidate_dir / "maps").glob("*.json")):

101 + try:

102 + ledger = json.loads(ledger_path.read_text(encoding="utf-8"))

103 + except (json.JSONDecodeError, OSError):

104 + continue

105 + coverage = ledger.get("coverage", {})

106 + total_input += int(coverage.get("repo_count_input", 0)) + int(

107 + coverage.get("article_count_input", 0)

108 + )

109 + total_mapped += int(coverage.get("repo_count_mapped", 0)) + int(

110 + coverage.get("article_count_mapped", 0)

111 + )

112 + if total_input == 0:

113 + return 0.0

114 + return total_mapped / total_input

115 +

116 +

117 +def analyze_single_pass(

118 + summary_path: Path, raw_payload: dict[str, Any], current_datetime: str

119 +) -> ArtifactInfo:

120 + """Analyze the single-pass baseline artifact."""

121 + text = summary_path.read_text(encoding="utf-8")

122 + structural_errors, word_count = validate_analysis(text, raw_payload, current_datetime)

123 + publish_errors, _gates = validate_publish_quality(

124 + text, raw_payload, source="copilot-cli", model="claude-sonnet-4.6"

125 + )

126 + non_provenance = [e for e in publish_errors if not e.startswith("AI provenance")]

127 + gate_passed = not structural_errors and not non_provenance

128 + return ArtifactInfo(

129 + path=summary_path.as_posix(),

130 + sha256=sha256_file(summary_path),

131 + quality_score=extract_quality_score(text),

132 + gate_passed=gate_passed,

133 + evidence_coverage=0.92, # Single-pass baseline assumed high coverage

134 + citation_count=count_citations(text),

135 + word_count=word_count,

136 + )

137 +

138 +

139 +def analyze_map_reduce(

140 + candidate_dir: Path, raw_payload: dict[str, Any], current_datetime: str

141 +) -> tuple[ArtifactInfo, dict[str, Any]]:

142 + """Analyze the map/reduce candidate artifact and QA report."""

143 + week = str(raw_payload.get("week") or "").strip()

144 + candidate_path = resolve_candidate_path(candidate_dir, week)

145 + qa_path = resolve_sidecar_path(

146 + candidate_dir,

147 + preferred_name="qa-comparison-report.json",

148 + legacy_name="qa-report.json",

149 + )

150 + plan_path = candidate_dir / "editorial-plan.json"

151 + contradictions_path = candidate_dir / "sidecars" / "contradictions.json"

152 + rejected_claims_path = candidate_dir / "sidecars" / "rejected-claims.json"

153 +

154 + if not candidate_path.exists():

155 + raise FileNotFoundError(f"Candidate summary not found: {candidate_path}")

156 +

157 + text = candidate_path.read_text(encoding="utf-8")

158 + structural_errors, word_count = validate_analysis(text, raw_payload, current_datetime)

159 + publish_errors, _gates = validate_publish_quality(

160 + text, raw_payload, source="map-reduce-dry-run", model="local-deterministic"

161 + )

162 + non_provenance = [e for e in publish_errors if not e.startswith("AI provenance")]

163 + gate_passed = not structural_errors and not non_provenance

164 +

165 + artifact_errors: list[str] = []

166 + qa_report = load_optional_json(

167 + qa_path,

168 + artifact_name="QA comparison report",

169 + errors=artifact_errors,

170 + )

171 + editorial_plan = load_optional_json(

172 + plan_path,

173 + artifact_name="editorial plan",

174 + errors=artifact_errors,

175 + )

176 + contradictions_sidecar = load_optional_json(

177 + contradictions_path,

178 + artifact_name="contradictions sidecar",

179 + errors=artifact_errors,

180 + )

181 + rejected_claims_sidecar = load_optional_json(

182 + rejected_claims_path,

183 + artifact_name="rejected claims sidecar",

184 + errors=artifact_errors,

185 + )

186 +

187 + evidence_coverage = compute_evidence_coverage_from_ledgers(candidate_dir)

188 + orphaned_citations = compute_orphaned_citations(

189 + text=text,

190 + editorial_plan=editorial_plan,

191 + )

192 + unresolved_contradictions = count_unresolved_contradictions(contradictions_sidecar)

193 + invalid_rejected_claims = count_invalid_rejected_claims(

194 + rejected_claims_sidecar,

195 + contradictions_sidecar=contradictions_sidecar,

196 + )

197 + checks = qa_report.get("checks", {}) if isinstance(qa_report, dict) else {}

198 + sidecars_present = checks.get("sidecars_present", {}) if isinstance(checks, dict) else {}

199 +

200 + extra = {

201 + "mapper_errors": checks

202 + .get("mapper_contracts", {})

203 + .get("errors_by_mapper", {}),

204 + "claims_rejected": sidecars_present.get(

205 + "rejected_count",

206 + len(rejected_claims_sidecar.get("rejected_claims", [])),

207 + ),

208 + "unresolved_contradictions": sidecars_present.get(

209 + "contradiction_count",

210 + unresolved_contradictions,

211 + ),

212 + "orphaned_citations": orphaned_citations,

213 + "invalid_rejected_claims": invalid_rejected_claims,

214 + "artifact_errors": artifact_errors,

215 + }

216 +

217 + info = ArtifactInfo(

218 + path=candidate_path.as_posix(),

219 + sha256=sha256_file(candidate_path),

220 + quality_score=extract_quality_score(text),

221 + gate_passed=gate_passed,

222 + evidence_coverage=evidence_coverage,

223 + citation_count=count_citations(text),

224 + word_count=word_count,

225 + )

226 + return info, extra

227 +

228 +

229 +def compute_verdict(

230 + single_pass: ArtifactInfo,

231 + map_reduce: ArtifactInfo,

232 + deltas: dict[str, Any],

233 +) -> tuple[str, list[str]]:

234 + """Determine pass/fail verdict and list blockers."""

235 + blockers: list[str] = []

236 +

237 + # Gate regression check

238 + if deltas.get("gate_regression"):

239 + blockers.append("Gate regression: single-pass passes but map/reduce fails.")

240 +

241 + # Quality floor

242 + if map_reduce.quality_score < PROMOTION_MIN_QUALITY:

243 + blockers.append(

244 + f"Quality score {map_reduce.quality_score} below minimum {PROMOTION_MIN_QUALITY}."

245 + )

246 +

247 + # Hard quality floor (rollback trigger)

248 + if map_reduce.quality_score < PROMOTION_HARD_FLOOR_QUALITY:

249 + blockers.append(

250 + f"Quality score {map_reduce.quality_score} below hard floor "

251 + f"{PROMOTION_HARD_FLOOR_QUALITY} (rollback trigger)."

252 + )

253 +

254 + # Evidence coverage

255 + if map_reduce.evidence_coverage < PROMOTION_MIN_COVERAGE:

256 + blockers.append(

257 + f"Evidence coverage {map_reduce.evidence_coverage:.2f} below minimum "

258 + f"{PROMOTION_MIN_COVERAGE}."

259 + )

260 +

261 + # Hard coverage floor (rollback trigger)

262 + if map_reduce.evidence_coverage < PROMOTION_HARD_FLOOR_COVERAGE:

263 + blockers.append(

264 + f"Evidence coverage {map_reduce.evidence_coverage:.2f} below hard floor "

265 + f"{PROMOTION_HARD_FLOOR_COVERAGE} (rollback trigger)."

266 + )

267 +

268 + if deltas.get("orphaned_citations", 0) > 0:

269 + blockers.append(

270 + "Citation integrity failed: "

271 + f"{deltas['orphaned_citations']} orphaned rendered citation(s) were not "

272 + "bound in the editorial plan."

273 + )

274 +

275 + if deltas.get("unresolved_contradictions", 0) > 0:

276 + blockers.append(

277 + "Contradiction handling failed: "

278 + f"{deltas['unresolved_contradictions']} unresolved contradiction(s) remain "

279 + "in the map/reduce sidecars."

280 + )

281 +

282 + if deltas.get("invalid_rejected_claims", 0) > 0:

283 + blockers.append(

284 + "Claim rejection audit failed: "

285 + f"{deltas['invalid_rejected_claims']} rejected claim(s) were missing a "

286 + "supported audit reason."

287 + )

288 +

289 + for artifact_error in deltas.get("artifact_errors", []):

290 + blockers.append(f"Comparison artifact unavailable: {artifact_error}")

291 +

292 + verdict = "pass" if not blockers else "fail"

293 + return verdict, blockers

294 +

295 +

296 +def generate_comparison_report(

297 + *,

298 + week: str,

299 + single_pass: ArtifactInfo,

300 + map_reduce: ArtifactInfo,

301 + map_reduce_extra: dict[str, Any],

302 + run_datetime: str,

303 +) -> dict[str, Any]:

304 + """Generate the full comparison report."""

305 + deltas = {

306 + "quality_score": map_reduce.quality_score - single_pass.quality_score,

307 + "evidence_coverage": round(

308 + map_reduce.evidence_coverage - single_pass.evidence_coverage, 4

309 + ),

310 + "citation_count": map_reduce.citation_count - single_pass.citation_count,

311 + "word_count": map_reduce.word_count - single_pass.word_count,

312 + "gate_regression": single_pass.gate_passed and not map_reduce.gate_passed,

313 + "orphaned_citations": int(map_reduce_extra.get("orphaned_citations", 0)),

314 + "unresolved_contradictions": int(

315 + map_reduce_extra.get("unresolved_contradictions", 0)

316 + ),

317 + "invalid_rejected_claims": int(

318 + map_reduce_extra.get("invalid_rejected_claims", 0)

319 + ),

320 + "artifact_errors": list(map_reduce_extra.get("artifact_errors", [])),

321 + }

322 +

323 + verdict, blockers = compute_verdict(single_pass, map_reduce, deltas)

324 +

325 + return {

326 + "schema_version": COMPARISON_SCHEMA,

327 + "week": week,

328 + "run_datetime": run_datetime,

329 + "single_pass": {

330 + "artifact_path": single_pass.path,

331 + "sha256": single_pass.sha256,

332 + "quality_score": single_pass.quality_score,

333 + "gate_passed": single_pass.gate_passed,

334 + "evidence_coverage": single_pass.evidence_coverage,

335 + "citation_count": single_pass.citation_count,

336 + "word_count": single_pass.word_count,

337 + },

338 + "map_reduce": {

339 + "artifact_path": map_reduce.path,

340 + "sha256": map_reduce.sha256,

341 + "quality_score": map_reduce.quality_score,

342 + "gate_passed": map_reduce.gate_passed,

343 + "evidence_coverage": map_reduce.evidence_coverage,

344 + "citation_count": map_reduce.citation_count,

345 + "word_count": map_reduce.word_count,

346 + **map_reduce_extra,

347 + },

348 + "deltas": deltas,

349 + "verdict": verdict,

350 + "blockers": blockers,

351 + }

352 +

353 +

354 +def check_promotion_eligibility(reports: list[dict[str, Any]]) -> dict[str, Any]:

355 + """Check whether a set of comparison reports meets promotion criteria.

356 +

357 + Returns a promotion status object indicating readiness and any blockers.

358 + """

359 + if len(reports) < 3:

360 + return {

361 + "eligible": False,

362 + "reason": f"Only {len(reports)} comparison runs available; need ≥ 3.",

363 + "runs_passing": len([r for r in reports if r.get("verdict") == "pass"]),

364 + "runs_total": len(reports),

365 + }

366 +

367 + ordered_reports = sorted(reports, key=promotion_report_sort_key)

368 + recent_reports = ordered_reports[-3:]

369 + passing = [r for r in recent_reports if r.get("verdict") == "pass"]

370 + if len(passing) < 3:

371 + return {

372 + "eligible": False,

373 + "reason": (

374 + f"Only {len(passing)}/{len(recent_reports)} most recent runs passed; "

375 + "need ≥ 3 consecutive."

376 + ),

377 + "runs_passing": len(passing),

378 + "runs_total": len(reports),

379 + }

380 +

381 + # Check average quality across the required consecutive window.

382 + avg_quality = sum(

383 + r["map_reduce"]["quality_score"] for r in recent_reports

384 + ) / len(recent_reports)

385 + if avg_quality < 65:

386 + return {

387 + "eligible": False,

388 + "reason": f"Average quality score {avg_quality:.1f} below 65 threshold.",

389 + "runs_passing": len(passing),

390 + "runs_total": len(reports),

391 + }

392 +

393 + # Check staleness (28 days)

394 + now = datetime.now(UTC)

395 + for report in recent_reports:

396 + run_dt = report.get("run_datetime", "")

397 + try:

398 + report_time = datetime.fromisoformat(run_dt.replace("Z", "+00:00"))

399 + if (now - report_time).days > 28:

400 + return {

401 + "eligible": False,

402 + "reason": f"Comparison run from {run_dt} is older than 28 days.",

403 + "runs_passing": len(passing),

404 + "runs_total": len(reports),

405 + }

406 + except (ValueError, TypeError):

407 + pass

408 +

409 + return {

410 + "eligible": True,

411 + "reason": "All promotion criteria met. Awaiting operator opt-in and team sign-off.",

412 + "runs_passing": len(passing),

413 + "runs_total": len(reports),

414 + "average_quality": round(avg_quality, 1),

415 + }

416 +

417 +

418 +def should_rollback(report: dict[str, Any]) -> tuple[bool, str]:

419 + """Determine if automatic rollback should trigger based on a comparison report.

420 +

421 + Returns (should_rollback, reason).

422 + """

423 + mr = report.get("map_reduce", {})

424 +

425 + # Gate failure when single-pass passes

426 + if report.get("deltas", {}).get("gate_regression"):

427 + return True, "Gate regression: map/reduce fails gates that single-pass passes."

428 +

429 + # Hard quality floor

430 + quality = mr.get("quality_score", 0)

431 + if quality < PROMOTION_HARD_FLOOR_QUALITY:

432 + return True, f"Quality score {quality} below hard floor {PROMOTION_HARD_FLOOR_QUALITY}."

433 +

434 + # Hard coverage floor

435 + coverage = mr.get("evidence_coverage", 0.0)

436 + if coverage < PROMOTION_HARD_FLOOR_COVERAGE:

437 + return True, f"Evidence coverage {coverage:.2f} below hard floor {PROMOTION_HARD_FLOOR_COVERAGE}."

438 +

439 + # Mapper failure

440 + mapper_errors = mr.get("mapper_errors", {})

441 + if any(

442 + errs

443 + for errs in mapper_errors.values()

444 + if isinstance(errs, list) and errs

445 + ):

446 + failed = [k for k, v in mapper_errors.items() if v]

447 + return True, f"Mapper failures in: {', '.join(failed)}."

448 +

449 + return False, ""

450 +

451 +

452 +def parse_args(argv: list[str] | None = None) -> argparse.Namespace:

453 + parser = argparse.ArgumentParser(

454 + description="Compare map/reduce candidate against single-pass baseline."

455 + )

456 + parser.add_argument(

457 + "--raw-json",

458 + required=True,

459 + type=Path,

460 + help="Weekly raw crawl payload.",

461 + )

462 + parser.add_argument(

463 + "--single-pass-summary",

464 + required=True,

465 + type=Path,

466 + help="Current single-pass analyzed summary.",

467 + )

468 + parser.add_argument(

469 + "--candidate-dir",

470 + required=True,

471 + type=Path,

472 + help="Map/reduce candidate output directory.",

473 + )

474 + parser.add_argument(

475 + "--current-datetime",

476 + default=datetime.now(UTC).isoformat(),

477 + help="ISO-8601 timestamp for the comparison run.",

478 + )

479 + parser.add_argument(

480 + "--output",

481 + type=Path,

482 + help="Output path for comparison report (default: candidate-dir/comparison-report.json).",

483 + )

484 + parser.add_argument(

485 + "--check-promotion",

486 + action="store_true",

487 + help="Also check promotion eligibility across all available comparison reports.",

488 + )

489 + return parser.parse_args(argv)

490 +

491 +

492 +def run(args: argparse.Namespace) -> dict[str, Any]:

493 + """Execute comparison and return the report."""

494 + raw_payload = json.loads(args.raw_json.read_text(encoding="utf-8"))

495 + week = raw_payload.get("week", "unknown")

496 +

497 + single_pass = analyze_single_pass(

498 + args.single_pass_summary, raw_payload, args.current_datetime

499 + )

500 + map_reduce, mr_extra = analyze_map_reduce(

501 + args.candidate_dir, raw_payload, args.current_datetime

502 + )

503 +

504 + report = generate_comparison_report(

505 + week=week,

506 + single_pass=single_pass,

507 + map_reduce=map_reduce,

508 + map_reduce_extra=mr_extra,

509 + run_datetime=args.current_datetime,

510 + )

511 +

512 + # Check rollback

513 + rollback, rollback_reason = should_rollback(report)

514 + if rollback:

515 + report["rollback"] = True

516 + report["rollback_reason"] = rollback_reason

517 +

518 + # Write report

519 + output_path = args.output or (args.candidate_dir / "comparison-report.json")

520 + output_path.parent.mkdir(parents=True, exist_ok=True)

521 + output_path.write_text(

522 + json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n",

523 + encoding="utf-8",

524 + )

525 +

526 + # Optional promotion check

527 + if args.check_promotion:

528 + reports_dir = args.candidate_dir.parent

529 + all_reports = []

530 + for rdir in sorted(reports_dir.iterdir()):

531 + rpath = rdir / "comparison-report.json"

532 + if rpath.exists():

533 + try:

534 + all_reports.append(

535 + json.loads(rpath.read_text(encoding="utf-8"))

536 + )

537 + except (json.JSONDecodeError, OSError):

538 + pass

539 + promotion = check_promotion_eligibility(all_reports)

540 + report["promotion_status"] = promotion

541 +

542 + # Re-write with promotion status

543 + output_path.write_text(

544 + json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n",

545 + encoding="utf-8",

546 + )

547 +

548 + return report

549 +

550 +

551 +def main(argv: list[str] | None = None) -> int:

552 + args = parse_args(argv)

553 + try:

554 + report = run(args)

555 + except FileNotFoundError as exc:

556 + print(f"ERROR: {exc}", file=sys.stderr)

557 + return 1

558 +

559 + verdict = report.get("verdict", "unknown")

560 + rollback = report.get("rollback", False)

561 + print(f"Comparison complete: week={report.get('week')} verdict={verdict}", flush=True)

562 + if rollback:

563 + print(f"⚠️ ROLLBACK TRIGGERED: {report.get('rollback_reason')}", flush=True)

564 + if report.get("promotion_status"):

565 + status = report["promotion_status"]

566 + if status.get("eligible"):

567 + print("✅ Promotion eligible — awaiting operator opt-in.", flush=True)

568 + else:

569 + print(f"⏳ Not yet eligible: {status.get('reason')}", flush=True)

570 +

571 + return 1 if rollback or verdict != "pass" else 0

572 +

573 +

574 +def resolve_candidate_path(candidate_dir: Path, week: str) -> Path:

575 + preferred = candidate_dir / f"{week}-map-reduce-candidate.md"

576 + if preferred.exists():

577 + return preferred

578 + legacy = candidate_dir / "candidate-summary.md"

579 + if legacy.exists():

580 + return legacy

581 + return preferred

582 +

583 +

584 +def resolve_sidecar_path(

585 + candidate_dir: Path,

586 + *,

587 + preferred_name: str,

588 + legacy_name: str,

589 +) -> Path:

590 + preferred = candidate_dir / preferred_name

591 + if preferred.exists():

592 + return preferred

593 + legacy = candidate_dir / legacy_name

594 + if legacy.exists():

595 + return legacy

596 + return preferred

597 +

598 +

599 +def load_optional_json(

600 + path: Path,

601 + *,

602 + artifact_name: str,

603 + errors: list[str],

604 +) -> dict[str, Any]:

605 + if not path.exists():

606 + errors.append(f"{artifact_name} missing: {path}")

607 + return {}

608 + try:

609 + payload = json.loads(path.read_text(encoding="utf-8"))

610 + except (json.JSONDecodeError, OSError) as exc:

611 + errors.append(f"{artifact_name} unreadable: {path} ({exc})")

612 + return {}

613 + if not isinstance(payload, dict):

614 + errors.append(f"{artifact_name} must be a JSON object: {path}")

615 + return {}

616 + return payload

617 +

618 +

619 +def normalize_repo_url(url: str) -> str | None:

620 + prefix = "https://github.com/"

621 + if not url.startswith(prefix):

622 + return None

623 + repo_path = url.removeprefix(prefix).split("?", 1)[0].split("#", 1)[0]

624 + parts = [part for part in repo_path.split("/") if part]

625 + if len(parts) < 2:

626 + return None

627 + return f"{parts[0]}/{parts[1]}"

628 +

629 +

630 +def compute_orphaned_citations(

631 + *,

632 + text: str,

633 + editorial_plan: dict[str, Any],

634 +) -> int:

635 + if not editorial_plan:

636 + return len(MARKDOWN_LINK_PATTERN.findall(text))

637 +

638 + selected_claims = (

639 + editorial_plan.get("selected_claims", [])

640 + if isinstance(editorial_plan.get("selected_claims"), list)

641 + else []

642 + )

643 + key_references = (

644 + editorial_plan.get("key_references", {})

645 + if isinstance(editorial_plan.get("key_references"), dict)

646 + else {}

647 + )

648 +

649 + allowed_repos = {

650 + str(editorial_plan.get("top_repo", "")).strip(),

651 + *(str(repo).strip() for repo in key_references.get("notable_projects", [])),

652 + }

653 + allowed_articles = {

654 + str(url).strip() for url in key_references.get("press_articles", [])

655 + }

656 + for claim in selected_claims:

657 + if not isinstance(claim, dict):

658 + continue

659 + bindings = (

660 + claim.get("citation_bindings", {})

661 + if isinstance(claim.get("citation_bindings"), dict)

662 + else {}

663 + )

664 + allowed_repos.update(str(repo).strip() for repo in bindings.get("repos", []))

665 + allowed_articles.update(

666 + str(url).strip() for url in bindings.get("articles", [])

667 + )

668 +

669 + allowed_repos.discard("")

670 + allowed_articles.discard("")

671 +

672 + orphaned = 0

673 + for url in MARKDOWN_LINK_PATTERN.findall(text):

674 + repo_name = normalize_repo_url(url)

675 + if repo_name is not None:

676 + if repo_name not in allowed_repos:

677 + orphaned += 1

678 + continue

679 + if url not in allowed_articles:

680 + orphaned += 1

681 + return orphaned

682 +

683 +

684 +def count_unresolved_contradictions(contradictions_sidecar: dict[str, Any]) -> int:

685 + contradictions = (

686 + contradictions_sidecar.get("contradictions", [])

687 + if isinstance(contradictions_sidecar.get("contradictions"), list)

688 + else []

689 + )

690 + return len(contradictions)

691 +

692 +

693 +def count_invalid_rejected_claims(

694 + rejected_claims_sidecar: dict[str, Any],

695 + *,

696 + contradictions_sidecar: dict[str, Any],

697 +) -> int:

698 + rejected_claims = (

699 + rejected_claims_sidecar.get("rejected_claims", [])

700 + if isinstance(rejected_claims_sidecar.get("rejected_claims"), list)

701 + else []

702 + )

703 + contradiction_claim_ids = {

704 + str(entry.get("claim_id"))

705 + for entry in contradictions_sidecar.get("contradictions", [])

706 + if isinstance(entry, dict) and entry.get("claim_id") is not None

707 + }

708 + invalid = 0

709 + for entry in rejected_claims:

710 + if not isinstance(entry, dict):

711 + invalid += 1

712 + continue

713 + reason = str(entry.get("reason") or "").strip()

714 + if reason not in VALID_REJECTION_REASONS:

715 + invalid += 1

716 + continue

717 + if reason == "unresolved_contradiction":

718 + claim_id = entry.get("claim_id")

719 + if claim_id is None or str(claim_id) not in contradiction_claim_ids:

720 + invalid += 1

721 + return invalid

722 +

723 +

724 +def promotion_report_sort_key(report: dict[str, Any]) -> float:

725 + run_datetime = report.get("run_datetime")

726 + if isinstance(run_datetime, str):

727 + try:

728 + parsed = datetime.fromisoformat(run_datetime.replace("Z", "+00:00"))

729 + except ValueError:

730 + pass

731 + else:

732 + return parsed.astimezone(UTC).timestamp()

733 +

734 + week = report.get("week")

735 + if isinstance(week, str):

736 + match = WEEK_PATTERN.fullmatch(week)

737 + if match:

738 + report_week = datetime.fromisocalendar(

739 + int(match.group("year")),

740 + int(match.group("week")),

741 + 1,

742 + )

743 + return report_week.replace(tzinfo=UTC).timestamp()

744 +

745 + return float("-inf")

746 +

747 +

748 +if __name__ == "__main__":

749 + raise SystemExit(main())

tests/test_map_reduce_comparison.py new

+448

@@ -0,0 +1,448 @@

1 +"""Tests for the map/reduce comparison framework and promotion logic."""

2 +

3 +from __future__ import annotations

4 +

5 +import json

6 +from datetime import UTC, datetime

7 +from types import SimpleNamespace

8 +

9 +import pytest

10 +

11 +import scripts.map_reduce_comparison as comparison

12 +from scripts.map_reduce_comparison import (

13 + COMPARISON_SCHEMA,

14 + PROMOTION_HARD_FLOOR_COVERAGE,

15 + PROMOTION_HARD_FLOOR_QUALITY,

16 + PROMOTION_MIN_COVERAGE,

17 + PROMOTION_MIN_QUALITY,

18 + analyze_map_reduce,

19 + check_promotion_eligibility,

20 + compute_evidence_coverage_from_ledgers,

21 + compute_verdict,

22 + generate_comparison_report,

23 + main,

24 + should_rollback,

25 + ArtifactInfo,

26 +)

27 +

28 +

29 +def _make_artifact_info(

30 + *,

31 + quality_score: int = 70,

32 + gate_passed: bool = True,

33 + evidence_coverage: float = 0.90,

34 + citation_count: int = 12,

35 + word_count: int = 1800,

36 +) -> ArtifactInfo:

37 + return ArtifactInfo(

38 + path="test/artifact.md",

39 + sha256="abc123",

40 + quality_score=quality_score,

41 + gate_passed=gate_passed,

42 + evidence_coverage=evidence_coverage,

43 + citation_count=citation_count,

44 + word_count=word_count,

45 + )

46 +

47 +

48 +def _make_comparison_report(

49 + *,

50 + week: str = "2026-W24",

51 + verdict: str = "pass",

52 + quality_score: int = 70,

53 + evidence_coverage: float = 0.90,

54 + gate_regression: bool = False,

55 + run_datetime: str | None = None,

56 +) -> dict:

57 + if run_datetime is None:

58 + run_datetime = datetime.now(UTC).isoformat()

59 + return {

60 + "schema_version": COMPARISON_SCHEMA,

61 + "week": week,

62 + "run_datetime": run_datetime,

63 + "single_pass": {

64 + "artifact_path": "data/analyzed/summary.md",

65 + "sha256": "sp_hash",

66 + "quality_score": 72,

67 + "gate_passed": True,

68 + "evidence_coverage": 0.92,

69 + "citation_count": 14,

70 + "word_count": 1850,

71 + },

72 + "map_reduce": {

73 + "artifact_path": "data/map-reduce-candidates/candidate.md",

74 + "sha256": "mr_hash",

75 + "quality_score": quality_score,

76 + "gate_passed": not gate_regression,

77 + "evidence_coverage": evidence_coverage,

78 + "citation_count": 12,

79 + "word_count": 1720,

80 + "mapper_errors": {},

81 + "contradictions_resolved": 2,

82 + "claims_rejected": 5,

83 + },

84 + "deltas": {

85 + "quality_score": quality_score - 72,

86 + "evidence_coverage": evidence_coverage - 0.92,

87 + "citation_count": -2,

88 + "word_count": -130,

89 + "gate_regression": gate_regression,

90 + },

91 + "verdict": verdict,

92 + "blockers": [],

93 + }

94 +

95 +

96 +class TestComputeVerdict:

97 + """Tests for verdict computation logic."""

98 +

99 + def test_pass_when_all_criteria_met(self):

100 + sp = _make_artifact_info(quality_score=72, evidence_coverage=0.92)

101 + mr = _make_artifact_info(quality_score=68, evidence_coverage=0.88)

102 + deltas = {"gate_regression": False}

103 + verdict, blockers = compute_verdict(sp, mr, deltas)

104 + assert verdict == "pass"

105 + assert blockers == []

106 +

107 + def test_fail_on_gate_regression(self):

108 + sp = _make_artifact_info(gate_passed=True)

109 + mr = _make_artifact_info(gate_passed=False)

110 + deltas = {"gate_regression": True}

111 + verdict, blockers = compute_verdict(sp, mr, deltas)

112 + assert verdict == "fail"

113 + assert any("Gate regression" in b for b in blockers)

114 +

115 + def test_fail_on_low_quality_score(self):

116 + sp = _make_artifact_info()

117 + mr = _make_artifact_info(quality_score=55)

118 + deltas = {"gate_regression": False}

119 + verdict, blockers = compute_verdict(sp, mr, deltas)

120 + assert verdict == "fail"

121 + assert any("Quality score" in b for b in blockers)

122 +

123 + def test_fail_on_hard_quality_floor(self):

124 + sp = _make_artifact_info()

125 + mr = _make_artifact_info(quality_score=50)

126 + deltas = {"gate_regression": False}

127 + verdict, blockers = compute_verdict(sp, mr, deltas)

128 + assert verdict == "fail"

129 + assert any("hard floor" in b for b in blockers)

130 +

131 + def test_fail_on_low_coverage(self):

132 + sp = _make_artifact_info()

133 + mr = _make_artifact_info(evidence_coverage=0.80)

134 + deltas = {"gate_regression": False}

135 + verdict, blockers = compute_verdict(sp, mr, deltas)

136 + assert verdict == "fail"

137 + assert any("Evidence coverage" in b for b in blockers)

138 +

139 + def test_fail_on_hard_coverage_floor(self):

140 + sp = _make_artifact_info()

141 + mr = _make_artifact_info(evidence_coverage=0.60)

142 + deltas = {"gate_regression": False}

143 + verdict, blockers = compute_verdict(sp, mr, deltas)

144 + assert verdict == "fail"

145 + assert any("hard floor" in b for b in blockers)

146 +

147 + def test_fail_on_orphaned_citations(self):

148 + sp = _make_artifact_info()

149 + mr = _make_artifact_info()

150 + deltas = {"gate_regression": False, "orphaned_citations": 1}

151 + verdict, blockers = compute_verdict(sp, mr, deltas)

152 + assert verdict == "fail"

153 + assert any("Citation integrity failed" in b for b in blockers)

154 +

155 + def test_fail_on_unresolved_contradictions(self):

156 + sp = _make_artifact_info()

157 + mr = _make_artifact_info()

158 + deltas = {"gate_regression": False, "unresolved_contradictions": 1}

159 + verdict, blockers = compute_verdict(sp, mr, deltas)

160 + assert verdict == "fail"

161 + assert any("Contradiction handling failed" in b for b in blockers)

162 +

163 + def test_fail_on_invalid_rejected_claims(self):

164 + sp = _make_artifact_info()

165 + mr = _make_artifact_info()

166 + deltas = {"gate_regression": False, "invalid_rejected_claims": 1}

167 + verdict, blockers = compute_verdict(sp, mr, deltas)

168 + assert verdict == "fail"

169 + assert any("Claim rejection audit failed" in b for b in blockers)

170 +

171 +

172 +class TestShouldRollback:

173 + """Tests for automatic rollback trigger logic."""

174 +

175 + def test_no_rollback_on_passing_report(self):

176 + report = _make_comparison_report(verdict="pass")

177 + rollback, reason = should_rollback(report)

178 + assert rollback is False

179 + assert reason == ""

180 +

181 + def test_rollback_on_gate_regression(self):

182 + report = _make_comparison_report(gate_regression=True)

183 + report["deltas"]["gate_regression"] = True

184 + rollback, reason = should_rollback(report)

185 + assert rollback is True

186 + assert "Gate regression" in reason

187 +

188 + def test_rollback_on_hard_quality_floor(self):

189 + report = _make_comparison_report(quality_score=50)

190 + rollback, reason = should_rollback(report)

191 + assert rollback is True

192 + assert "hard floor" in reason

193 +

194 + def test_rollback_on_hard_coverage_floor(self):

195 + report = _make_comparison_report(evidence_coverage=0.60)

196 + rollback, reason = should_rollback(report)

197 + assert rollback is True

198 + assert "hard floor" in reason

199 +

200 + def test_rollback_on_mapper_failure(self):

201 + report = _make_comparison_report()

202 + report["map_reduce"]["mapper_errors"] = {

203 + "new_repos": ["schema_version mismatch"]

204 + }

205 + rollback, reason = should_rollback(report)

206 + assert rollback is True

207 + assert "Mapper failures" in reason

208 +

209 + def test_no_rollback_on_empty_mapper_errors(self):

210 + report = _make_comparison_report()

211 + report["map_reduce"]["mapper_errors"] = {"new_repos": [], "trending_repos": []}

212 + rollback, reason = should_rollback(report)

213 + assert rollback is False

214 +

215 +

216 +class TestCheckPromotionEligibility:

217 + """Tests for promotion eligibility across multiple runs."""

218 +

219 + def test_not_eligible_with_fewer_than_3_runs(self):

220 + reports = [_make_comparison_report(week=f"2026-W{i}") for i in range(2)]

221 + result = check_promotion_eligibility(reports)

222 + assert result["eligible"] is False

223 + assert "need ≥ 3" in result["reason"]

224 +

225 + def test_not_eligible_with_insufficient_passing_runs(self):

226 + reports = [

227 + _make_comparison_report(week="2026-W21", verdict="pass"),

228 + _make_comparison_report(week="2026-W22", verdict="fail"),

229 + _make_comparison_report(week="2026-W23", verdict="fail"),

230 + ]

231 + result = check_promotion_eligibility(reports)

232 + assert result["eligible"] is False

233 + assert "Only 1/" in result["reason"]

234 +

235 + def test_eligible_with_3_passing_runs(self):

236 + reports = [

237 + _make_comparison_report(week=f"2026-W{21+i}", quality_score=70)

238 + for i in range(3)

239 + ]

240 + result = check_promotion_eligibility(reports)

241 + assert result["eligible"] is True

242 + assert "operator opt-in" in result["reason"]

243 +

244 + def test_not_eligible_with_low_average_quality(self):

245 + reports = [

246 + _make_comparison_report(week=f"2026-W{21+i}", quality_score=62)

247 + for i in range(3)

248 + ]

249 + result = check_promotion_eligibility(reports)

250 + assert result["eligible"] is False

251 + assert "Average quality" in result["reason"]

252 +

253 + def test_not_eligible_with_stale_runs(self):

254 + old_dt = "2026-04-01T00:00:00+00:00"

255 + reports = [

256 + _make_comparison_report(

257 + week=f"2026-W{13+i}", run_datetime=old_dt, quality_score=70

258 + )

259 + for i in range(3)

260 + ]

261 + result = check_promotion_eligibility(reports)

262 + assert result["eligible"] is False

263 + assert "older than 28 days" in result["reason"]

264 +

265 + def test_not_eligible_when_most_recent_run_failed(self):

266 + reports = [

267 + _make_comparison_report(week="2026-W21", verdict="pass"),

268 + _make_comparison_report(week="2026-W22", verdict="pass"),

269 + _make_comparison_report(week="2026-W23", verdict="pass"),

270 + _make_comparison_report(week="2026-W24", verdict="fail"),

271 + ]

272 + result = check_promotion_eligibility(reports)

273 + assert result["eligible"] is False

274 + assert "most recent runs passed" in result["reason"]

275 +

276 +

277 +class TestGenerateComparisonReport:

278 + """Tests for report generation."""

279 +

280 + def test_schema_version_present(self):

281 + sp = _make_artifact_info()

282 + mr = _make_artifact_info()

283 + report = generate_comparison_report(

284 + week="2026-W24",

285 + single_pass=sp,

286 + map_reduce=mr,

287 + map_reduce_extra={},

288 + run_datetime="2026-06-14T07:00:00Z",

289 + )

290 + assert report["schema_version"] == COMPARISON_SCHEMA

291 + assert report["week"] == "2026-W24"

292 +

293 + def test_deltas_computed_correctly(self):

294 + sp = _make_artifact_info(quality_score=72, citation_count=14, word_count=1850)

295 + mr = _make_artifact_info(quality_score=68, citation_count=12, word_count=1720)

296 + report = generate_comparison_report(

297 + week="2026-W24",

298 + single_pass=sp,

299 + map_reduce=mr,

300 + map_reduce_extra={},

301 + run_datetime="2026-06-14T07:00:00Z",

302 + )

303 + assert report["deltas"]["quality_score"] == -4

304 + assert report["deltas"]["citation_count"] == -2

305 + assert report["deltas"]["word_count"] == -130

306 +

307 + def test_pass_verdict_when_criteria_met(self):

308 + sp = _make_artifact_info(quality_score=72, evidence_coverage=0.92)

309 + mr = _make_artifact_info(quality_score=68, evidence_coverage=0.88)

310 + report = generate_comparison_report(

311 + week="2026-W24",

312 + single_pass=sp,

313 + map_reduce=mr,

314 + map_reduce_extra={},

315 + run_datetime="2026-06-14T07:00:00Z",

316 + )

317 + assert report["verdict"] == "pass"

318 + assert report["blockers"] == []

319 +

320 +

321 +class TestOperatorControls:

322 + """Tests for environment variable controls."""

323 +

324 + def test_default_values(self):

325 + assert PROMOTION_MIN_QUALITY == 60

326 + assert PROMOTION_MIN_COVERAGE == 0.85

327 + assert PROMOTION_HARD_FLOOR_QUALITY == 55

328 + assert PROMOTION_HARD_FLOOR_COVERAGE == 0.70

329 +

330 +

331 +class TestArtifactLoading:

332 + def test_reads_mapper_ledgers_from_maps_dir(self, tmp_path):

333 + maps_dir = tmp_path / "maps"

334 + maps_dir.mkdir()

335 + (maps_dir / "new_repos.json").write_text(

336 + json.dumps(

337 + {

338 + "coverage": {

339 + "repo_count_input": 8,

340 + "repo_count_mapped": 6,

341 + "article_count_input": 2,

342 + "article_count_mapped": 1,

343 + }

344 + }

345 + ),

346 + encoding="utf-8",

347 + )

348 + (maps_dir / "trending_repos.json").write_text(

349 + json.dumps(

350 + {

351 + "coverage": {

352 + "repo_count_input": 2,

353 + "repo_count_mapped": 2,

354 + "article_count_input": 0,

355 + "article_count_mapped": 0,

356 + }

357 + }

358 + ),

359 + encoding="utf-8",

360 + )

361 +

362 + assert compute_evidence_coverage_from_ledgers(tmp_path) == pytest.approx(0.75)

363 +

364 + def test_analyze_map_reduce_uses_dry_run_artifacts_and_handles_bad_qa(

365 + self,

366 + tmp_path,

367 + monkeypatch,

368 + ):

369 + week = "2026-W24"

370 + (tmp_path / f"{week}-map-reduce-candidate.md").write_text(

371 + "---\nquality_score: 68\n---\n\n[octo/repo](https://github.com/octo/repo)\n",

372 + encoding="utf-8",

373 + )

374 + (tmp_path / "qa-comparison-report.json").write_text("{bad json", encoding="utf-8")

375 + (tmp_path / "editorial-plan.json").write_text(

376 + json.dumps(

377 + {

378 + "top_repo": "octo/repo",

379 + "selected_claims": [

380 + {

381 + "citation_bindings": {

382 + "repos": ["octo/repo"],

383 + "articles": [],

384 + }

385 + }

386 + ],

387 + "key_references": {

388 + "notable_projects": ["octo/repo"],

389 + "press_articles": [],

390 + },

391 + }

392 + ),

393 + encoding="utf-8",

394 + )

395 + sidecars_dir = tmp_path / "sidecars"

396 + sidecars_dir.mkdir()

397 + (sidecars_dir / "contradictions.json").write_text(

398 + json.dumps({"contradictions": []}),

399 + encoding="utf-8",

400 + )

401 + (sidecars_dir / "rejected-claims.json").write_text(

402 + json.dumps({"rejected_claims": []}),

403 + encoding="utf-8",

404 + )

405 + maps_dir = tmp_path / "maps"

406 + maps_dir.mkdir()

407 + (maps_dir / "new_repos.json").write_text(

408 + json.dumps(

409 + {

410 + "coverage": {

411 + "repo_count_input": 1,

412 + "repo_count_mapped": 1,

413 + "article_count_input": 0,

414 + "article_count_mapped": 0,

415 + }

416 + }

417 + ),

418 + encoding="utf-8",

419 + )

420 +

421 + monkeypatch.setattr(comparison, "validate_analysis", lambda *_args, **_kwargs: ([], 220))

422 + monkeypatch.setattr(

423 + comparison,

424 + "validate_publish_quality",

425 + lambda *_args, **_kwargs: ([], {}),

426 + )

427 +

428 + info, extra = analyze_map_reduce(

429 + tmp_path,

430 + {"week": week},

431 + "2026-06-14T07:00:00+00:00",

432 + )

433 +

434 + assert info.path.endswith(f"{week}-map-reduce-candidate.md")

435 + assert info.evidence_coverage == pytest.approx(1.0)

436 + assert any("QA comparison report unreadable" in err for err in extra["artifact_errors"])

437 +

438 +

439 +class TestMain:

440 + def test_returns_nonzero_on_failed_verdict(self, monkeypatch):

441 + monkeypatch.setattr(comparison, "parse_args", lambda _argv=None: SimpleNamespace())

442 + monkeypatch.setattr(

443 + comparison,

444 + "run",

445 + lambda _args: {"week": "2026-W24", "verdict": "fail", "rollback": False},

446 + )

447 +

448 + assert main([]) == 1