fix(scoring): objective quality score + audited force-replace override (W30 #583) (#584)

* Fix objective quality scoring and force replace Implement deterministic pipeline-owned quality scores and audited force-replace handling for jmservera/SquadScope#583. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Add regression tests for objective quality scoring and force-replace override (jmservera/SquadScope#583) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * test(pipeline): reflect pipeline-owned quality_score in analysis gate test (jmservera/SquadScope#583) quality_score is now overwritten by the deterministic objective scorer, so a hand-set low score no longer fails the gate. Assert the value is overwritten and use a genuine structural violation for the rejection case. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(scoring): guard missing frontmatter + exclude all GitHub hosts from press (jmservera/SquadScope#583) Address Copilot review on jmservera/SquadScope#584: - Wrap objective-score rewrite in try/except ValueError so a summary missing YAML frontmatter fails the gate cleanly (with the existing validation errors) instead of raising an uncaught exception and skipping the gate report. - Exclude all *.github.com and githubusercontent.com hosts (gist, raw, api, ...) from the external press citation set so non-press GitHub links never earn press points. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(scoring): isolate press subsection from trailing ### blocks (jmservera/SquadScope#583) Address Copilot review on jmservera/SquadScope#584: section_text() only stops at the next level-2 heading, so URLs in a ### subsection following '### Press & Industry' could inflate the press bonus. Trim the press section at the next level-3 heading. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Juan Manuel Servera committed Jul 20, 2026 at 15:06 UTC 357a6e8579c7bce0e4464054712026e9fa11c360
10 files changed +546 -15
.github/workflows/crawl-and-publish.yml
+12
@@ -44,6 +44,11 @@ on:
44 required: false
45 default: ''
46 type: string
47 + force_reason:
48 + description: 'Audited reason to force replacement while preserving all gates except candidate-vs-published score comparison.'
49 + required: false
50 + default: ''
51 + type: string
52 analysis_path:
53 description: 'Analysis path. map-reduce-dry-run is allowed only with dry-run/candidate-only and never promotes content.'
54 required: false
@@ -954,6 +959,8 @@ jobs:
959 PREFLIGHT_REPORT: ${{ steps.prompt-preflight.outputs.preflight_report_json }}
960 SYNTHESIS_STATUS: ${{ steps.synthesis.outputs.synthesis_status }}
961 SYNTHESIS_FILE: ${{ steps.synthesis.outputs.synthesis_file }}
962 + FORCE_REASON: ${{ inputs.force_reason || '' }}
963 + GH_ACTOR: ${{ github.actor }}
964 run: |
965 set -euo pipefail
966 git fetch origin publish 2>/dev/null && git checkout origin/publish -- "$PUBLISHED_SUMMARY" 2>/dev/null || true
@@ -978,6 +985,10 @@ jobs:
985 done
986 SYNTHESIS_ARGS=(--synthesis-status "${SYNTHESIS_STATUS:-missing}")
987 [ -n "${SYNTHESIS_FILE:-}" ] && [ -f "${SYNTHESIS_FILE:-}" ] && SYNTHESIS_ARGS+=(--synthesis-file "$SYNTHESIS_FILE")
988 + FORCE_ARGS=()
989 + if [ -n "$FORCE_REASON" ]; then
990 + FORCE_ARGS=(--publish-policy force-replace --force-reason "$FORCE_REASON" --actor "$GH_ACTOR")
991 + fi
992 python3 scripts/publish_manifest.py create \
993 --week "$WEEK" \
994 --run-id "$RUN_ID" \
@@ -995,6 +1006,7 @@ jobs:
1006 "${SYNTHESIS_ARGS[@]}" \
1007 --output "$MANIFEST_FILE" \
1008 "${RESTORE_ARGS[@]}" \
1009 + "${FORCE_ARGS[@]}" \
1010 "${ARTIFACT_ARGS[@]}"
1011
1012 - name: Assert candidate is eligible for promotion
prompts/analyze-topic.md
+1
@@ -169,6 +169,7 @@ Be critical, selective, and opinionated.
169 10. `stars_tracked` should equal the total stars across those repos.
170 11. `top_repo` should be the repo that best anchors the editorial narrative, not automatically the most-starred repo.
171 12. `quality_score` must be an honest 0-100 self-assessment; publishable work is `>= 60`.
172 + This value is advisory and is recomputed deterministically by the pipeline gate.
173 13. Include all required sections in this exact order:
174
175 ```md
prompts/analyze-weekly.md
+1
@@ -166,6 +166,7 @@ is the story of the period, not a list.
166 11. `stars_tracked` should equal the total stars across those repos.
167 12. `top_repo` should be the repo that best anchors the editorial narrative, not automatically the most-starred repo.
168 13. `quality_score` must be an honest 0-100 self-assessment; publishable work is `>= 60`, and a narrative-led, continuity-aware, evidence-dense weekly analysis that fully satisfies this contract should land at the top of the band (high 80s or above). Score honestly — do not inflate — but do not under-rate work that genuinely meets every requirement here. The `summary` field must be ≤155 characters, a complete sentence crafted as the meta description for search engines and social sharing. Do not let it exceed 155 characters.
169 + This value is advisory and is recomputed deterministically by the pipeline gate.
170 14. If you include `predictions`, each entry must be `{repo, claim_type, direction, confidence}` with `claim_type` in `signal|noise|gap`, `direction` in `up|flat|down`, and `confidence` from `0` to `1`.
171 15. Open with a narrative lede (2-4 paragraphs, ~120-220 words, no heading) as described in
172 "Lead with the story", then include all required sections in this exact order:
scripts/analysis_gate.py
+85
@@ -559,6 +559,73 @@ def raw_repo_names(raw_payload: dict[str, Any]) -> set[str]:
559 return {name for name in names if TOP_REPO_PATTERN.fullmatch(name)}
560
561
562 +def compute_objective_quality(
563 + text: str, raw_payload: dict, press_context_available: bool
564 +) -> tuple[int, dict]:
565 + _, body = extract_frontmatter(text)
566 + words = len(WORD_PATTERN.findall(body))
567 + depth = round(min(15, max(0, (words - 200) / 1000 * 15)))
568 +
569 + available_repos = raw_repo_names(raw_payload)
570 + cited_repos = set(REPO_LINK_PATTERN.findall(body)).intersection(available_repos)
571 + evidence_target = min(10, len(available_repos))
572 + evidence = (
573 + 0 if evidence_target == 0 else round(min(10, len(cited_repos) / evidence_target * 10))
574 + )
575 +
576 + press_citations = 0
577 + press = 0
578 + if press_context_available:
579 + key_references = section_text(body, "## Key References")
580 + press_section = section_text(key_references, "### Press & Industry")
581 + # section_text() only terminates on the next level-2 heading, so trim at the next
582 + # level-3 subsection to avoid counting URLs from later ### blocks as press citations.
583 + next_subsection = re.search(r"(?m)^###\s+", press_section)
584 + if next_subsection:
585 + press_section = press_section[: next_subsection.start()]
586 + urls = set(re.findall(r"https?://[^\s)\]]+", press_section))
587 + external_urls = {
588 + url
589 + for url in urls
590 + if not re.match(
591 + r"https?://(?:[^/\s]+\.)?(?:github\.com|githubusercontent\.com)(?:[/:?#]|$)",
592 + url,
593 + re.IGNORECASE,
594 + )
595 + }
596 + press_citations = len(external_urls)
597 + press = round(min(15, press_citations / 3 * 15))
598 +
599 + # Identical content with cited press must score strictly above its press-less variant.
600 + score = min(100, 60 + depth + evidence + press)
601 + return score, {
602 + "base": 60,
603 + "depth": depth,
604 + "evidence": evidence,
605 + "press": press,
606 + "words": words,
607 + "repo_citations": len(cited_repos),
608 + "press_citations": press_citations,
609 + "press_available": press_context_available,
610 + }
611 +
612 +
613 +def set_frontmatter_quality_score(text: str, score: int) -> str:
614 + match = FRONTMATTER_PATTERN.match(text)
615 + if not match:
616 + raise ValueError("Analysis output is missing YAML frontmatter.")
617 + frontmatter_text, body = match.groups()
618 + rewritten_frontmatter, replacements = re.subn(
619 + r"(?m)^quality_score:.*$",
620 + f"quality_score: {score}",
621 + frontmatter_text,
622 + count=1,
623 + )
624 + if replacements == 0:
625 + rewritten_frontmatter = f"{frontmatter_text}\nquality_score: {score}"
626 + return f"---\n{rewritten_frontmatter}\n---\n{body}"
627 +
628 +
629 def raw_artifact_week_errors(raw_payload: dict[str, Any], expected_week: Any) -> list[str]:
630 if not isinstance(expected_week, str):
631 return []
@@ -874,6 +941,7 @@ def write_gate_report(
941 repair_actions: list[str],
942 word_count: int,
943 gate_results: dict[str, dict[str, Any]],
944 + quality_breakdown: dict[str, Any] | None = None,
945 ) -> None:
946 if path is None:
947 return
@@ -884,6 +952,7 @@ def write_gate_report(
952 "model": model,
953 "passed": not errors_after,
954 "word_count": word_count,
955 + "quality_breakdown": quality_breakdown,
956 "gates": gate_results,
957 "failure_summary": build_failure_summary(errors_after, gate_results),
958 "errors_before_repair": errors_before,
@@ -984,6 +1053,21 @@ def main(argv: list[str] | None = None) -> int:
1053 f"::notice::Analysis gate applied safe repairs: {', '.join(repair_actions)}",
1054 file=sys.stderr,
1055 )
1056 + objective_score: int | None = None
1057 + quality_breakdown: dict | None = None
1058 + try:
1059 + objective_score, quality_breakdown = compute_objective_quality(
1060 + text, raw_payload, press_context_available
1061 + )
1062 + rewritten = set_frontmatter_quality_score(text, objective_score)
1063 + except ValueError:
1064 + # Missing/invalid frontmatter is already reported by validate_analysis(); let the
1065 + # gate fail cleanly with those errors instead of raising an uncaught exception.
1066 + rewritten = text
1067 + if rewritten != text:
1068 + args.analysis_file.write_text(rewritten, encoding="utf-8")
1069 + text = rewritten
1070 + errors, word_count = validate_analysis(text, raw_payload, args.current_datetime)
1071 publish_errors, _ = validate_publish_quality(
1072 text,
1073 raw_payload,
@@ -1003,6 +1087,7 @@ def main(argv: list[str] | None = None) -> int:
1087 repair_actions=repair_actions,
1088 word_count=word_count,
1089 gate_results=gate_results,
1090 + quality_breakdown=quality_breakdown,
1091 )
1092 if errors:
1093 fail(errors, summary_path)
scripts/promotion_guard.py
+8 -2
@@ -293,8 +293,14 @@ def _validate_manifest(
293 reasons.append("no-AI fallback provenance requires fallback_reason.")
294 if not ai_provenance.get("attempted_ai_paths"):
295 reasons.append("no-AI fallback provenance requires attempted_ai_paths.")
296 - elif source in {None, ""}:
297 - reasons.append("AI-authored provenance is required for normal promotion.")
296 + else:
297 + if policy_mode == "force-replace":
298 + if not policy.get("reason"):
299 + reasons.append("force-replace requires a reason.")
300 + if not policy.get("actor"):
301 + reasons.append("force-replace requires an actor.")
302 + if source in {None, ""}:
303 + reasons.append("AI-authored provenance is required for normal promotion.")
304 if ai_provenance.get("degraded") is True:
305 reasons.append("degraded AI provenance is not eligible for normal promotion.")
306
scripts/publish_manifest.py
+13 -12
@@ -112,7 +112,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
112 "--publish-policy",
113 choices=["default", "allow-no-ai-first-publish", "force-replace"],
114 default="default",
115 - help="Explicit operator policy for no-AI fallback publication.",
115 + help="Explicit operator policy for publication overrides.",
116 )
117 create.add_argument(
118 "--force-reason", default="", help="Operator reason required for force-replace."
@@ -120,7 +120,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
120 create.add_argument(
121 "--actor",
122 default="",
123 - help="Operator or automation actor requesting explicit fallback policy.",
123 + help="Operator or automation actor requesting an explicit publication policy.",
124 )
125
126 check = subparsers.add_parser(
@@ -709,7 +709,7 @@ def create_manifest(args: argparse.Namespace) -> int:
709 gates_passed = gate_report.get("present") is True and gate_report.get("passed") is True
710 candidate_quality = candidate_metadata.get("quality_score")
711 attempted_ai_paths = [path for path in args.attempted_ai_path if path.strip()]
712 - force_replacing_no_ai = ai_status == "no-ai" and args.publish_policy == "force-replace"
712 + force_replacing = args.publish_policy == "force-replace"
713 comparison_reasons: list[str] = []
714 if candidate_exists:
715 if candidate_metadata.get("week") not in {None, args.week}:
@@ -725,7 +725,7 @@ def create_manifest(args: argparse.Namespace) -> int:
725 if (
726 published_status.get("good")
727 and isinstance(candidate_quality, (int, float))
728 - and not force_replacing_no_ai
728 + and not force_replacing
729 ):
730 published_quality = published_status.get("quality_score")
731 if (
@@ -738,6 +738,11 @@ def create_manifest(args: argparse.Namespace) -> int:
738
739 reasons: list[str] = []
740 fallback_errors: list[str] = []
741 + if force_replacing:
742 + if not args.force_reason.strip():
743 + reasons.append("force-replace requires force_reason")
744 + if not args.actor.strip():
745 + reasons.append("force-replace requires actor")
746 if ai_status == "no-ai":
747 if not args.fallback_reason.strip():
748 reasons.append("fallback_reason is required for no-AI fallback candidates")
@@ -761,10 +766,6 @@ def create_manifest(args: argparse.Namespace) -> int:
766 )
767 fallback_errors = fallback_quality_errors(args.summary, validation_passed)
768 elif args.publish_policy == "force-replace":
764 - if not args.force_reason.strip():
765 - reasons.append("force-replace requires force_reason")
766 - if not args.actor.strip():
767 - reasons.append("force-replace requires actor")
769 fallback_errors = fallback_quality_errors(args.summary, validation_passed)
770 reasons.extend(fallback_errors)
771
@@ -1041,6 +1042,10 @@ def assert_eligible(args: argparse.Namespace) -> int:
1042 if isinstance(payload.get("promotion"), dict)
1043 else None
1044 )
1045 + if promotion_policy == "force-replace":
1046 + audit = payload.get("audit")
1047 + if not isinstance(audit, dict) or not audit.get("actor") or not audit.get("reason"):
1048 + raise SystemExit("Force replacement requires actor and reason in manifest audit.")
1049 if ai_status == "no-ai":
1050 provenance = analysis.get("provenance") if isinstance(analysis, dict) else {}
1051 if not isinstance(provenance, dict) or provenance.get("authorship") != "no-ai-fallback":
@@ -1049,10 +1054,6 @@ def assert_eligible(args: argparse.Namespace) -> int:
1054 raise SystemExit("Manifest lacks no-AI fallback reason.")
1055 if not provenance.get("attempted_ai_paths"):
1056 raise SystemExit("Manifest lacks attempted AI path audit.")
1052 - if promotion_policy == "force-replace":
1053 - audit = payload.get("audit")
1054 - if not isinstance(audit, dict) or not audit.get("actor") or not audit.get("reason"):
1055 - raise SystemExit("Force replacement requires actor and reason in manifest audit.")
1057 elif ai_status != "ai":
1058 raise SystemExit("Manifest lacks publishable AI provenance.")
1059 promotion = payload.get("promotion")
tests/test_analysis_gate.py
+201
@@ -111,6 +111,207 @@ summary: "A grounded week focused on practical tools."'''.strip()
111
112
113 class AnalysisGateTests(unittest.TestCase):
114 + def test_objective_quality_press_included_never_scores_below_press_less(self) -> None:
115 + body = make_body().replace(
116 + "No press data was provided this week.",
117 + "- [Industry report](https://example.com/industry-report) — confirms the trend.",
118 + )
119 + analysis = make_analysis(VALID_FRONTMATTER, body)
120 +
121 + press_score, press_breakdown = analysis_gate.compute_objective_quality(
122 + analysis, RAW_PAYLOAD_WITH_REPOS, press_context_available=True
123 + )
124 + press_less_score, press_less_breakdown = analysis_gate.compute_objective_quality(
125 + analysis, RAW_PAYLOAD_WITH_REPOS, press_context_available=False
126 + )
127 +
128 + # jmservera/SquadScope#583: prevent the W30 paradox where adding real press
129 + # evidence made an otherwise identical summary score lower.
130 + self.assertGreaterEqual(press_score, press_less_score)
131 + self.assertGreater(press_score, press_less_score)
132 + self.assertEqual(press_breakdown["words"], press_less_breakdown["words"])
133 + self.assertEqual(press_breakdown["repo_citations"], press_less_breakdown["repo_citations"])
134 + self.assertEqual(press_breakdown["press_citations"], 1)
135 + self.assertEqual(press_less_breakdown["press"], 0)
136 +
137 + def test_objective_quality_no_press_week_retains_publishable_base_score(self) -> None:
138 + score, breakdown = analysis_gate.compute_objective_quality(
139 + make_analysis("week: 2026-W23", "A short press-less summary."),
140 + RAW_PAYLOAD,
141 + press_context_available=False,
142 + )
143 +
144 + self.assertGreaterEqual(score, 60)
145 + self.assertEqual(breakdown["base"], 60)
146 + self.assertEqual(breakdown["press"], 0)
147 + self.assertFalse(breakdown["press_available"])
148 +
149 + def test_objective_quality_depth_and_evidence_scale_and_cap(self) -> None:
150 + short_text = make_analysis("week: 2026-W23", "word " * 200)
151 + medium_text = make_analysis("week: 2026-W23", "word " * 700)
152 + long_text = make_analysis("week: 2026-W23", "word " * 1400)
153 +
154 + _, short = analysis_gate.compute_objective_quality(short_text, {}, False)
155 + _, medium = analysis_gate.compute_objective_quality(medium_text, {}, False)
156 + _, long = analysis_gate.compute_objective_quality(long_text, {}, False)
157 +
158 + self.assertEqual(short["depth"], 0)
159 + self.assertGreater(medium["depth"], short["depth"])
160 + self.assertEqual(long["depth"], 15)
161 +
162 + repos = [f"owner/repo-{index}" for index in range(12)]
163 + raw_payload = {
164 + "new_repos": [{"full_name": repo} for repo in repos[:6]],
165 + "trending_repos": [{"full_name": repo} for repo in repos[6:]],
166 + }
167 +
168 + def cited_analysis(count: int) -> str:
169 + links = "\n".join(f"- [{repo}](https://github.com/{repo})" for repo in repos[:count])
170 + return make_analysis("week: 2026-W23", links)
171 +
172 + _, none = analysis_gate.compute_objective_quality(cited_analysis(0), raw_payload, False)
173 + _, half = analysis_gate.compute_objective_quality(cited_analysis(5), raw_payload, False)
174 + _, full = analysis_gate.compute_objective_quality(cited_analysis(12), raw_payload, False)
175 + _, no_inventory = analysis_gate.compute_objective_quality(cited_analysis(1), {}, False)
176 +
177 + self.assertEqual(none["evidence"], 0)
178 + self.assertEqual(half["evidence"], 5)
179 + self.assertEqual(full["evidence"], 10)
180 + self.assertEqual(full["repo_citations"], 12)
181 + self.assertEqual(no_inventory["evidence"], 0)
182 + self.assertEqual(no_inventory["repo_citations"], 0)
183 +
184 + def test_objective_quality_press_bonus_caps_and_requires_press_context(self) -> None:
185 + body = """## Key References
186 +
187 +### Notable Projects
188 +
189 +- [owner/repo-a](https://github.com/owner/repo-a)
190 +
191 +### Press & Industry
192 +
193 +- [One](https://one.example/article)
194 +- [Two](https://two.example/article)
195 +- [Three](https://three.example/article)
196 +- [Four](https://four.example/article)
197 +- [GitHub](https://github.com/owner/repo-a)
198 +"""
199 + analysis = make_analysis("week: 2026-W23", body)
200 +
201 + _, available = analysis_gate.compute_objective_quality(
202 + analysis, RAW_PAYLOAD_WITH_REPOS, True
203 + )
204 + _, unavailable = analysis_gate.compute_objective_quality(
205 + analysis, RAW_PAYLOAD_WITH_REPOS, False
206 + )
207 +
208 + self.assertEqual(available["press"], 15)
209 + self.assertEqual(available["press_citations"], 4)
210 + self.assertEqual(unavailable["press"], 0)
211 + self.assertEqual(unavailable["press_citations"], 0)
212 +
213 + def test_objective_quality_excludes_github_owned_hosts_from_press(self) -> None:
214 + body = """## Key References
215 +
216 +### Notable Projects
217 +
218 +- [owner/repo-a](https://github.com/owner/repo-a)
219 +
220 +### Press & Industry
221 +
222 +- [Real press](https://press.example/article)
223 +- [Gist](https://gist.github.com/owner/abc123)
224 +- [Raw](https://raw.githubusercontent.com/owner/repo-a/main/README.md)
225 +- [Sub](https://api.github.com/repos/owner/repo-a)
226 +"""
227 + analysis = make_analysis("week: 2026-W23", body)
228 +
229 + _, breakdown = analysis_gate.compute_objective_quality(
230 + analysis, RAW_PAYLOAD_WITH_REPOS, True
231 + )
232 +
233 + self.assertEqual(breakdown["press_citations"], 1)
234 +
235 + def test_objective_quality_ignores_urls_after_press_subsection(self) -> None:
236 + body = """## Key References
237 +
238 +### Press & Industry
239 +
240 +- [Real press](https://press.example/article)
241 +
242 +### Further Reading
243 +
244 +- [Not press](https://blog.example/post)
245 +- [Also not](https://docs.example/guide)
246 +"""
247 + analysis = make_analysis("week: 2026-W23", body)
248 +
249 + _, breakdown = analysis_gate.compute_objective_quality(
250 + analysis, RAW_PAYLOAD_WITH_REPOS, True
251 + )
252 +
253 + self.assertEqual(breakdown["press_citations"], 1)
254 +
255 + def test_set_frontmatter_quality_score_replaces_inserts_and_preserves_body(self) -> None:
256 + body = "Body with quality_score: 999 that must remain untouched.\n"
257 + existing = make_analysis("week: 2026-W23\nquality_score: 99", body)
258 + missing = make_analysis("week: 2026-W23", body)
259 +
260 + replaced = analysis_gate.set_frontmatter_quality_score(existing, 72)
261 + inserted = analysis_gate.set_frontmatter_quality_score(missing, 68)
262 +
263 + self.assertIn("\nquality_score: 72\n---", replaced)
264 + self.assertEqual(replaced.split("---\n", 2)[2], f"\n{body}\n")
265 + self.assertIn("\nquality_score: 68\n---", inserted)
266 + self.assertEqual(inserted.split("---\n", 2)[2], f"\n{body}\n")
267 + with self.assertRaisesRegex(ValueError, "missing YAML frontmatter"):
268 + analysis_gate.set_frontmatter_quality_score(body, 70)
269 +
270 + def test_main_overwrites_llm_quality_score_and_reports_objective_breakdown(self) -> None:
271 + tests_root = Path(__file__).resolve().parent
272 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
273 + workspace = Path(tmpdir)
274 + analysis_path = workspace / "candidate.md"
275 + raw_path = workspace / "raw.json"
276 + report_path = workspace / "report.json"
277 + original = make_analysis(
278 + VALID_FRONTMATTER.replace("quality_score: 82", "quality_score: 99"),
279 + make_body(),
280 + )
281 + expected_score, expected_breakdown = analysis_gate.compute_objective_quality(
282 + original, RAW_PAYLOAD_WITH_REPOS, False
283 + )
284 + analysis_path.write_text(original, encoding="utf-8")
285 + raw_path.write_text(json.dumps(RAW_PAYLOAD_WITH_REPOS), encoding="utf-8")
286 +
287 + self.assertEqual(
288 + analysis_gate.main(
289 + [
290 + "--analysis-file",
291 + str(analysis_path),
292 + "--raw-json",
293 + str(raw_path),
294 + "--current-datetime",
295 + CURRENT_DATETIME,
296 + "--source",
297 + "copilot-cli",
298 + "--model",
299 + "copilot-default",
300 + "--report-json",
301 + str(report_path),
302 + ]
303 + ),
304 + 0,
305 + )
306 +
307 + frontmatter, _ = analysis_gate.extract_frontmatter(
308 + analysis_path.read_text(encoding="utf-8")
309 + )
310 + report = json.loads(report_path.read_text(encoding="utf-8"))
311 + self.assertEqual(frontmatter["quality_score"], expected_score)
312 + self.assertNotEqual(frontmatter["quality_score"], 99)
313 + self.assertEqual(report["quality_breakdown"], expected_breakdown)
314 +
315 def test_validate_analysis_accepts_block_style_lists(self) -> None:
316 errors, word_count = analysis_gate.validate_analysis(
317 make_analysis(VALID_FRONTMATTER, make_body()),
tests/test_pipeline.py
+29 -1
@@ -847,9 +847,37 @@ class PipelineIntegrationTests(unittest.TestCase):
847 0,
848 )
849
850 + # quality_score is now pipeline-owned: a hand-set value is overwritten by the
851 + # deterministic objective score, so an otherwise-valid summary still passes even
852 + # when the authored score is low (jmservera/SquadScope#583).
853 + overwritten_path = base / "data" / "analyzed" / "overwritten-summary.md"
854 + overwritten_path.write_text(
855 + make_analysis_markdown().replace("quality_score: 86", "quality_score: 40"),
856 + encoding="utf-8",
857 + )
858 + self.assertEqual(
859 + analysis_gate.main(
860 + [
861 + "--analysis-file",
862 + str(overwritten_path),
863 + "--raw-json",
864 + str(raw_path),
865 + "--current-datetime",
866 + FIXED_RUN_DATETIME,
867 + "--source",
868 + "copilot-cli",
869 + ]
870 + ),
871 + 0,
872 + )
873 + rewritten = overwritten_path.read_text(encoding="utf-8")
874 + self.assertNotIn("quality_score: 40", rewritten)
875 + self.assertRegex(rewritten, r"(?m)^quality_score: (?:6[0-9]|[7-9][0-9]|100)$")
876 +
877 + # A genuine gate violation (missing required section heading) must still be rejected.
878 invalid_path = base / "data" / "analyzed" / "invalid-summary.md"
879 invalid_path.write_text(
852 - make_analysis_markdown().replace("quality_score: 86", "quality_score: 40"),
880 + make_analysis_markdown().replace("## The Week Ahead", "## Looking Forward"),
881 encoding="utf-8",
882 )
883
tests/test_promotion_guard.py
+44
@@ -517,6 +517,50 @@ class PromotionGuardTests(unittest.TestCase):
517 audit = json.loads(audit_path.read_text(encoding="utf-8"))
518 self.assertEqual(audit["actor"], "jmservera")
519
520 + def test_force_replace_ai_candidate_requires_reason_and_actor(self) -> None:
521 + tests_root = Path(__file__).resolve().parent
522 + for name, policy, expected_reason in (
523 + (
524 + "ai-force-missing-reason",
525 + {"mode": "force-replace", "actor": "jmservera"},
526 + "force-replace requires a reason.",
527 + ),
528 + (
529 + "ai-force-missing-actor",
530 + {"mode": "force-replace", "reason": "operator approved correction"},
531 + "force-replace requires an actor.",
532 + ),
533 + ):
534 + with self.subTest(name=name):
535 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
536 + root = Path(tmpdir)
537 + manifest_path = manifest_for(root, name, promotion_policy=policy)
538 +
539 + with self.assertRaises(promotion_guard.PromotionBlocked) as blocked:
540 + promotion_guard.promote_candidate(manifest_path, root=root)
541 +
542 + self.assertIn(expected_reason, blocked.exception.reasons)
543 +
544 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
545 + root = Path(tmpdir)
546 + manifest_path = manifest_for(
547 + root,
548 + "ai-force-complete",
549 + promotion_policy={
550 + "mode": "force-replace",
551 + "reason": "operator approved correction",
552 + "actor": "jmservera",
553 + },
554 + )
555 +
556 + summary_path, _ = promotion_guard.promote_candidate(manifest_path, root=root)
557 +
558 + self.assertIn("Better candidate analysis.", summary_path.read_text(encoding="utf-8"))
559 + audit_path = root / "data/diagnostics/promotion/2026-W23-force-replace-audit.json"
560 + audit = json.loads(audit_path.read_text(encoding="utf-8"))
561 + self.assertEqual(audit["reason"], "operator approved correction")
562 + self.assertEqual(audit["actor"], "jmservera")
563 +
564 def test_same_day_reused_source_candidate_can_promote_when_manifest_is_fresh(self) -> None:
565 tests_root = Path(__file__).resolve().parent
566 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
tests/test_publish_manifest.py
+152
@@ -742,6 +742,155 @@ class PublishManifestTests(unittest.TestCase):
742 self.assertEqual(payload["audit"]["actor"], "jmservera")
743 self.assertEqual(assert_eligible_from_root(base, manifest), 0)
744
745 + def test_force_replace_allows_lower_quality_ai_candidate_over_good_baseline(self) -> None:
746 + tests_root = Path(__file__).resolve().parent
747 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
748 + base = Path(tmpdir)
749 + raw = base / "data/raw/2026-W21.json"
750 + summary = base / "data/candidates/2026-W21/123456/2026-W21-summary.md"
751 + published = base / "data/analyzed/2026-W21-summary.md"
752 + manifest = base / "data/candidates/2026-W21/123456/publish-manifest.json"
753 + gate_report = base / "data/candidates/2026-W21/123456/analysis-gate-report.json"
754 + write_raw(raw)
755 + write_good_summary(summary, quality_score=70)
756 + write_good_summary(published, quality_score=90)
757 + write_gate_report(gate_report)
758 + args = create_args(base, raw, summary, manifest, gate_report=gate_report)
759 + args.extend(
760 + [
761 + "--publish-policy",
762 + "force-replace",
763 + "--force-reason",
764 + "operator approved W30 correction",
765 + "--actor",
766 + "jmservera",
767 + ]
768 + )
769 +
770 + # jmservera/SquadScope#583: the W30 correction must bypass only the
771 + # candidate-vs-published score comparison when explicitly audited.
772 + self.assertEqual(publish_manifest.main(args), 0)
773 +
774 + payload = json.loads(manifest.read_text(encoding="utf-8"))
775 + self.assertTrue(payload["promotion"]["eligible"])
776 + self.assertEqual(payload["promotion"]["decision"], "promote")
777 + self.assertFalse(
778 + any(
779 + "lower than published good quality_score" in reason
780 + for reason in payload["promotion"]["reasons"]
781 + )
782 + )
783 + self.assertEqual(assert_eligible_from_root(base, manifest), 0)
784 +
785 + def test_force_replace_ai_candidate_requires_reason_and_actor(self) -> None:
786 + tests_root = Path(__file__).resolve().parent
787 + for missing_flag, expected_reason in (
788 + ("--force-reason", "force-replace requires force_reason"),
789 + ("--actor", "force-replace requires actor"),
790 + ):
791 + with self.subTest(missing_flag=missing_flag):
792 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
793 + base = Path(tmpdir)
794 + raw = base / "data/raw/2026-W21.json"
795 + summary = base / "data/candidates/2026-W21/123456/2026-W21-summary.md"
796 + published = base / "data/analyzed/2026-W21-summary.md"
797 + manifest = base / "data/candidates/2026-W21/123456/publish-manifest.json"
798 + gate_report = base / "data/candidates/2026-W21/123456/analysis-gate-report.json"
799 + write_raw(raw)
800 + write_good_summary(summary, quality_score=70)
801 + write_good_summary(published, quality_score=90)
802 + write_gate_report(gate_report)
803 + args = create_args(base, raw, summary, manifest, gate_report=gate_report)
804 + audit_args = [
805 + "--publish-policy",
806 + "force-replace",
807 + "--force-reason",
808 + "operator approved correction",
809 + "--actor",
810 + "jmservera",
811 + ]
812 + missing_index = audit_args.index(missing_flag)
813 + del audit_args[missing_index : missing_index + 2]
814 + args.extend(audit_args)
815 +
816 + self.assertEqual(publish_manifest.main(args), 0)
817 +
818 + payload = json.loads(manifest.read_text(encoding="utf-8"))
819 + self.assertFalse(payload["promotion"]["eligible"])
820 + self.assertIn(expected_reason, payload["promotion"]["reasons"])
821 +
822 + def test_force_replace_does_not_bypass_minimum_quality_score(self) -> None:
823 + tests_root = Path(__file__).resolve().parent
824 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
825 + base = Path(tmpdir)
826 + raw = base / "data/raw/2026-W21.json"
827 + summary = base / "data/candidates/2026-W21/123456/2026-W21-summary.md"
828 + published = base / "data/analyzed/2026-W21-summary.md"
829 + manifest = base / "data/candidates/2026-W21/123456/publish-manifest.json"
830 + gate_report = base / "data/candidates/2026-W21/123456/analysis-gate-report.json"
831 + write_raw(raw)
832 + write_good_summary(summary, quality_score=59)
833 + write_good_summary(published, quality_score=90)
834 + write_gate_report(gate_report)
835 + args = create_args(base, raw, summary, manifest, gate_report=gate_report)
836 + args.extend(
837 + [
838 + "--publish-policy",
839 + "force-replace",
840 + "--force-reason",
841 + "operator approved correction",
842 + "--actor",
843 + "jmservera",
844 + ]
845 + )
846 +
847 + self.assertEqual(publish_manifest.main(args), 0)
848 +
849 + payload = json.loads(manifest.read_text(encoding="utf-8"))
850 + self.assertFalse(payload["promotion"]["eligible"])
851 + self.assertIn("candidate quality_score below 60: 59", payload["promotion"]["reasons"])
852 +
853 + def test_assert_eligible_force_replace_requires_complete_audit(self) -> None:
854 + tests_root = Path(__file__).resolve().parent
855 + with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
856 + base = Path(tmpdir)
857 + raw = base / "data/raw/2026-W21.json"
858 + summary = base / "data/candidates/2026-W21/123456/2026-W21-summary.md"
859 + published = base / "data/analyzed/2026-W21-summary.md"
860 + manifest = base / "data/candidates/2026-W21/123456/publish-manifest.json"
861 + gate_report = base / "data/candidates/2026-W21/123456/analysis-gate-report.json"
862 + write_raw(raw)
863 + write_good_summary(summary, quality_score=70)
864 + write_good_summary(published, quality_score=90)
865 + write_gate_report(gate_report)
866 + args = create_args(base, raw, summary, manifest, gate_report=gate_report)
867 + args.extend(
868 + [
869 + "--publish-policy",
870 + "force-replace",
871 + "--force-reason",
872 + "operator approved correction",
873 + "--actor",
874 + "jmservera",
875 + ]
876 + )
877 + self.assertEqual(publish_manifest.main(args), 0)
878 + valid_payload = json.loads(manifest.read_text(encoding="utf-8"))
879 +
880 + for missing_field in ("actor", "reason"):
881 + with self.subTest(missing_field=missing_field):
882 + tampered = json.loads(json.dumps(valid_payload))
883 + tampered["audit"][missing_field] = None
884 + manifest.write_text(json.dumps(tampered), encoding="utf-8")
885 + with self.assertRaisesRegex(
886 + SystemExit,
887 + "Force replacement requires actor and reason in manifest audit",
888 + ):
889 + assert_eligible_from_root(base, manifest)
890 +
891 + manifest.write_text(json.dumps(valid_payload), encoding="utf-8")
892 + self.assertEqual(assert_eligible_from_root(base, manifest), 0)
893 +
894 def test_missing_candidate_summary_only_reports_missing_summary(self) -> None:
895 tests_root = Path(__file__).resolve().parent
896 with tempfile.TemporaryDirectory(dir=tests_root) as tmpdir:
@@ -920,6 +1069,9 @@ class PublishManifestTests(unittest.TestCase):
1069 )
1070
1071 payload = json.loads(manifest.read_text(encoding="utf-8"))
1072 + # The jmservera/SquadScope#583 override is force-replace-only; normal
1073 + # publication must continue protecting the higher-quality baseline.
1074 + self.assertFalse(payload["promotion"]["eligible"])
1075 self.assertEqual(payload["promotion"]["decision"], "preserve")
1076 self.assertTrue(
1077 any(