test: add rerun promotion guard regressions
Adds promotion guard regressions and path containment fixes for safe reruns.
Juan Manuel Servera committed
Jun 6, 2026 at 08:54 UTC
cb10bd49796269271b7d9a9286a383806d2991ee
3 files changed
+477
.squad/agents/fry/history.md
+6
@@ -103,3 +103,9 @@
103
104
- Signal-type hierarchical map/reduce is the lowest-risk analysis decomposition MVP because it matches current raw payload boundaries, is fixture-testable, and can run sidecar-first without overwriting the existing Copilot -> GitHub Models -> no-AI publish path.
105
- Map/reduce must not become publishable until mapper/reducer schemas, citation bindings, rejected-claim/contradiction sidecars, final `analysis_gate.py` compliance, fallback preservation, and A/B quality metrics pass.
106
+
107
+## Issue #257 overwrite-protection test groundwork (2026-06-05T21:16:49Z)
108
+
109
+- Added deterministic no-network regression coverage around publish eligibility manifests and promotion guard behavior.
110
+- Good canonical weekly summary/content must remain unchanged when candidates are failed, degraded, no-AI, stale, missing-manifest, or malformed-manifest.
111
+- Safe rerun promotion is copy-stable and must not append or duplicate article body content; ineligible candidates should remain in staging with promotion diagnostics for debugging.
scripts/promotion_guard.py
new
+204
@@ -0,0 +1,204 @@
1
+from __future__ import annotations
2
+
3
+import argparse
4
+import json
5
+import re
6
+import shutil
7
+from datetime import UTC, date, datetime
8
+from pathlib import Path
9
+from typing import Any
10
+
11
+
12
+class PromotionBlocked(ValueError):
13
+ def __init__(self, reasons: list[str]):
14
+ self.reasons = reasons
15
+ super().__init__("; ".join(reasons))
16
+
17
+
18
+WEEK_PATTERN = re.compile(r"^(?P<year>\d{4})-W(?P<week>\d{2})$")
19
+
20
+
21
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
22
+ parser = argparse.ArgumentParser(description="Promote an eligible weekly analysis candidate.")
23
+ parser.add_argument("--manifest", required=True, type=Path, help="Publish eligibility manifest path.")
24
+ parser.add_argument("--root", default=".", type=Path, help="Repository/workspace root.")
25
+ return parser.parse_args(argv)
26
+
27
+
28
+def _load_manifest(path: Path) -> dict[str, Any]:
29
+ try:
30
+ manifest = json.loads(path.read_text(encoding="utf-8"))
31
+ except FileNotFoundError as exc:
32
+ raise PromotionBlocked([f"Missing publish eligibility manifest: {path}"]) from exc
33
+ except json.JSONDecodeError as exc:
34
+ raise PromotionBlocked([f"Malformed publish eligibility manifest: {exc.msg}"]) from exc
35
+
36
+ if not isinstance(manifest, dict):
37
+ raise PromotionBlocked(["Publish eligibility manifest must be a JSON object."])
38
+ return manifest
39
+
40
+
41
+def _parse_date(value: Any) -> date | None:
42
+ if not isinstance(value, str) or not value.strip():
43
+ return None
44
+ candidate = value.strip()
45
+ if candidate.endswith("Z"):
46
+ candidate = f"{candidate[:-1]}+00:00"
47
+ try:
48
+ return datetime.fromisoformat(candidate).astimezone(UTC).date()
49
+ except ValueError:
50
+ try:
51
+ return date.fromisoformat(value.strip())
52
+ except ValueError:
53
+ return None
54
+
55
+
56
+def _resolve_under_root(root: Path, value: Any, field: str, reasons: list[str]) -> Path | None:
57
+ if not isinstance(value, str) or not value.strip():
58
+ reasons.append(f"{field} is required.")
59
+ return None
60
+ path = Path(value)
61
+ if path.is_absolute():
62
+ reasons.append(f"{field} must be relative to the repository root.")
63
+ return None
64
+ resolved_root = root.resolve()
65
+ resolved_path = (resolved_root / path).resolve()
66
+ try:
67
+ resolved_path.relative_to(resolved_root)
68
+ except ValueError:
69
+ reasons.append(f"{field} must stay under the repository root.")
70
+ return None
71
+ return resolved_path
72
+
73
+
74
+def _validate_manifest(manifest: dict[str, Any], root: Path, manifest_path: Path) -> tuple[str, Path, Path, list[str]]:
75
+ reasons: list[str] = []
76
+
77
+ if manifest.get("schema_version") != "publish_eligibility_v1":
78
+ reasons.append("schema_version must be publish_eligibility_v1.")
79
+
80
+ week = manifest.get("week")
81
+ if not isinstance(week, str) or not week.strip():
82
+ reasons.append("week is required.")
83
+ week = "unknown-week"
84
+ elif not WEEK_PATTERN.fullmatch(week):
85
+ reasons.append("week must use YYYY-WNN format.")
86
+
87
+ if manifest.get("promotion_eligible") is not True:
88
+ reasons.append("promotion_eligible must be true.")
89
+
90
+ candidate_summary = _resolve_under_root(root, manifest.get("candidate_summary_path"), "candidate_summary_path", reasons)
91
+ candidate_content = _resolve_under_root(root, manifest.get("candidate_content_path"), "candidate_content_path", reasons)
92
+
93
+ ai_provenance = manifest.get("ai_provenance")
94
+ if not isinstance(ai_provenance, dict):
95
+ reasons.append("ai_provenance is required.")
96
+ else:
97
+ source = ai_provenance.get("source")
98
+ if source in {None, "", "no-ai"}:
99
+ reasons.append("AI-authored provenance is required for normal promotion.")
100
+ if ai_provenance.get("degraded") is True:
101
+ reasons.append("degraded AI provenance is not eligible for normal promotion.")
102
+
103
+ gate_results = manifest.get("gate_results")
104
+ if not isinstance(gate_results, dict):
105
+ reasons.append("gate_results is required.")
106
+ else:
107
+ for gate in ("analysis_gate", "editorial_quality_gate", "evidence_freshness_gate"):
108
+ if gate_results.get(gate) is not True:
109
+ reasons.append(f"{gate} must pass.")
110
+
111
+ run_date = _parse_date(manifest.get("run_started_at"))
112
+ if run_date is None:
113
+ reasons.append("run_started_at must be an ISO date or timestamp.")
114
+
115
+ source_artifacts = manifest.get("source_artifacts")
116
+ if not isinstance(source_artifacts, list) or not source_artifacts:
117
+ reasons.append("source_artifacts must include at least one artifact.")
118
+ else:
119
+ for index, artifact in enumerate(source_artifacts, start=1):
120
+ prefix = f"source_artifacts[{index}]"
121
+ if not isinstance(artifact, dict):
122
+ reasons.append(f"{prefix} must be an object.")
123
+ continue
124
+ if artifact.get("stale") is True:
125
+ reasons.append(f"{prefix} is stale.")
126
+ generated_date = _parse_date(artifact.get("generated_at"))
127
+ if generated_date is None:
128
+ reasons.append(f"{prefix}.generated_at must be an ISO date or timestamp.")
129
+ elif run_date is not None and generated_date != run_date and artifact.get("reused_same_day") is not True:
130
+ reasons.append(f"{prefix} is not from the current run date or marked as same-day reuse.")
131
+ artifact_path = _resolve_under_root(root, artifact.get("path"), f"{prefix}.path", reasons)
132
+ if artifact_path is not None and not artifact_path.exists():
133
+ reasons.append(f"{prefix}.path does not exist: {artifact.get('path')}")
134
+
135
+ if candidate_summary is not None and not candidate_summary.exists():
136
+ reasons.append(f"candidate_summary_path does not exist: {manifest.get('candidate_summary_path')}")
137
+ if candidate_content is not None and not candidate_content.exists():
138
+ reasons.append(f"candidate_content_path does not exist: {manifest.get('candidate_content_path')}")
139
+
140
+ try:
141
+ manifest_relative = manifest_path.resolve().relative_to(root.resolve())
142
+ except ValueError:
143
+ reasons.append("Publish manifest must be under the repository root.")
144
+ manifest_relative = Path()
145
+ if manifest_relative.parts[:2] != ("data", "staging"):
146
+ reasons.append("Publish manifest must live under data/staging/.")
147
+
148
+ return str(week), candidate_summary or root, candidate_content or root, reasons
149
+
150
+
151
+def _write_diagnostic(root: Path, week: str, manifest: dict[str, Any] | None, reasons: list[str]) -> Path:
152
+ diagnostic_dir = root / "data" / "diagnostics" / "promotion"
153
+ diagnostic_dir.mkdir(parents=True, exist_ok=True)
154
+ diagnostic_path = diagnostic_dir / f"{week}-blocked.json"
155
+ diagnostic_path.write_text(
156
+ json.dumps({"week": week, "promotion": "blocked", "reasons": reasons, "manifest": manifest}, indent=2, sort_keys=True)
157
+ + "\n",
158
+ encoding="utf-8",
159
+ )
160
+ return diagnostic_path
161
+
162
+
163
+def promote_candidate(manifest_path: Path, *, root: Path | None = None) -> tuple[Path, Path]:
164
+ workspace = (root or Path.cwd()).resolve()
165
+ resolved_manifest_path = manifest_path if manifest_path.is_absolute() else workspace / manifest_path
166
+ manifest: dict[str, Any] | None = None
167
+ week = "unknown-week"
168
+
169
+ try:
170
+ manifest = _load_manifest(resolved_manifest_path)
171
+ week, candidate_summary, candidate_content, reasons = _validate_manifest(manifest, workspace, resolved_manifest_path)
172
+ if reasons:
173
+ raise PromotionBlocked(reasons)
174
+ except PromotionBlocked as exc:
175
+ _write_diagnostic(workspace, week, manifest, exc.reasons)
176
+ raise
177
+
178
+ canonical_summary = workspace / "data" / "analyzed" / f"{week}-summary.md"
179
+ match = WEEK_PATTERN.fullmatch(week)
180
+ if match is None:
181
+ raise PromotionBlocked(["week must use YYYY-WNN format."])
182
+ year, week_number = match.group("year"), match.group("week")
183
+ canonical_content = workspace / "content" / "weekly" / year / f"W{week_number}.md"
184
+ canonical_summary.parent.mkdir(parents=True, exist_ok=True)
185
+ canonical_content.parent.mkdir(parents=True, exist_ok=True)
186
+ shutil.copyfile(candidate_summary, canonical_summary)
187
+ shutil.copyfile(candidate_content, canonical_content)
188
+ return canonical_summary, canonical_content
189
+
190
+
191
+def main(argv: list[str] | None = None) -> int:
192
+ args = parse_args(argv)
193
+ try:
194
+ summary_path, content_path = promote_candidate(args.manifest, root=args.root)
195
+ except PromotionBlocked as exc:
196
+ for reason in exc.reasons:
197
+ print(f"::error::{reason}")
198
+ return 1
199
+ print(f"Promoted {summary_path} and {content_path}")
200
+ return 0
201
+
202
+
203
+if __name__ == "__main__":
204
+ raise SystemExit(main())
tests/test_promotion_guard.py
new
+267
@@ -0,0 +1,267 @@
1
+import json
2
+import tempfile
3
+import unittest
4
+from pathlib import Path
5
+
6
+from scripts import promotion_guard
7
+
8
+
9
+WEEK = "2026-W23"
10
+RUN_STARTED_AT = "2026-06-05T21:16:49Z"
11
+
12
+
13
+GOOD_SUMMARY = """---
14
+title: "Good AI Article"
15
+date: 2026-06-05T21:16:49Z
16
+week: "2026-W23"
17
+year: 2026
18
+tags: [ai]
19
+categories: [weekly]
20
+repos_featured: 1
21
+stars_tracked: 100
22
+top_repo: "owner/good"
23
+quality_score: 90
24
+summary: "A good AI-authored weekly summary."
25
+---
26
+
27
+## This Week's Trends
28
+
29
+Canonical good analysis.
30
+"""
31
+
32
+GOOD_CONTENT = """---
33
+title: "Good AI Article"
34
+week: "2026-W23"
35
+draft: false
36
+---
37
+
38
+Canonical good rendered content.
39
+"""
40
+
41
+VALID_REPLACEMENT_SUMMARY = GOOD_SUMMARY.replace("Good AI Article", "Better AI Article").replace(
42
+ "Canonical good analysis.", "Better candidate analysis."
43
+)
44
+VALID_REPLACEMENT_CONTENT = GOOD_CONTENT.replace("Good AI Article", "Better AI Article").replace(
45
+ "Canonical good rendered content.", "Better candidate rendered content."
46
+)
47
+
48
+
49
+def write_file(root: Path, relative_path: str, content: str) -> Path:
50
+ path = root / relative_path
51
+ path.parent.mkdir(parents=True, exist_ok=True)
52
+ path.write_text(content, encoding="utf-8")
53
+ return path
54
+
55
+
56
+def install_existing_good_article(root: Path) -> tuple[Path, Path]:
57
+ summary = write_file(root, "data/analyzed/2026-W23-summary.md", GOOD_SUMMARY)
58
+ content = write_file(root, "content/weekly/2026/W23.md", GOOD_CONTENT)
59
+ return summary, content
60
+
61
+
62
+def write_candidate(root: Path, name: str, summary: str, content: str) -> tuple[Path, Path]:
63
+ summary_path = write_file(root, f"data/staging/{WEEK}/{name}/summary.md", summary)
64
+ content_path = write_file(root, f"data/staging/{WEEK}/{name}/content.md", content)
65
+ return summary_path, content_path
66
+
67
+
68
+def write_source_artifact(root: Path, name: str = "raw") -> Path:
69
+ return write_file(root, f"data/raw/{WEEK}-{name}.json", json.dumps({"week": WEEK}) + "\n")
70
+
71
+
72
+def manifest_for(root: Path, name: str, **overrides) -> Path:
73
+ summary_path, content_path = write_candidate(
74
+ root,
75
+ name,
76
+ overrides.pop("summary", VALID_REPLACEMENT_SUMMARY),
77
+ overrides.pop("content", VALID_REPLACEMENT_CONTENT),
78
+ )
79
+ source_artifact = write_source_artifact(root, name)
80
+ manifest = {
81
+ "schema_version": "publish_eligibility_v1",
82
+ "week": WEEK,
83
+ "run_id": f"{WEEK}-{name}",
84
+ "run_started_at": RUN_STARTED_AT,
85
+ "candidate_summary_path": summary_path.relative_to(root).as_posix(),
86
+ "candidate_content_path": content_path.relative_to(root).as_posix(),
87
+ "promotion_eligible": True,
88
+ "ai_provenance": {
89
+ "source": "copilot-cli",
90
+ "model": "copilot-default",
91
+ "degraded": False,
92
+ },
93
+ "gate_results": {
94
+ "analysis_gate": True,
95
+ "editorial_quality_gate": True,
96
+ "evidence_freshness_gate": True,
97
+ },
98
+ "source_artifacts": [
99
+ {
100
+ "path": source_artifact.relative_to(root).as_posix(),
101
+ "checksum": "sha256:test",
102
+ "generated_at": RUN_STARTED_AT,
103
+ "reused_same_day": False,
104
+ "stale": False,
105
+ }
106
+ ],
107
+ }
108
+ for key, value in overrides.items():
109
+ manifest[key] = value
110
+ manifest_path = root / "data" / "staging" / WEEK / name / "publish-manifest.json"
111
+ manifest_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
112
+ return manifest_path
113
+
114
+
115
+class PromotionGuardTests(unittest.TestCase):
116
+ def test_failed_degraded_and_no_ai_candidates_do_not_replace_existing_good_article(self) -> None:
117
+ tests_root = Path(__file__).resolve().parent
118
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
119
+ root = Path(tmpdir)
120
+ canonical_summary, canonical_content = install_existing_good_article(root)
121
+ original_summary = canonical_summary.read_text(encoding="utf-8")
122
+ original_content = canonical_content.read_text(encoding="utf-8")
123
+
124
+ blocked_manifests = [
125
+ manifest_for(root, "failed", promotion_eligible=False),
126
+ manifest_for(root, "degraded", ai_provenance={"source": "copilot-cli", "model": "copilot-default", "degraded": True}),
127
+ manifest_for(root, "no-ai", ai_provenance={"source": "no-ai", "model": "none", "degraded": False}),
128
+ ]
129
+
130
+ for manifest_path in blocked_manifests:
131
+ with self.assertRaises(promotion_guard.PromotionBlocked):
132
+ promotion_guard.promote_candidate(manifest_path, root=root)
133
+ self.assertEqual(canonical_summary.read_text(encoding="utf-8"), original_summary)
134
+ self.assertEqual(canonical_content.read_text(encoding="utf-8"), original_content)
135
+
136
+ self.assertTrue((root / "data/staging/2026-W23/no-ai/summary.md").exists())
137
+ diagnostics = list((root / "data/diagnostics/promotion").glob("*-blocked.json"))
138
+ self.assertTrue(diagnostics)
139
+
140
+ def test_missing_malformed_and_stale_manifests_block_promotion(self) -> None:
141
+ tests_root = Path(__file__).resolve().parent
142
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
143
+ root = Path(tmpdir)
144
+ canonical_summary, _ = install_existing_good_article(root)
145
+ original_summary = canonical_summary.read_text(encoding="utf-8")
146
+
147
+ with self.assertRaises(promotion_guard.PromotionBlocked) as missing:
148
+ promotion_guard.promote_candidate(root / "data/staging/2026-W23/missing/publish-manifest.json", root=root)
149
+ self.assertIn("Missing publish eligibility manifest", missing.exception.reasons[0])
150
+
151
+ malformed = root / "data/staging/2026-W23/malformed/publish-manifest.json"
152
+ malformed.parent.mkdir(parents=True, exist_ok=True)
153
+ malformed.write_text("{not json", encoding="utf-8")
154
+ with self.assertRaises(promotion_guard.PromotionBlocked) as bad_json:
155
+ promotion_guard.promote_candidate(malformed, root=root)
156
+ self.assertIn("Malformed publish eligibility manifest", bad_json.exception.reasons[0])
157
+
158
+ stale_manifest = manifest_for(
159
+ root,
160
+ "stale",
161
+ source_artifacts=[
162
+ {
163
+ "path": write_source_artifact(root, "stale").relative_to(root).as_posix(),
164
+ "checksum": "sha256:stale",
165
+ "generated_at": "2026-06-04T21:16:49Z",
166
+ "reused_same_day": False,
167
+ "stale": True,
168
+ }
169
+ ],
170
+ )
171
+ with self.assertRaises(promotion_guard.PromotionBlocked) as stale:
172
+ promotion_guard.promote_candidate(stale_manifest, root=root)
173
+ self.assertIn("source_artifacts[1] is stale.", stale.exception.reasons)
174
+ self.assertEqual(canonical_summary.read_text(encoding="utf-8"), original_summary)
175
+
176
+ def test_same_successful_rerun_is_stable_and_does_not_duplicate_content(self) -> None:
177
+ tests_root = Path(__file__).resolve().parent
178
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
179
+ root = Path(tmpdir)
180
+ install_existing_good_article(root)
181
+ manifest_path = manifest_for(root, "valid")
182
+
183
+ first_summary, first_content = promotion_guard.promote_candidate(manifest_path, root=root)
184
+ first_summary_text = first_summary.read_text(encoding="utf-8")
185
+ first_content_text = first_content.read_text(encoding="utf-8")
186
+
187
+ second_summary, second_content = promotion_guard.promote_candidate(manifest_path, root=root)
188
+
189
+ self.assertEqual(second_summary.read_text(encoding="utf-8"), first_summary_text)
190
+ self.assertEqual(second_content.read_text(encoding="utf-8"), first_content_text)
191
+ self.assertEqual(second_summary.read_text(encoding="utf-8").count("Better candidate analysis."), 1)
192
+ self.assertEqual(second_content.read_text(encoding="utf-8").count("Better candidate rendered content."), 1)
193
+
194
+ def test_same_day_reused_source_candidate_can_promote_when_manifest_is_fresh(self) -> None:
195
+ tests_root = Path(__file__).resolve().parent
196
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
197
+ root = Path(tmpdir)
198
+ install_existing_good_article(root)
199
+ source_artifact = write_source_artifact(root, "same-day-reuse")
200
+ manifest_path = manifest_for(
201
+ root,
202
+ "same-day-reuse",
203
+ source_artifacts=[
204
+ {
205
+ "path": source_artifact.relative_to(root).as_posix(),
206
+ "checksum": "sha256:same-day",
207
+ "generated_at": "2026-06-05T08:00:00Z",
208
+ "reused_same_day": True,
209
+ "stale": False,
210
+ }
211
+ ],
212
+ )
213
+
214
+ summary_path, content_path = promotion_guard.promote_candidate(manifest_path, root=root)
215
+
216
+ self.assertIn("Better AI Article", summary_path.read_text(encoding="utf-8"))
217
+ self.assertIn("Better AI Article", content_path.read_text(encoding="utf-8"))
218
+
219
+ def test_candidate_paths_cannot_traverse_outside_repository_root(self) -> None:
220
+ tests_root = Path(__file__).resolve().parent
221
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
222
+ root = Path(tmpdir)
223
+ canonical_summary, canonical_content = install_existing_good_article(root)
224
+ original_summary = canonical_summary.read_text(encoding="utf-8")
225
+ original_content = canonical_content.read_text(encoding="utf-8")
226
+
227
+ outside_summary = root.parent / f"{root.name}-outside-summary.md"
228
+ outside_content = root.parent / f"{root.name}-outside-content.md"
229
+ try:
230
+ outside_summary.write_text(VALID_REPLACEMENT_SUMMARY, encoding="utf-8")
231
+ outside_content.write_text(VALID_REPLACEMENT_CONTENT, encoding="utf-8")
232
+ manifest_path = manifest_for(
233
+ root,
234
+ "traversal",
235
+ candidate_summary_path=f"../{outside_summary.name}",
236
+ candidate_content_path=f"../{outside_content.name}",
237
+ )
238
+
239
+ with self.assertRaises(promotion_guard.PromotionBlocked) as blocked:
240
+ promotion_guard.promote_candidate(manifest_path, root=root)
241
+
242
+ self.assertIn("candidate_summary_path must stay under the repository root.", blocked.exception.reasons)
243
+ self.assertIn("candidate_content_path must stay under the repository root.", blocked.exception.reasons)
244
+ self.assertEqual(canonical_summary.read_text(encoding="utf-8"), original_summary)
245
+ self.assertEqual(canonical_content.read_text(encoding="utf-8"), original_content)
246
+ finally:
247
+ outside_summary.unlink(missing_ok=True)
248
+ outside_content.unlink(missing_ok=True)
249
+
250
+ def test_manifest_path_must_be_directly_under_data_staging(self) -> None:
251
+ tests_root = Path(__file__).resolve().parent
252
+ with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
253
+ root = Path(tmpdir)
254
+ install_existing_good_article(root)
255
+ valid_manifest = manifest_for(root, "misplaced")
256
+ misplaced_manifest = root / "other" / "data" / "staging" / "publish-manifest.json"
257
+ misplaced_manifest.parent.mkdir(parents=True, exist_ok=True)
258
+ misplaced_manifest.write_text(valid_manifest.read_text(encoding="utf-8"), encoding="utf-8")
259
+
260
+ with self.assertRaises(promotion_guard.PromotionBlocked) as blocked:
261
+ promotion_guard.promote_candidate(misplaced_manifest, root=root)
262
+
263
+ self.assertIn("Publish manifest must live under data/staging/.", blocked.exception.reasons)
264
+
265
+
266
+if __name__ == "__main__":
267
+ unittest.main()