analyze_fallback.py · cryptotaxi247/SquadScope-1

Code Commits Issues Pulls Sessions CI/CD

main

py 2,141 lines 82.3 KB

Raw

1	#!/usr/bin/env python3
2	from __future__ import annotations
3
4	import argparse
5	import hashlib
6	import json
7	import os
8	import re
9	import secrets
10	import sys
11	import time
12	from dataclasses import asdict, dataclass
13	from datetime import datetime
14	from pathlib import Path
15	from typing import Any
16	from urllib import error, parse, request
17
18	try:
19	from scripts.assemble_historical_context import (
20	DEFAULT_CONTENT_ROOT,
21	assemble_historical_context,
22	)
23	from scripts.learned_context import render_continuity
24	from scripts.sanitize_repo_content import sanitize_repo_payload
25	except ModuleNotFoundError: # pragma: no cover - script execution path
26	sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
27	from scripts.assemble_historical_context import (
28	DEFAULT_CONTENT_ROOT,
29	assemble_historical_context,
30	)
31	from scripts.learned_context import render_continuity
32	from scripts.sanitize_repo_content import sanitize_repo_payload
33
34	ROOT = Path(__file__).resolve().parent.parent
35	DEFAULT_PROMPT_TEMPLATE = ROOT / "prompts" / "analyze-weekly.md"
36	DEFAULT_ANALYZED_DIR = ROOT / "data" / "analyzed"
37	DEFAULT_WISDOM_FILE = ROOT / ".squad" / "identity" / "wisdom.md"
38	DEFAULT_SKILLS_DIR = ROOT / ".squad" / "skills"
39	DEFAULT_CONTINUITY_FILE = ROOT / ".squad" / "identity" / "continuity.md"
40	DEFAULT_MODELS_ENDPOINT = "https://models.github.ai/inference/chat/completions"
41	DEFAULT_MODELS_MODEL = "openai/gpt-4o"
42	# Synthesis step defaults to the same GitHub Models model unless overridden.
43	DEFAULT_SYNTHESIS_MODEL = DEFAULT_MODELS_MODEL
44	DEFAULT_MODELS_TIMEOUT = 30
45	ALLOWED_MODELS_HOSTS: frozenset[str] = frozenset({"models.github.ai"})
46	_JITTER_RANDOM = secrets.SystemRandom()
47	NO_AI_DIAGNOSTIC_QUALITY_SCORE = 40
48	DEFAULT_PROMPT_TOKEN_BUDGET = 90_000
49	COMPACTED_NEW_REPOS_LIMIT = 25
50	COMPACTED_TRENDING_REPOS_LIMIT = 25
51	COMPACTED_PREVIOUS_SUMMARY_CHARS = 8_000
52	COMPACTED_WISDOM_CHARS = 8_000
53	COMPACTED_SKILLS_CHARS = 10_000
54	COMPACTED_CONTINUITY_CHARS = 8_000
55	COMPACTED_PRESS_CONTEXT_CHARS = 14_000
56	COMPACTED_HISTORICAL_CONTEXT_CHARS = 12_000
57	SYNTHESIS_MAX_TOKENS = 2_000
58	SYNTHESIS_PROMPT_TOKEN_BUDGET = 20_000
59	RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}
60	NON_RETRYABLE_STATUS_CLASSES = {400, 401, 403, 404}
61	MAX_RETRIES = 3
62	BASE_DELAY = 2 # seconds
63
64
65	@dataclass
66	class PromptComponent:
67	name: str
68	path: str \| None
69	included: bool
70	inclusion_reason: str
71	compaction_decision: str
72	bytes: int
73	token_estimate: int
74	checksum_sha256: str
75
76
77	@dataclass
78	class EvidenceRepoRef:
79	full_name: str
80	url: str \| None
81	description: str \| None
82	language: str \| None
83	topics: list[str]
84	source: str
85	stars: int \| None
86	stars_gained: int \| None
87	created_at: str \| None
88
89
90	@dataclass
91	class EvidencePressRef:
92	title: str \| None
93	url: str
94	source: str \| None
95	published_at: str \| None
96	categories: list[str]
97	relevance_score: float \| None
98	correlation_repos: list[str]
99
100
101	@dataclass
102	class EvidenceInventory:
103	name: str
104	path: str
105	item_count: int
106	bytes: int
107	token_estimate: int
108	checksum_sha256: str
109	repos: list[EvidenceRepoRef]
110
111
112	@dataclass
113	class PressInventory:
114	name: str
115	path: str \| None
116	item_count: int
117	bytes: int
118	token_estimate: int
119	checksum_sha256: str
120	articles: list[EvidencePressRef]
121
122
123	@dataclass
124	class EvidenceSliceRef:
125	name: str
126	path: str \| None
127	item_count: int
128	bytes: int
129	token_estimate: int
130	checksum_sha256: str
131	provenance: dict[str, Any]
132	validation_errors: list[str]
133
134
135	@dataclass
136	class PromptPreflight:
137	schema_version: str
138	prompt_token_budget: int
139	prompt_tokens: int
140	prompt_bytes: int
141	prompt_checksum_sha256: str
142	rendered_prompt_estimate: dict[str, int \| str]
143	prompt_within_budget: bool
144	degraded: bool
145	publish_eligible: bool
146	promotion_policy: str
147	degradation_reason: str \| None
148	fallback_policy: str
149	components: list[PromptComponent]
150	deterministic_slices: list[str]
151	generated_evidence_slices: list[EvidenceSliceRef]
152	evidence_slice_payloads: dict[str, dict[str, Any]]
153	evidence_inventories: list[EvidenceInventory]
154	press_inventories: list[PressInventory]
155
156
157	def parse_args(argv: list[str] \| None = None) -> argparse.Namespace:
158	parser = argparse.ArgumentParser(
159	description="Render/preflight weekly analysis prompts or generate diagnostic no-AI output."
160	)
161	parser.add_argument(
162	"--raw-json", required=True, type=Path, help="Path to the weekly raw JSON payload."
163	)
164	parser.add_argument(
165	"--output", required=True, type=Path, help="Path to write the analyzed markdown output."
166	)
167	parser.add_argument(
168	"--current-datetime", required=True, help="ISO-8601 timestamp for the analysis run."
169	)
170	parser.add_argument(
171	"--prompt-template",
172	type=Path,
173	default=DEFAULT_PROMPT_TEMPLATE,
174	help="Prompt template path (defaults to prompts/analyze-weekly.md).",
175	)
176	parser.add_argument(
177	"--analyzed-dir",
178	type=Path,
179	default=DEFAULT_ANALYZED_DIR,
180	help="Directory containing prior weekly summaries.",
181	)
182	parser.add_argument(
183	"--wisdom-file",
184	type=Path,
185	default=DEFAULT_WISDOM_FILE,
186	help="Path to the learned wisdom markdown file.",
187	)
188	parser.add_argument(
189	"--skills-dir",
190	type=Path,
191	default=DEFAULT_SKILLS_DIR,
192	help="Directory containing learned skill markdown files.",
193	)
194	parser.add_argument(
195	"--continuity-file",
196	type=Path,
197	default=DEFAULT_CONTINUITY_FILE,
198	help="Path to the learned continuity capsule markdown file.",
199	)
200	parser.add_argument(
201	"--content-root",
202	type=Path,
203	default=DEFAULT_CONTENT_ROOT,
204	help="Path to the content root used for historical context assembly.",
205	)
206	parser.add_argument(
207	"--press-context",
208	type=Path,
209	default=None,
210	help="Path to rendered press context markdown (appended to prompt).",
211	)
212	parser.add_argument(
213	"--print-prompt",
214	action="store_true",
215	help="Render the prompt to stdout without calling GitHub Models.",
216	)
217	parser.add_argument(
218	"--no-ai",
219	action="store_true",
220	help="Generate a data-only summary without calling any AI API.",
221	)
222	parser.add_argument(
223	"--prompt-token-budget",
224	type=int,
225	default=DEFAULT_PROMPT_TOKEN_BUDGET,
226	help=f"Maximum rendered prompt tokens before model invocation (default: {DEFAULT_PROMPT_TOKEN_BUDGET}).",
227	)
228	parser.add_argument(
229	"--preflight-report-json",
230	type=Path,
231	help="Write deterministic rendered-prompt preflight details as JSON.",
232	)
233	parser.add_argument(
234	"--preflight-report-md",
235	type=Path,
236	help="Write deterministic rendered-prompt preflight details as Markdown.",
237	)
238	parser.add_argument(
239	"--run-synthesis",
240	action="store_true",
241	help="Run Step 1 synthesis (press/historical context → compact narrative) and exit.",
242	)
243	parser.add_argument(
244	"--synthesis-output",
245	type=Path,
246	default=None,
247	help="Path to write the synthesis narrative output (used with --run-synthesis).",
248	)
249	parser.add_argument(
250	"--synthesis-input",
251	type=Path,
252	default=None,
253	help="Path to a pre-computed synthesis narrative to inject into the analysis prompt (Step 2).",
254	)
255	return parser.parse_args(argv)
256
257
258	def load_json(path: Path) -> dict[str, Any]:
259	return json.loads(path.read_text(encoding="utf-8"))
260
261
262	def estimate_tokens(text: str) -> int:
263	"""Deterministic local estimate used for preflight bounds."""
264	return (len(text.encode("utf-8")) + 3) // 4
265
266
267	def checksum_text(text: str) -> str:
268	return hashlib.sha256(text.encode("utf-8")).hexdigest()
269
270
271	def stable_json(payload: Any) -> str:
272	return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
273
274
275	def checksum_payload(payload: Any) -> str:
276	return checksum_text(stable_json(payload))
277
278
279	def _component(
280	*,
281	name: str,
282	content: str,
283	path: Path \| None,
284	included: bool,
285	inclusion_reason: str,
286	compaction_decision: str,
287	) -> PromptComponent:
288	return PromptComponent(
289	name=name,
290	path=path.as_posix() if path else None,
291	included=included,
292	inclusion_reason=inclusion_reason,
293	compaction_decision=compaction_decision,
294	bytes=len(content.encode("utf-8")),
295	token_estimate=estimate_tokens(content),
296	checksum_sha256=checksum_text(content),
297	)
298
299
300	def _repo_int(value: Any) -> int \| None:
301	return value if isinstance(value, int) and not isinstance(value, bool) else None
302
303
304	def _repo_topics(value: Any) -> list[str]:
305	if not isinstance(value, list):
306	return []
307	return [
308	str(topic) for topic in value if isinstance(topic, (str, int, float)) and str(topic).strip()
309	]
310
311
312	REQUIRED_REPO_SLICE_FIELDS = (
313	"full_name",
314	"url",
315	"description",
316	"language",
317	"topics",
318	"stars",
319	"stars_gained",
320	"created_at",
321	)
322
323
324	def compact_repo_record(repo: dict[str, Any], *, source: str) -> dict[str, Any]:
325	full_name = str(repo.get("full_name") or "").strip()
326	url = repo.get("url")
327	return {
328	"full_name": full_name,
329	"url": url
330	if isinstance(url, str) and url.strip()
331	else (f"https://github.com/{full_name}" if full_name else None),
332	"description": repo.get("description")
333	if isinstance(repo.get("description"), str)
334	else None,
335	"language": repo.get("language") if isinstance(repo.get("language"), str) else None,
336	"topics": _repo_topics(repo.get("topics")),
337	"stars": _repo_int(repo.get("stars")),
338	"stars_gained": _repo_int(repo.get("stars_gained")),
339	"created_at": repo.get("created_at") if isinstance(repo.get("created_at"), str) else None,
340	"source": source,
341	}
342
343
344	def _inventory_repo_refs(payload: dict[str, Any], field: str) -> list[EvidenceRepoRef]:
345	repos = payload.get(field)
346	if not isinstance(repos, list):
347	return []
348	refs: list[EvidenceRepoRef] = []
349	for repo in repos:
350	if not isinstance(repo, dict):
351	continue
352	full_name = repo.get("full_name")
353	if not isinstance(full_name, str) or "/" not in full_name:
354	continue
355	url = repo.get("url")
356	refs.append(
357	EvidenceRepoRef(
358	full_name=full_name.strip(),
359	url=url if isinstance(url, str) and url.strip() else None,
360	description=repo.get("description")
361	if isinstance(repo.get("description"), str)
362	else None,
363	language=repo.get("language") if isinstance(repo.get("language"), str) else None,
364	topics=_repo_topics(repo.get("topics")),
365	source=field,
366	stars=_repo_int(repo.get("stars")),
367	stars_gained=_repo_int(repo.get("stars_gained")),
368	created_at=repo.get("created_at")
369	if isinstance(repo.get("created_at"), str)
370	else None,
371	)
372	)
373	return refs
374
375
376	def _evidence_inventory(
377	name: str, payload: dict[str, Any], field: str, path: Path
378	) -> EvidenceInventory:
379	content = json.dumps(payload.get(field, []), indent=2, ensure_ascii=False)
380	repos = _inventory_repo_refs(payload, field)
381	return EvidenceInventory(
382	name=name,
383	path=path.as_posix(),
384	item_count=len(repos),
385	bytes=len(content.encode("utf-8")),
386	token_estimate=estimate_tokens(content),
387	checksum_sha256=checksum_text(content),
388	repos=repos,
389	)
390
391
392	def _press_paths_for_context(
393	press_context_path: Path \| None, week: str
394	) -> tuple[Path \| None, Path \| None]:
395	if press_context_path is None:
396	return None, None
397	data_dir = press_context_path.parent.parent
398	external_path = data_dir / "raw" / f"{week}-external-news.json"
399	legacy_path = data_dir / "raw" / f"{week}-techcrunch.json"
400	corr_path = data_dir / "analyzed" / f"{week}-correlations.json"
401	news_path = (
402	external_path if external_path.exists() else legacy_path if legacy_path.exists() else None
403	)
404	return news_path, corr_path if corr_path.exists() else None
405
406
407	def _safe_load_json(path: Path \| None) -> dict[str, Any] \| None:
408	if path is None or not path.exists():
409	return None
410	try:
411	payload = json.loads(path.read_text(encoding="utf-8"))
412	except (OSError, json.JSONDecodeError):
413	return None
414	return payload if isinstance(payload, dict) else None
415
416
417	def _article_inventory(
418	news_payload: dict[str, Any] \| None,
419	correlation_payload: dict[str, Any] \| None,
420	path: Path \| None,
421	) -> PressInventory:
422	articles = news_payload.get("articles", []) if news_payload else []
423	correlations = correlation_payload.get("correlations", []) if correlation_payload else []
424	repo_by_url: dict[str, set[str]] = {}
425	for corr in correlations if isinstance(correlations, list) else []:
426	if not isinstance(corr, dict):
427	continue
428	repo = corr.get("repo")
429	for url in (
430	corr.get("matched_articles", [])
431	if isinstance(corr.get("matched_articles"), list)
432	else []
433	):
434	if isinstance(url, str) and isinstance(repo, str):
435	repo_by_url.setdefault(url, set()).add(repo)
436	for detail in (
437	corr.get("matched_article_details", [])
438	if isinstance(corr.get("matched_article_details"), list)
439	else []
440	):
441	if (
442	isinstance(detail, dict)
443	and isinstance(detail.get("url"), str)
444	and isinstance(repo, str)
445	):
446	repo_by_url.setdefault(detail["url"], set()).add(repo)
447	refs: list[EvidencePressRef] = []
448	for article in articles if isinstance(articles, list) else []:
449	if (
450	not isinstance(article, dict)
451	or not isinstance(article.get("url"), str)
452	or not article["url"].strip()
453	):
454	continue
455	categories = (
456	article.get("categories") if isinstance(article.get("categories"), list) else []
457	)
458	relevance = article.get("relevance_score")
459	refs.append(
460	EvidencePressRef(
461	title=article.get("title") if isinstance(article.get("title"), str) else None,
462	url=article["url"],
463	source=article.get("source") if isinstance(article.get("source"), str) else None,
464	published_at=article.get("published_at")
465	if isinstance(article.get("published_at"), str)
466	else None,
467	categories=[str(category) for category in categories],
468	relevance_score=float(relevance)
469	if isinstance(relevance, (int, float)) and not isinstance(relevance, bool)
470	else None,
471	correlation_repos=sorted(repo_by_url.get(article["url"], set())),
472	)
473	)
474	content = stable_json([asdict(ref) for ref in refs])
475	return PressInventory(
476	name="press_articles",
477	path=path.as_posix() if path else None,
478	item_count=len(refs),
479	bytes=len(content.encode("utf-8")),
480	token_estimate=estimate_tokens(content),
481	checksum_sha256=checksum_text(content),
482	articles=refs,
483	)
484
485
486	def _source_ref(path: Path \| None, content: str \| None = None) -> dict[str, Any] \| None:
487	if path is None and content is None:
488	return None
489	if content is None:
490	if path is None or not path.exists():
491	return None
492	data = path.read_bytes()
493	return {
494	"path": path.as_posix(),
495	"bytes": len(data),
496	"sha256": hashlib.sha256(data).hexdigest(),
497	}
498	encoded = content.encode("utf-8")
499	return {
500	"path": path.as_posix() if path else None,
501	"bytes": len(encoded),
502	"sha256": checksum_text(content),
503	}
504
505
506	def _slice_checksum_payload(payload: dict[str, Any]) -> dict[str, Any]:
507	stripped = dict(payload)
508	stripped.pop("checksum_sha256", None)
509	return stripped
510
511
512	def validate_evidence_slice(
513	payload: dict[str, Any], *, expected_checksum: str \| None = None
514	) -> list[str]:
515	errors: list[str] = []
516	for field in (
517	"schema_version",
518	"slice_name",
519	"component",
520	"records",
521	"provenance",
522	"checksum_sha256",
523	):
524	if field not in payload:
525	errors.append(f"slice missing {field}")
526	checksum = payload.get("checksum_sha256")
527	if isinstance(checksum, str):
528	actual = checksum_payload(_slice_checksum_payload(payload))
529	if checksum != actual:
530	errors.append("slice checksum mismatch")
531	if expected_checksum is not None and checksum != expected_checksum:
532	errors.append("slice checksum does not match manifest reference")
533	elif "checksum_sha256" in payload:
534	errors.append("slice checksum_sha256 must be a string")
535	records = payload.get("records")
536	if not isinstance(records, list):
537	errors.append("slice records must be a list")
538	records = []
539	provenance = payload.get("provenance")
540	if not isinstance(provenance, dict):
541	errors.append("slice provenance must be an object")
542	else:
543	sources = provenance.get("sources")
544	if not isinstance(sources, dict) or not sources:
545	errors.append("slice provenance sources missing")
546	else:
547	for name, source in sources.items():
548	if (
549	not isinstance(source, dict)
550	or not source.get("sha256")
551	or not isinstance(source.get("bytes"), int)
552	):
553	errors.append(f"slice provenance source {name} missing checksum/bytes")
554	if payload.get("component") in {"new_repos", "trending_repos"}:
555	for index, record in enumerate(records):
556	if not isinstance(record, dict):
557	errors.append(f"record {index} must be an object")
558	continue
559	for field in REQUIRED_REPO_SLICE_FIELDS:
560	if field not in record:
561	errors.append(f"record {index} missing {field}")
562	return errors
563
564
565	def _build_slice(
566	name: str, records: list[dict[str, Any]], provenance: dict[str, Any]
567	) -> dict[str, Any]:
568	payload = {
569	"schema_version": "analysis_evidence_slice_v1",
570	"slice_name": name,
571	"component": name,
572	"records": records,
573	"provenance": provenance,
574	}
575	payload["checksum_sha256"] = checksum_payload(payload)
576	return payload
577
578
579	def build_evidence_slices(
580	*,
581	week: str,
582	raw_path: Path,
583	sanitized_payload: dict[str, Any],
584	payload_for_prompt: dict[str, Any],
585	press_context_path: Path \| None,
586	press_content: str,
587	previous_summary_path: Path \| None,
588	previous_summary_content: str,
589	) -> dict[str, dict[str, Any]]:
590	raw_source = _source_ref(raw_path)
591	press_source = _source_ref(press_context_path, press_content) if press_content else None
592	previous_source = (
593	_source_ref(previous_summary_path, previous_summary_content)
594	if previous_summary_content
595	else None
596	)
597	news_path, corr_path = _press_paths_for_context(press_context_path, week)
598	corr_payload = _safe_load_json(corr_path)
599	news_source = _source_ref(news_path)
600	corr_source = _source_ref(corr_path)
601	base_provenance = {"week": week, "sources": {"raw_json": raw_source} if raw_source else {}}
602	slices = {
603	"new_repos": _build_slice(
604	"new_repos",
605	[
606	compact_repo_record(repo, source="new_repos")
607	for repo in payload_for_prompt.get("new_repos", [])
608	if isinstance(repo, dict)
609	],
610	base_provenance,
611	),
612	"trending_repos": _build_slice(
613	"trending_repos",
614	[
615	compact_repo_record(repo, source="trending_repos")
616	for repo in payload_for_prompt.get("trending_repos", [])
617	if isinstance(repo, dict)
618	],
619	base_provenance,
620	),
621	}
622	press_sources = {}
623	for key, source in (
624	("raw_json", raw_source),
625	("press_context", press_source),
626	("external_news", news_source),
627	("correlations", corr_source),
628	):
629	if source:
630	press_sources[key] = source
631	press_records: list[dict[str, Any]] = []
632	correlations = corr_payload.get("correlations", []) if corr_payload else []
633	for corr in correlations if isinstance(correlations, list) else []:
634	if isinstance(corr, dict):
635	press_records.append(
636	{
637	"repo": corr.get("repo"),
638	"matched_articles": corr.get("matched_articles", []),
639	"matched_article_details": corr.get("matched_article_details", []),
640	"match_type": corr.get("match_type"),
641	"correlation_confidence": corr.get("correlation_confidence"),
642	"correlation_strength": corr.get("correlation_strength"),
643	"hype_risk": corr.get("hype_risk"),
644	}
645	)
646	if not press_records and press_content:
647	urls = sorted(set(re.findall(r"https?://[^\s)\]]+", press_content)))
648	press_records = [
649	{"url": url.rstrip(".,"), "source": "rendered_press_context"} for url in urls
650	]
651	slices["press_correlations"] = _build_slice(
652	"press_correlations",
653	press_records,
654	{"week": week, "sources": press_sources},
655	)
656	prior_sources = {"raw_json": raw_source} if raw_source else {}
657	if previous_source:
658	prior_sources["prior_summary"] = previous_source
659	slices["prior_continuity"] = _build_slice(
660	"prior_continuity",
661	[
662	{
663	"source_path": previous_summary_path.as_posix() if previous_summary_path else None,
664	"present": bool(previous_summary_content),
665	"excerpt": previous_summary_content[:1000],
666	}
667	],
668	{"week": week, "sources": prior_sources},
669	)
670	return slices
671
672
673	def write_evidence_slices(
674	slices: dict[str, dict[str, Any]], manifest_path: Path \| None
675	) -> list[EvidenceSliceRef]:
676	refs: list[EvidenceSliceRef] = []
677	output_dir = manifest_path.parent / "evidence-slices" if manifest_path else None
678	if output_dir:
679	output_dir.mkdir(parents=True, exist_ok=True)
680	for name in sorted(slices):
681	payload = slices[name]
682	checksum = str(payload["checksum_sha256"])
683	text = stable_json(payload)
684	path = output_dir / f"{name}-{checksum[:12]}.json" if output_dir else None
685	if path:
686	path.write_text(text, encoding="utf-8")
687	refs.append(
688	EvidenceSliceRef(
689	name=name,
690	path=path.as_posix() if path else None,
691	item_count=len(payload.get("records", []))
692	if isinstance(payload.get("records"), list)
693	else 0,
694	bytes=len(text.encode("utf-8")),
695	token_estimate=estimate_tokens(text),
696	checksum_sha256=checksum,
697	provenance=payload.get("provenance", {})
698	if isinstance(payload.get("provenance"), dict)
699	else {},
700	validation_errors=validate_evidence_slice(payload),
701	)
702	)
703	return refs
704
705
706	def truncate_with_notice(content: str, limit: int, label: str) -> tuple[str, str]:
707	if len(content) <= limit:
708	return content, "included"
709	omitted = len(content) - limit
710	return (
711	content[:limit].rstrip()
712	+ f"\n\n[Preflight compaction: truncated {label}; omitted {omitted} characters to stay within prompt budget.]",
713	"compacted",
714	)
715
716
717	def _load_yaml(path: Path) -> dict[str, Any]:
718	try:
719	import yaml # type: ignore[import-untyped]
720	except ImportError:
721	return {}
722	if not path.exists():
723	return {}
724	with open(path, encoding="utf-8") as f:
725	payload = yaml.safe_load(f) or {}
726	return payload if isinstance(payload, dict) else {}
727
728
729	def _resolve_existing_path(configured: str \| None, fallback: Path) -> Path:
730	candidates: list[Path] = []
731	if configured:
732	configured_path = Path(configured)
733	candidates.append(
734	configured_path if configured_path.is_absolute() else ROOT / configured_path
735	)
736	if not configured_path.is_absolute():
737	candidates.append(ROOT / ".squad" / configured_path)
738	candidates.append(fallback)
739	for candidate in candidates:
740	if candidate.exists():
741	return candidate
742	return fallback
743
744
745	def resolve_analysis_context_paths() -> tuple[Path, Path, Path]:
746	"""Resolve analysis-specific learned context, avoiding unrelated squad workflow context."""
747	config = _load_yaml(ROOT / "squadscope.topic.yml")
748	topic = config.get("topic") if isinstance(config.get("topic"), dict) else {}
749	learning = config.get("learning") if isinstance(config.get("learning"), dict) else {}
750	topic_id = str(topic.get("id") or "general")
751	wisdom_path = _resolve_existing_path(
752	learning.get("wisdom_file"),
753	ROOT / ".squad" / "topics" / topic_id / "wisdom.md",
754	)
755	skills_path = _resolve_existing_path(
756	learning.get("skills_dir"),
757	ROOT / ".squad" / "topics" / topic_id / "skills",
758	)
759	continuity_path = _resolve_existing_path(
760	learning.get("continuity_file"),
761	ROOT / ".squad" / "topics" / topic_id / "continuity.md",
762	)
763	return wisdom_path, skills_path, continuity_path
764
765
766	def find_previous_summary(current_week: str, analyzed_dir: Path) -> Path \| None:
767	if not analyzed_dir.exists():
768	return None
769
770	candidates = []
771	for path in analyzed_dir.glob("*-summary.md"):
772	week = path.name.removesuffix("-summary.md")
773	if week < current_week:
774	candidates.append(path)
775	return max(candidates, default=None)
776
777
778	def render_wisdom(wisdom_file: Path) -> str:
779	if not wisdom_file.exists():
780	return "_No learned wisdom has been recorded yet._"
781
782	content = wisdom_file.read_text(encoding="utf-8").strip()
783	if not content:
784	return "_No learned wisdom has been recorded yet._"
785	# Sanitize boundary markers to prevent fence escape from prior LLM output
786	from scripts.sanitize_repo_content import _escape_untrusted_boundaries
787
788	return _escape_untrusted_boundaries(content)
789
790
791	def iter_skill_files(skills_dir: Path) -> list[Path]:
792	if not skills_dir.exists():
793	return []
794	return sorted(path for path in skills_dir.rglob("*.md") if path.is_file())
795
796
797	def render_skills(skills_dir: Path) -> str:
798	skill_files = iter_skill_files(skills_dir)
799	if not skill_files:
800	return "_No learned skills have been extracted yet._"
801
802	from scripts.sanitize_repo_content import _escape_untrusted_boundaries
803
804	blocks = []
805	for path in skill_files:
806	relative_path = path.relative_to(ROOT) if path.is_relative_to(ROOT) else path
807	content = path.read_text(encoding="utf-8").strip()
808	if not content:
809	continue
810	# Sanitize boundary markers to prevent fence escape from prior LLM output
811	content = _escape_untrusted_boundaries(content)
812	blocks.append(f"--- Skill Source: {relative_path} ---\n{content}")
813	return "\n\n".join(blocks) if blocks else "_No learned skills have been extracted yet._"
814
815
816	def _sort_repos_for_compaction(repos: list[dict[str, Any]], score_key: str) -> list[dict[str, Any]]:
817	return sorted(
818	repos,
819	key=lambda repo: (
820	int(repo.get(score_key) or 0),
821	int(repo.get("stars") or 0),
822	str(repo.get("full_name") or ""),
823	),
824	reverse=True,
825	)
826
827
828	def compact_payload(payload: dict[str, Any]) -> tuple[dict[str, Any], dict[str, str]]:
829	compacted = dict(payload)
830	decisions = {"new_repos": "included", "trending_repos": "included"}
831	new_repos = payload.get("new_repos")
832	if isinstance(new_repos, list) and len(new_repos) > COMPACTED_NEW_REPOS_LIMIT:
833	compacted["new_repos"] = _sort_repos_for_compaction(new_repos, "stars")[
834	:COMPACTED_NEW_REPOS_LIMIT
835	]
836	decisions["new_repos"] = f"compacted to top {COMPACTED_NEW_REPOS_LIMIT} repos by stars"
837	trending_repos = payload.get("trending_repos")
838	if isinstance(trending_repos, list) and len(trending_repos) > COMPACTED_TRENDING_REPOS_LIMIT:
839	compacted["trending_repos"] = _sort_repos_for_compaction(trending_repos, "stars_gained")[
840	:COMPACTED_TRENDING_REPOS_LIMIT
841	]
842	decisions["trending_repos"] = (
843	f"compacted to top {COMPACTED_TRENDING_REPOS_LIMIT} repos by stars_gained/stars"
844	)
845	if decisions["new_repos"] != "included" or decisions["trending_repos"] != "included":
846	compacted["_preflight_compaction"] = {
847	"reason": "Rendered prompt exceeded explicit token budget before model invocation.",
848	"new_repos_original_count": len(new_repos) if isinstance(new_repos, list) else 0,
849	"trending_repos_original_count": len(trending_repos)
850	if isinstance(trending_repos, list)
851	else 0,
852	"new_repos_decision": decisions["new_repos"],
853	"trending_repos_decision": decisions["trending_repos"],
854	}
855	return compacted, decisions
856
857
858	def _strip_ai_instruction_blocks(text: str) -> str:
859	"""Remove AI-only instruction sections (### Instructions, directives) from press context.
860
861	These blocks are intended for the main analysis prompt and should not be
862	forwarded into synthesis to reduce prompt-injection surface area.
863	"""
864	# Remove markdown sections starting with ### Instructions (case-insensitive)
865	# up to the next same-or-higher-level heading or end of text
866	text = re.sub(
867	r"(?m)^###\s+Instructions?\b.*?(?=^#{1,3}\s\|\Z)",
868	"",
869	text,
870	flags=re.DOTALL \| re.IGNORECASE,
871	)
872	# Remove divergence directive blocks (commonly marked with special tags)
873	text = re.sub(
874	r"(?m)^<!--\s(?:ai-only\|divergence\|directive)\b.?-->.?(?:<!--\s/(?:ai-only\|divergence\|directive)\s*-->\|\Z)",
875	"",
876	text,
877	flags=re.DOTALL \| re.IGNORECASE,
878	)
879	return text.strip()
880
881
882	def _build_synthesis_prompt(
883	*,
884	press_content: str,
885	historical_context_content: str,
886	continuity_content: str,
887	current_week: str,
888	current_datetime: str,
889	) -> str:
890	"""Build a compact prompt for Step 1: Industry & Press Synthesis.
891
892	Input: press context + historical context + continuity capsule.
893	Output instruction: max 2K token narrative of the tech industry landscape this week.
894	"""
895	sections = []
896	sections.append(
897	"You are an expert technology industry analyst. Your task is to synthesize "
898	"the provided press context, historical context, and continuity notes into a "
899	"compact industry narrative (maximum 2000 tokens / ~1500 words).\n\n"
900	"Focus on:\n"
901	"- Key technology trends and shifts happening this week\n"
902	"- Notable industry movements (acquisitions, launches, pivots)\n"
903	"- Developer ecosystem changes\n"
904	"- Connections to longer-term patterns from historical context\n\n"
905	"Output ONLY the narrative — no headers, no metadata, no instructions. "
906	"Write in a dense, information-rich style suitable for feeding into a downstream "
907	"analysis step that will correlate this with GitHub repository data.\n\n"
908	f"Current week: {current_week}\n"
909	f"Current datetime: {current_datetime}\n"
910	)
911	if press_content:
912	sections.append(f"## Press Context\n\n{press_content}")
913	if historical_context_content:
914	sections.append(f"## Historical Context\n\n{historical_context_content}")
915	if (
916	continuity_content
917	and continuity_content != "_No continuity capsule has been recorded yet._"
918	):
919	sections.append(f"## Continuity Notes\n\n{continuity_content}")
920
921	return "\n\n---\n\n".join(sections)
922
923
924	def render_synthesis_prompt(
925	*,
926	press_context_path: Path \| None = None,
927	content_root: Path = DEFAULT_CONTENT_ROOT,
928	continuity_file: Path = DEFAULT_CONTINUITY_FILE,
929	current_datetime: str,
930	current_week: str,
931	previous_summary_path: Path \| None = None,
932	prompt_token_budget: int = SYNTHESIS_PROMPT_TOKEN_BUDGET,
933	) -> str:
934	"""Render the synthesis prompt to a string (for Copilot CLI to process).
935
936	Returns the prompt string, or empty string if no meaningful content exists.
937	"""
938	historical_context_content = assemble_historical_context(
939	current_datetime=current_datetime,
940	previous_summary_path=previous_summary_path,
941	content_root=content_root,
942	max_words=1_500,
943	prompt_token_budget=prompt_token_budget,
944	).strip()
945	from scripts.sanitize_repo_content import _escape_untrusted_boundaries
946
947	historical_context_content = _escape_untrusted_boundaries(historical_context_content)
948	if not historical_context_content:
949	historical_context_content = (
950	"_No historical context was available beyond the current weekly payload._"
951	)
952
953	continuity_content = render_continuity(continuity_file)
954
955	press_content = (
956	press_context_path.read_text(encoding="utf-8").strip()
957	if press_context_path
958	and press_context_path.exists()
959	and press_context_path.stat().st_size > 0
960	else ""
961	)
962
963	# The no-press sentinel is a NON-EMPTY string; treat it as ABSENT so the
964	# synthesis narrative is not built from a fake "## Press Context".
965	from scripts.render_press_context import NO_PRESS_SENTINEL_MARKER
966
967	if press_content and NO_PRESS_SENTINEL_MARKER.search(press_content):
968	press_content = ""
969
970	if press_content:
971	press_content = _strip_ai_instruction_blocks(press_content)
972	if press_content:
973	press_content = _escape_untrusted_boundaries(press_content)
974
975	# If there's no meaningful content to synthesize, return empty
976	if not press_content and historical_context_content.startswith("_No historical context"):
977	return ""
978
979	prompt = _build_synthesis_prompt(
980	press_content=press_content,
981	historical_context_content=historical_context_content,
982	continuity_content=continuity_content,
983	current_week=current_week,
984	current_datetime=current_datetime,
985	)
986
987	# Truncate press content if prompt exceeds budget
988	prompt_tokens = estimate_tokens(prompt)
989	if prompt_tokens > prompt_token_budget:
990	excess_chars = (prompt_tokens - prompt_token_budget) * 4
991	end_index = max(0, len(press_content) - excess_chars)
992	press_content = press_content[:end_index]
993	prompt = _build_synthesis_prompt(
994	press_content=press_content,
995	historical_context_content=historical_context_content,
996	continuity_content=continuity_content,
997	current_week=current_week,
998	current_datetime=current_datetime,
999	)
1000
1001	return prompt
1002
1003
1004	def run_synthesis_step(
1005	*,
1006	press_context_path: Path \| None = None,
1007	content_root: Path = DEFAULT_CONTENT_ROOT,
1008	continuity_file: Path = DEFAULT_CONTINUITY_FILE,
1009	current_datetime: str,
1010	current_week: str,
1011	previous_summary_path: Path \| None = None,
1012	prompt_token_budget: int = SYNTHESIS_PROMPT_TOKEN_BUDGET,
1013	model: str \| None = None,
1014	) -> str:
1015	"""Execute Step 1: synthesize press/historical context into a compact narrative.
1016
1017	Returns the narrative string (max ~2K tokens). Raises RuntimeError on API failure.
1018	"""
1019	historical_context_content = assemble_historical_context(
1020	current_datetime=current_datetime,
1021	previous_summary_path=previous_summary_path,
1022	content_root=content_root,
1023	max_words=1_500,
1024	prompt_token_budget=prompt_token_budget,
1025	).strip()
1026	from scripts.sanitize_repo_content import _escape_untrusted_boundaries
1027
1028	historical_context_content = _escape_untrusted_boundaries(historical_context_content)
1029	if not historical_context_content:
1030	historical_context_content = (
1031	"_No historical context was available beyond the current weekly payload._"
1032	)
1033
1034	continuity_content = render_continuity(continuity_file)
1035
1036	press_content = (
1037	press_context_path.read_text(encoding="utf-8").strip()
1038	if press_context_path
1039	and press_context_path.exists()
1040	and press_context_path.stat().st_size > 0
1041	else ""
1042	)
1043
1044	# The no-press sentinel is a NON-EMPTY string; treat it as ABSENT so
1045	# synthesis does not treat the sentinel as real press.
1046	from scripts.render_press_context import NO_PRESS_SENTINEL_MARKER
1047
1048	if press_content and NO_PRESS_SENTINEL_MARKER.search(press_content):
1049	press_content = ""
1050
1051	# Strip AI-only instruction blocks from press context before synthesis
1052	if press_content:
1053	press_content = _strip_ai_instruction_blocks(press_content)
1054	# Escape boundary markers in untrusted press content
1055	from scripts.sanitize_repo_content import _escape_untrusted_boundaries
1056
1057	if press_content:
1058	press_content = _escape_untrusted_boundaries(press_content)
1059
1060	# If there's no meaningful content to synthesize, return empty
1061	if not press_content and historical_context_content.startswith("_No historical context"):
1062	return ""
1063
1064	prompt = _build_synthesis_prompt(
1065	press_content=press_content,
1066	historical_context_content=historical_context_content,
1067	continuity_content=continuity_content,
1068	current_week=current_week,
1069	current_datetime=current_datetime,
1070	)
1071
1072	# Check that synthesis prompt is within its own budget
1073	prompt_tokens = estimate_tokens(prompt)
1074	if prompt_tokens > SYNTHESIS_PROMPT_TOKEN_BUDGET:
1075	# Truncate press content to fit (clamp to avoid negative index)
1076	excess_chars = (prompt_tokens - SYNTHESIS_PROMPT_TOKEN_BUDGET) * 4
1077	end_index = max(0, len(press_content) - excess_chars)
1078	press_content = press_content[:end_index]
1079	prompt = _build_synthesis_prompt(
1080	press_content=press_content,
1081	historical_context_content=historical_context_content,
1082	continuity_content=continuity_content,
1083	current_week=current_week,
1084	current_datetime=current_datetime,
1085	)
1086
1087	return _call_synthesis_api(prompt, model=model or DEFAULT_SYNTHESIS_MODEL)
1088
1089
1090	def _call_synthesis_api(prompt: str, *, model: str) -> str:
1091	"""Call GitHub Models API for the synthesis step."""
1092	token = os.environ.get("GITHUB_TOKEN")
1093	if not token:
1094	raise RuntimeError("GITHUB_TOKEN is required for synthesis step.")
1095
1096	# Inject canary token for output leak detection
1097	from scripts.canary_token import generate_canary, inject_canary
1098
1099	canary = generate_canary()
1100	prompt = inject_canary(prompt, canary)
1101
1102	endpoint = os.environ.get("GITHUB_MODELS_ENDPOINT", DEFAULT_MODELS_ENDPOINT)
1103	validate_https_url(endpoint, label="GitHub Models endpoint", allowed_hosts=ALLOWED_MODELS_HOSTS)
1104	timeout = int(os.environ.get("GITHUB_MODELS_TIMEOUT", str(DEFAULT_MODELS_TIMEOUT)))
1105	payload = {
1106	"model": model,
1107	"messages": [{"role": "user", "content": prompt}],
1108	"temperature": 0.2,
1109	"max_tokens": SYNTHESIS_MAX_TOKENS, # Cap at documented 2K
1110	}
1111	body = json.dumps(payload).encode("utf-8")
1112
1113	last_exc: Exception \| None = None
1114	for attempt in range(MAX_RETRIES + 1):
1115	req = request.Request(
1116	endpoint,
1117	data=body,
1118	headers={
1119	"Authorization": f"Bearer {token}",
1120	"Content-Type": "application/json",
1121	"Accept": "application/json",
1122	},
1123	method="POST",
1124	)
1125	try:
1126	with request.urlopen(req, timeout=timeout) as response: # nosec B310
1127	response_payload = json.load(response)
1128	markdown = extract_markdown(response_payload)
1129	# Validate output for canary leak and injection artifacts
1130	violations = validate_output_safety(markdown, canary)
1131	if violations:
1132	msg = f"Output safety violations detected: {'; '.join(violations)}"
1133	canary_leaked = any("Canary token leaked" in v for v in violations)
1134	if canary_leaked:
1135	raise RuntimeError(f"BLOCKED: {msg}")
1136	print(f"::warning::{msg}", file=sys.stderr)
1137	return markdown
1138	except error.HTTPError as exc:
1139	if exc.code not in RETRYABLE_STATUS_CODES or attempt == MAX_RETRIES:
1140	detail = exc.read().decode("utf-8", errors="replace")
1141	raise RuntimeError(f"Synthesis API request failed ({exc.code}): {detail}") from exc
1142	# Respect Retry-After header on 429
1143	retry_after = None
1144	if exc.code == 429:
1145	retry_after_header = exc.headers.get("Retry-After") if exc.headers else None
1146	if retry_after_header:
1147	try:
1148	retry_after = float(retry_after_header)
1149	except (ValueError, TypeError):
1150	pass
1151	delay = (
1152	retry_after
1153	if retry_after
1154	else BASE_DELAY ** (attempt + 1) + _JITTER_RANDOM.uniform(0, 1)
1155	)
1156	print(
1157	f"[retry] Synthesis API returned {exc.code}, "
1158	f"retrying in {delay:.1f}s (attempt {attempt + 1}/{MAX_RETRIES})",
1159	file=sys.stderr,
1160	)
1161	last_exc = exc
1162	time.sleep(delay)
1163	except error.URLError as exc:
1164	if attempt == MAX_RETRIES:
1165	raise RuntimeError(f"Synthesis API network error: {exc.reason}") from exc
1166	delay = BASE_DELAY ** (attempt + 1) + _JITTER_RANDOM.uniform(0, 1)
1167	print(
1168	f"[retry] Synthesis API network error: {exc.reason}, "
1169	f"retrying in {delay:.1f}s (attempt {attempt + 1}/{MAX_RETRIES})",
1170	file=sys.stderr,
1171	)
1172	last_exc = exc
1173	time.sleep(delay)
1174
1175	raise RuntimeError("Synthesis API request failed after retries") from last_exc
1176
1177
1178	def _build_prompt(
1179	*,
1180	prompt_template_path: Path,
1181	raw_json_path: Path,
1182	output_path: Path,
1183	current_datetime: str,
1184	analyzed_dir: Path,
1185	content_root: Path = DEFAULT_CONTENT_ROOT,
1186	wisdom_file: Path = DEFAULT_WISDOM_FILE,
1187	skills_dir: Path = DEFAULT_SKILLS_DIR,
1188	continuity_file: Path = DEFAULT_CONTINUITY_FILE,
1189	press_context_path: Path \| None = None,
1190	prompt_token_budget: int = DEFAULT_PROMPT_TOKEN_BUDGET,
1191	allow_compaction: bool = True,
1192	synthesis_narrative: str \| None = None,
1193	) -> tuple[str, PromptPreflight]:
1194	payload = load_json(raw_json_path)
1195	sanitized_payload = sanitize_repo_payload(payload)
1196	current_week = sanitized_payload["week"]
1197	previous_summary_path = find_previous_summary(current_week, analyzed_dir)
1198	previous_summary_content = (
1199	previous_summary_path.read_text(encoding="utf-8") if previous_summary_path else ""
1200	)
1201	historical_context_content = assemble_historical_context(
1202	current_datetime=current_datetime,
1203	previous_summary_path=previous_summary_path,
1204	content_root=content_root,
1205	max_words=1_500,
1206	prompt_token_budget=prompt_token_budget,
1207	).strip()
1208	from scripts.sanitize_repo_content import _escape_untrusted_boundaries
1209
1210	historical_context_content = _escape_untrusted_boundaries(historical_context_content)
1211	previous_summary_content = _escape_untrusted_boundaries(previous_summary_content)
1212	if not historical_context_content:
1213	historical_context_content = (
1214	"_No historical context was available beyond the current weekly payload._"
1215	)
1216	wisdom_content = render_wisdom(wisdom_file)
1217	skills_content = render_skills(skills_dir)
1218	continuity_content = render_continuity(continuity_file)
1219	press_content = (
1220	press_context_path.read_text(encoding="utf-8").strip()
1221	if press_context_path
1222	and press_context_path.exists()
1223	and press_context_path.stat().st_size > 0
1224	else ""
1225	)
1226	# The no-press sentinel is a NON-EMPTY string, so treat it as ABSENT here:
1227	# blanking it keeps the "press exists" logic (## Press Context block,
1228	# included=bool(press_content), press_decision) and the required no-press
1229	# statement correct for genuinely press-less weeks.
1230	from scripts.render_press_context import NO_PRESS_SENTINEL_MARKER
1231
1232	if press_content and NO_PRESS_SENTINEL_MARKER.search(press_content):
1233	press_content = ""
1234	if press_content:
1235	press_content = _strip_ai_instruction_blocks(press_content)
1236	press_content = _escape_untrusted_boundaries(press_content)
1237	# When a synthesis narrative is available (Step 1 output), it distils the
1238	# historical context into a compact narrative that replaces the bulky
1239	# historical context block and saves tokens. It must NOT drop the press
1240	# context: the Step-2 sections "Where Industry Meets Code" and
1241	# "Press & Industry" still have to be written from the real press data.
1242	# jmservera/SquadScope#515 blanked press_content here, which silently
1243	# dropped a populated press context and forced the model to emit
1244	# "No industry press data was available...". Keep a condensed press
1245	# context so those sections stay evidence-backed.
1246	press_condensed_for_synthesis = False
1247	if synthesis_narrative:
1248	synthesis_source = "press & historical context" if press_content else "historical context"
1249	historical_context_content = (
1250	f"[Industry narrative synthesized from {synthesis_source}]\n\n{synthesis_narrative}"
1251	)
1252	if press_content and len(press_content) > COMPACTED_PRESS_CONTEXT_CHARS:
1253	press_content, _ = truncate_with_notice(
1254	press_content, COMPACTED_PRESS_CONTEXT_CHARS, "press context"
1255	)
1256	press_condensed_for_synthesis = True
1257	payload_for_prompt = sanitized_payload
1258	raw_decisions = {"new_repos": "included", "trending_repos": "included"}
1259	previous_decision = "included" if previous_summary_path else "not included: no previous summary"
1260	historical_context_decision = (
1261	"included"
1262	if historical_context_content
1263	!= "_No historical context was available beyond the current weekly payload._"
1264	else "not included: no historical sources available"
1265	)
1266	wisdom_decision = (
1267	"included" if wisdom_file.exists() else "not included: no analysis-specific wisdom file"
1268	)
1269	skills_decision = (
1270	"included"
1271	if skills_dir.exists() and iter_skill_files(skills_dir)
1272	else "not included: no analysis-specific skills"
1273	)
1274	continuity_decision = (
1275	"included"
1276	if continuity_file.exists()
1277	else "not included: no analysis-specific continuity capsule"
1278	)
1279	if not press_content:
1280	press_decision = "not included: no press context"
1281	elif press_condensed_for_synthesis:
1282	press_decision = "included: condensed alongside synthesis narrative"
1283	else:
1284	press_decision = "included"
1285	degraded = False
1286
1287	def assemble() -> str:
1288	raw_json_content = json.dumps(payload_for_prompt, indent=2, ensure_ascii=False)
1289	current_year, _, week_number = current_week.partition("-W")
1290	generic_title_example = (
1291	f"Week {int(week_number)}, {current_year} Analysis"
1292	if week_number.isdigit()
1293	else "Week NN, YYYY Analysis"
1294	)
1295	prompt = prompt_template_path.read_text(encoding="utf-8")
1296	try:
1297	current_month = datetime.fromisoformat(
1298	current_datetime.strip().replace("Z", "+00:00")
1299	).strftime("%B")
1300	except (ValueError, TypeError):
1301	current_month = ""
1302	replacements = {
1303	"{{CURRENT_DATETIME}}": current_datetime,
1304	"{{CURRENT_WEEK}}": current_week,
1305	"{{CURRENT_YEAR}}": current_year,
1306	"{{CURRENT_MONTH}}": current_month,
1307	"{{TITLE_TEMPLATE_HINT}}": (
1308	f"Specific editorial headline about {current_week}'s dominant themes "
1309	f'(not "{generic_title_example}")'
1310	),
1311	"{{RAW_JSON_PATH}}": str(raw_json_path),
1312	"{{OUTPUT_PATH}}": str(output_path),
1313	"{{PREVIOUS_SUMMARY_PATH_OR_NONE}}": str(previous_summary_path)
1314	if previous_summary_path
1315	else "None",
1316	"{{HISTORICAL_CONTEXT}}": historical_context_content,
1317	"{{RAW_JSON_CONTENT}}": raw_json_content,
1318	"{{PREVIOUS_SUMMARY_CONTENT_OR_EMPTY}}": previous_summary_content.strip(),
1319	"{{WISDOM}}": wisdom_content,
1320	"{{SKILLS}}": skills_content,
1321	"{{CONTINUITY}}": continuity_content,
1322	}
1323	for needle, value in replacements.items():
1324	prompt = prompt.replace(needle, value)
1325	if press_content:
1326	prompt += f"\n\n---\n## Press Context\n\n{press_content}\n"
1327	return prompt
1328
1329	prompt = assemble()
1330	if allow_compaction and estimate_tokens(prompt) > prompt_token_budget:
1331	degraded = True
1332	payload_for_prompt, raw_decisions = compact_payload(sanitized_payload)
1333	previous_summary_content, previous_decision = truncate_with_notice(
1334	previous_summary_content, COMPACTED_PREVIOUS_SUMMARY_CHARS, "prior continuity"
1335	)
1336	if historical_context_decision == "included":
1337	historical_context_content, historical_context_decision = truncate_with_notice(
1338	historical_context_content,
1339	COMPACTED_HISTORICAL_CONTEXT_CHARS,
1340	"historical context",
1341	)
1342	wisdom_content, wisdom_decision = truncate_with_notice(
1343	wisdom_content, COMPACTED_WISDOM_CHARS, "analysis wisdom"
1344	)
1345	skills_content, skills_decision = truncate_with_notice(
1346	skills_content, COMPACTED_SKILLS_CHARS, "analysis skills"
1347	)
1348	continuity_content, continuity_decision = truncate_with_notice(
1349	continuity_content, COMPACTED_CONTINUITY_CHARS, "analysis continuity"
1350	)
1351	press_content, press_decision = truncate_with_notice(
1352	press_content, COMPACTED_PRESS_CONTEXT_CHARS, "press correlations"
1353	)
1354	prompt = assemble()
1355
1356	raw_json_content = json.dumps(payload_for_prompt, indent=2, ensure_ascii=False)
1357	current_year, _, week_number = current_week.partition("-W")
1358	components = [
1359	_component(
1360	name="prompt_template",
1361	content=prompt_template_path.read_text(encoding="utf-8"),
1362	path=prompt_template_path,
1363	included=True,
1364	inclusion_reason="Base weekly analysis instructions.",
1365	compaction_decision="included",
1366	),
1367	_component(
1368	name="new_repos",
1369	content=json.dumps(
1370	payload_for_prompt.get("new_repos", []), indent=2, ensure_ascii=False
1371	),
1372	path=raw_json_path,
1373	included=True,
1374	inclusion_reason="Deterministic mapper slice: newly discovered repositories.",
1375	compaction_decision=raw_decisions["new_repos"],
1376	),
1377	_component(
1378	name="trending_repos",
1379	content=json.dumps(
1380	payload_for_prompt.get("trending_repos", []), indent=2, ensure_ascii=False
1381	),
1382	path=raw_json_path,
1383	included=True,
1384	inclusion_reason="Deterministic mapper slice: continuing/trending repositories.",
1385	compaction_decision=raw_decisions["trending_repos"],
1386	),
1387	_component(
1388	name="raw_metadata",
1389	content=raw_json_content,
1390	path=raw_json_path,
1391	included=True,
1392	inclusion_reason=f"Sanitized current weekly payload for {current_year}-W{week_number}.",
1393	compaction_decision="included"
1394	if not degraded
1395	else "included with compacted repo slices",
1396	),
1397	_component(
1398	name="prior_continuity",
1399	content=previous_summary_content,
1400	path=previous_summary_path,
1401	included=bool(previous_summary_path),
1402	inclusion_reason="Deterministic mapper slice: prior weekly continuity.",
1403	compaction_decision=previous_decision,
1404	),
1405	_component(
1406	name="historical_context",
1407	content=historical_context_content,
1408	path=content_root,
1409	included=bool(historical_context_content),
1410	inclusion_reason="Bounded historical context synthesized from rolling, previous-week, monthly, and yearly reports.",
1411	compaction_decision=historical_context_decision,
1412	),
1413	_component(
1414	name="analysis_wisdom",
1415	content=wisdom_content,
1416	path=wisdom_file,
1417	included=wisdom_file.exists(),
1418	inclusion_reason="Analysis-specific wisdom capsule from topic learning state.",
1419	compaction_decision=wisdom_decision,
1420	),
1421	_component(
1422	name="analysis_skills",
1423	content=skills_content,
1424	path=skills_dir,
1425	included=skills_dir.exists() and bool(iter_skill_files(skills_dir)),
1426	inclusion_reason="Analysis-specific learned skill capsule from topic learning state.",
1427	compaction_decision=skills_decision,
1428	),
1429	_component(
1430	name="analysis_continuity",
1431	content=continuity_content,
1432	path=continuity_file,
1433	included=continuity_file.exists(),
1434	inclusion_reason="Analysis continuity capsule distilled from recent multi-week learnings.",
1435	compaction_decision=continuity_decision,
1436	),
1437	_component(
1438	name="press_correlations",
1439	content=press_content,
1440	path=press_context_path,
1441	included=bool(press_content),
1442	inclusion_reason="Deterministic mapper slice: press/developer correlation context.",
1443	compaction_decision=press_decision,
1444	),
1445	_component(
1446	name="rendered_prompt",
1447	content=prompt,
1448	path=None,
1449	included=True,
1450	inclusion_reason="Exact prompt that will be passed to Copilot CLI.",
1451	compaction_decision="included"
1452	if not degraded
1453	else "included after deterministic compaction",
1454	),
1455	]
1456	prompt_tokens = estimate_tokens(prompt)
1457	prompt_within_budget = prompt_tokens <= prompt_token_budget
1458	degradation_reason = (
1459	"Prompt was deterministically compacted to fit the configured token budget."
1460	if degraded
1461	else None
1462	)
1463	evidence_slices = build_evidence_slices(
1464	week=current_week,
1465	raw_path=raw_json_path,
1466	sanitized_payload=sanitized_payload,
1467	payload_for_prompt=payload_for_prompt,
1468	press_context_path=press_context_path,
1469	press_content=press_content,
1470	previous_summary_path=previous_summary_path,
1471	previous_summary_content=previous_summary_content,
1472	)
1473	news_path, corr_path = _press_paths_for_context(press_context_path, current_week)
1474	press_inventory = _article_inventory(
1475	_safe_load_json(news_path), _safe_load_json(corr_path), news_path
1476	)
1477	slice_refs = write_evidence_slices(evidence_slices, None)
1478	preflight = PromptPreflight(
1479	schema_version="analysis_input_manifest_v1",
1480	prompt_token_budget=prompt_token_budget,
1481	prompt_tokens=prompt_tokens,
1482	prompt_bytes=len(prompt.encode("utf-8")),
1483	prompt_checksum_sha256=checksum_text(prompt),
1484	rendered_prompt_estimate={
1485	"bytes": len(prompt.encode("utf-8")),
1486	"tokens": prompt_tokens,
1487	"checksum_sha256": checksum_text(prompt),
1488	},
1489	prompt_within_budget=prompt_within_budget,
1490	degraded=degraded,
1491	publish_eligible=prompt_within_budget,
1492	promotion_policy=(
1493	"normal-promotion"
1494	if prompt_within_budget
1495	else "staged/candidate-only by default; prompt exceeds token budget."
1496	),
1497	degradation_reason=degradation_reason,
1498	fallback_policy=(
1499	"copilot-only; no GitHub Models/OpenAI fallback. no-ai is diagnostic/staged-only and publish-ineligible. "
1500	"degraded/compacted prompts are staged/candidate-only by default."
1501	),
1502	components=components,
1503	deterministic_slices=[
1504	"new_repos",
1505	"trending_repos",
1506	"press_correlations",
1507	"prior_continuity",
1508	],
1509	generated_evidence_slices=slice_refs,
1510	evidence_slice_payloads=evidence_slices,
1511	evidence_inventories=[
1512	_evidence_inventory("raw_new_repos", sanitized_payload, "new_repos", raw_json_path),
1513	_evidence_inventory(
1514	"raw_trending_repos", sanitized_payload, "trending_repos", raw_json_path
1515	),
1516	_evidence_inventory("prompt_new_repos", payload_for_prompt, "new_repos", raw_json_path),
1517	_evidence_inventory(
1518	"prompt_trending_repos", payload_for_prompt, "trending_repos", raw_json_path
1519	),
1520	],
1521	press_inventories=[press_inventory],
1522	)
1523	return prompt, preflight
1524
1525
1526	def render_prompt(
1527	*,
1528	prompt_template_path: Path,
1529	raw_json_path: Path,
1530	output_path: Path,
1531	current_datetime: str,
1532	analyzed_dir: Path,
1533	content_root: Path = DEFAULT_CONTENT_ROOT,
1534	wisdom_file: Path = DEFAULT_WISDOM_FILE,
1535	skills_dir: Path = DEFAULT_SKILLS_DIR,
1536	continuity_file: Path = DEFAULT_CONTINUITY_FILE,
1537	press_context_path: Path \| None = None,
1538	) -> str:
1539	if (
1540	wisdom_file == DEFAULT_WISDOM_FILE
1541	and skills_dir == DEFAULT_SKILLS_DIR
1542	and continuity_file == DEFAULT_CONTINUITY_FILE
1543	):
1544	wisdom_file, skills_dir, continuity_file = resolve_analysis_context_paths()
1545	prompt, _ = _build_prompt(
1546	prompt_template_path=prompt_template_path,
1547	raw_json_path=raw_json_path,
1548	output_path=output_path,
1549	current_datetime=current_datetime,
1550	analyzed_dir=analyzed_dir,
1551	content_root=content_root,
1552	wisdom_file=wisdom_file,
1553	skills_dir=skills_dir,
1554	continuity_file=continuity_file,
1555	press_context_path=press_context_path,
1556	allow_compaction=False,
1557	)
1558	return prompt
1559
1560
1561	def write_preflight_reports(
1562	preflight: PromptPreflight, json_path: Path \| None, md_path: Path \| None
1563	) -> None:
1564	if json_path and preflight.evidence_slice_payloads:
1565	preflight.generated_evidence_slices = write_evidence_slices(
1566	preflight.evidence_slice_payloads, json_path
1567	)
1568	if json_path:
1569	json_path.parent.mkdir(parents=True, exist_ok=True)
1570	json_path.write_text(
1571	json.dumps(asdict(preflight), indent=2, sort_keys=True) + "\n", encoding="utf-8"
1572	)
1573	if md_path:
1574	md_path.parent.mkdir(parents=True, exist_ok=True)
1575	rows = [
1576	"# Analysis Prompt Preflight",
1577	"",
1578	f"- Prompt budget: `{preflight.prompt_token_budget}` tokens",
1579	f"- Rendered prompt: `{preflight.prompt_tokens}` tokens / `{preflight.prompt_bytes}` bytes",
1580	f"- Prompt checksum: `{preflight.prompt_checksum_sha256}`",
1581	f"- Degraded/compacted: `{str(preflight.degraded).lower()}`",
1582	f"- Degradation reason: {preflight.degradation_reason or 'none'}",
1583	f"- Publish eligible: `{str(preflight.publish_eligible).lower()}`",
1584	f"- Promotion policy: {preflight.promotion_policy}",
1585	f"- Fallback policy: {preflight.fallback_policy}",
1586	f"- Deterministic slices: {', '.join(preflight.deterministic_slices)}",
1587	"",
1588	"\| Component \| Included \| Bytes \| Tokens \| Checksum \| Path \| Inclusion reason \| Compaction decision \|",
1589	"\| --- \| --- \| ---: \| ---: \| --- \| --- \| --- \| --- \|",
1590	]
1591	for component in preflight.components:
1592	rows.append(
1593	"\| "
1594	+ " \| ".join(
1595	[
1596	component.name,
1597	str(component.included).lower(),
1598	str(component.bytes),
1599	str(component.token_estimate),
1600	component.checksum_sha256,
1601	component.path or "",
1602	component.inclusion_reason.replace("\|", "\\\|"),
1603	component.compaction_decision.replace("\|", "\\\|"),
1604	]
1605	)
1606	+ " \|"
1607	)
1608	md_path.write_text("\n".join(rows) + "\n", encoding="utf-8")
1609
1610
1611	def extract_markdown(response_payload: dict[str, Any]) -> str:
1612	choices = response_payload.get("choices") or []
1613	if not choices:
1614	raise ValueError("GitHub Models response did not include any choices.")
1615
1616	message = choices[0].get("message") or {}
1617	content = message.get("content")
1618
1619	if isinstance(content, str):
1620	return content.strip() + "\n"
1621
1622	if isinstance(content, list):
1623	parts: list[str] = []
1624	for item in content:
1625	if isinstance(item, dict):
1626	text = item.get("text") or item.get("output_text")
1627	if text:
1628	parts.append(text)
1629	if parts:
1630	return "\n".join(parts).strip() + "\n"
1631
1632	text = choices[0].get("text")
1633	if isinstance(text, str) and text.strip():
1634	return text.strip() + "\n"
1635
1636	raise ValueError("GitHub Models response did not contain markdown output.")
1637
1638
1639	def validate_https_url(
1640	url: str, *, label: str, allowed_hosts: frozenset[str] \| None = None
1641	) -> None:
1642	parsed = parse.urlparse(url)
1643	if parsed.scheme.lower() != "https":
1644	raise ValueError(f"{label} must use HTTPS: {url}")
1645	if parsed.username or parsed.password:
1646	raise ValueError(f"{label} must not include credentials: {url}")
1647	if not parsed.hostname:
1648	raise ValueError(f"{label} must include a hostname: {url}")
1649	try:
1650	port = parsed.port
1651	except ValueError as exc:
1652	raise ValueError(f"{label} has an invalid port: {url}") from exc
1653	if port not in (None, 443):
1654	raise ValueError(f"{label} must not use unexpected ports: {url}")
1655	if allowed_hosts is not None and parsed.hostname.lower() not in allowed_hosts:
1656	raise ValueError(f"{label} host must be one of {sorted(allowed_hosts)}: {url}")
1657
1658
1659	def validate_output_safety(output: str, canary: str \| None = None) -> list[str]:
1660	"""Check generated analysis output for canary leaks and injection artifacts.
1661
1662	Returns a list of security violation messages (empty = safe).
1663	"""
1664	from scripts.canary_token import check_output_for_any_canary, check_output_for_leak
1665
1666	violations: list[str] = []
1667
1668	# Check for specific canary leak
1669	if canary:
1670	result = check_output_for_leak(output, canary)
1671	if result.leaked:
1672	violations.append(
1673	f"Canary token leaked at position {result.match_position}: "
1674	f"model may have been manipulated by injected instructions"
1675	)
1676
1677	# Check for any canary pattern (catches leaks from prior invocations)
1678	any_result = check_output_for_any_canary(output)
1679	if any_result.leaked and (not canary or any_result.canary.lower() != canary.lower()):
1680	violations.append(
1681	f"Unknown canary pattern '{any_result.canary}' found at position "
1682	f"{any_result.match_position}: possible cross-invocation leak"
1683	)
1684
1685	# Check for boundary marker leaks (model reproduced internal framing)
1686	from scripts.sanitize_repo_content import BOUNDARY_CLOSE, BOUNDARY_OPEN
1687
1688	if BOUNDARY_OPEN in output:
1689	violations.append(
1690	"Output contains <untrusted-content> boundary marker — "
1691	"model may have leaked prompt structure"
1692	)
1693	if BOUNDARY_CLOSE in output:
1694	violations.append(
1695	"Output contains </untrusted-content> boundary marker — "
1696	"model may have leaked prompt structure"
1697	)
1698
1699	return violations
1700
1701
1702	def call_github_models(prompt: str) -> str:
1703	token = os.environ.get("GITHUB_TOKEN")
1704	if not token:
1705	raise RuntimeError("GITHUB_TOKEN is required for GitHub Models fallback.")
1706
1707	# Inject canary token for output leak detection
1708	from scripts.canary_token import generate_canary, inject_canary
1709
1710	canary = generate_canary()
1711	prompt = inject_canary(prompt, canary)
1712
1713	endpoint = os.environ.get("GITHUB_MODELS_ENDPOINT", DEFAULT_MODELS_ENDPOINT)
1714	validate_https_url(endpoint, label="GitHub Models endpoint", allowed_hosts=ALLOWED_MODELS_HOSTS)
1715	model = os.environ.get("GITHUB_MODELS_MODEL", DEFAULT_MODELS_MODEL)
1716	timeout = int(os.environ.get("GITHUB_MODELS_TIMEOUT", str(DEFAULT_MODELS_TIMEOUT)))
1717	payload = {
1718	"model": model,
1719	"messages": [{"role": "user", "content": prompt}],
1720	"temperature": 0.3,
1721	}
1722	body = json.dumps(payload).encode("utf-8")
1723
1724	last_exc: Exception \| None = None
1725	for attempt in range(MAX_RETRIES + 1):
1726	req = request.Request(
1727	endpoint,
1728	data=body,
1729	headers={
1730	"Authorization": f"Bearer {token}",
1731	"Content-Type": "application/json",
1732	"Accept": "application/json",
1733	},
1734	method="POST",
1735	)
1736	try:
1737	with request.urlopen(req, timeout=timeout) as response: # nosec B310
1738	response_payload = json.load(response)
1739	markdown = extract_markdown(response_payload)
1740	# Validate output for canary leak and injection artifacts
1741	violations = validate_output_safety(markdown, canary)
1742	if violations:
1743	msg = f"Output safety violations detected: {'; '.join(violations)}"
1744	# Full canary leak = prompt injection confirmed; block publishing
1745	canary_leaked = any("Canary token leaked" in v for v in violations)
1746	if canary_leaked:
1747	raise RuntimeError(f"BLOCKED: {msg}")
1748	# Partial/boundary leaks are warnings — log but allow
1749	print(f"::warning::{msg}", file=sys.stderr)
1750	return markdown
1751	except error.HTTPError as exc:
1752	if exc.code not in RETRYABLE_STATUS_CODES or attempt == MAX_RETRIES:
1753	detail = exc.read().decode("utf-8", errors="replace")
1754	retry_class = (
1755	"non-retryable"
1756	if exc.code in NON_RETRYABLE_STATUS_CLASSES
1757	or exc.code not in RETRYABLE_STATUS_CODES
1758	else "retry-exhausted"
1759	)
1760	access_hint = (
1761	" GitHub Models access is unavailable for this model."
1762	if exc.code == 403
1763	else ""
1764	)
1765	raise RuntimeError(
1766	f"GitHub Models API request failed ({exc.code}, {retry_class}): {detail}{access_hint}"
1767	) from exc
1768	# Determine delay: respect Retry-After header on 429
1769	retry_after = (
1770	exc.headers.get("Retry-After")
1771	if exc.code == 429 and exc.headers is not None
1772	else None
1773	)
1774	if retry_after is not None:
1775	try:
1776	delay = float(retry_after)
1777	except ValueError:
1778	delay = BASE_DELAY ** (attempt + 1)
1779	else:
1780	delay = BASE_DELAY ** (attempt + 1)
1781	jitter = _JITTER_RANDOM.uniform(0, 1)
1782	total_delay = delay + jitter
1783	print(
1784	f"[retry] GitHub Models API returned {exc.code}, "
1785	f"retrying in {total_delay:.1f}s (attempt {attempt + 1}/{MAX_RETRIES})",
1786	file=sys.stderr,
1787	)
1788	last_exc = exc
1789	time.sleep(total_delay)
1790	except error.URLError as exc:
1791	if attempt == MAX_RETRIES:
1792	raise RuntimeError(f"GitHub Models API request failed: {exc.reason}") from exc
1793	delay = BASE_DELAY ** (attempt + 1) + _JITTER_RANDOM.uniform(0, 1)
1794	print(
1795	f"[retry] GitHub Models API network error: {exc.reason}, "
1796	f"retrying in {delay:.1f}s (attempt {attempt + 1}/{MAX_RETRIES})",
1797	file=sys.stderr,
1798	)
1799	last_exc = exc
1800	time.sleep(delay)
1801
1802	# Should not be reached, but satisfy type checkers
1803	raise RuntimeError("GitHub Models API request failed after retries") from last_exc
1804
1805
1806	def _strip_ai_instructions(content: str) -> str:
1807	"""Remove AI-facing instruction blocks from a rendered press context string.
1808
1809	Strips:
1810	- The "### Instructions" section (from that heading to the next "###" or EOF)
1811	- The "#### Divergence Instructions" block (heading + bullet items)
1812	- Truncates the "### Correlation Summary" list to the first 10 entries,
1813	appending a "…and N more" summary line when truncation occurs.
1814	"""
1815	import re # noqa: PLC0415
1816
1817	# Strip ### Instructions section (to next ### heading or EOF)
1818	content = re.sub(
1819	r"\n### Instructions\n.*?(?=\n###\|\Z)",
1820	"",
1821	content,
1822	flags=re.DOTALL,
1823	)
1824
1825	# Strip #### Divergence Instructions block (to next #### / ### heading or EOF)
1826	content = re.sub(
1827	r"\n#### Divergence Instructions\n.*?(?=\n####\|\n###\|\Z)",
1828	"",
1829	content,
1830	flags=re.DOTALL,
1831	)
1832
1833	# Truncate correlations list to top 10
1834	corr_match = re.search(
1835	r"(### Correlation Summary\n[^\n]\n)((?:- [^\n]\n?)+)",
1836	content,
1837	)
1838	if corr_match:
1839	header = corr_match.group(1)
1840	list_block = corr_match.group(2)
1841	list_lines = [ln for ln in list_block.splitlines() if ln.startswith("- ")]
1842	total = len(list_lines)
1843	if total > 10:
1844	omitted = total - 10
1845	truncated = "\n".join(list_lines[:10])
1846	truncated += f"\n…and {omitted} more repos with press correlation\n"
1847	content = (
1848	content[: corr_match.start()] + header + truncated + content[corr_match.end() :]
1849	)
1850
1851	# Truncate divergence lists to top 10 items each
1852	for section_header in (
1853	r"#### 🔍 Tech Trends Without Dev Activity",
1854	r"#### 🚀 Dev Activity Without Press Coverage",
1855	):
1856	div_match = re.search(
1857	rf"({re.escape(section_header)}\n[^\n]\n\n?)((?:- [^\n]\n?)+)",
1858	content,
1859	)
1860	if div_match:
1861	header = div_match.group(1)
1862	list_block = div_match.group(2)
1863	list_lines = [ln for ln in list_block.splitlines() if ln.startswith("- ")]
1864	total = len(list_lines)
1865	if total > 10:
1866	omitted = total - 10
1867	truncated = "\n".join(list_lines[:10])
1868	truncated += f"\n- …and {omitted} more topics\n"
1869	content = (
1870	content[: div_match.start()] + header + truncated + content[div_match.end() :]
1871	)
1872
1873	# Add reader-friendly conclusion if divergences exist but instructions were stripped
1874	if "### Divergence Analysis" in content and "Divergence Instructions" not in content:
1875	if "These divergences highlight" not in content:
1876	content = content.rstrip()
1877	content += (
1878	"\n\nThese divergences highlight gaps between what the tech industry "
1879	"is reporting and what developers are actually building.\n"
1880	)
1881
1882	return content.strip()
1883
1884
1885	def _render_press_section_no_ai(press_context_path: Path \| None) -> str:
1886	"""Render press context data for the no-AI summary (reader-facing)."""
1887	if (
1888	not press_context_path
1889	or not press_context_path.exists()
1890	or press_context_path.stat().st_size == 0
1891	):
1892	return (
1893	"No industry press data was available for this week's analysis. "
1894	"Future runs with TechCrunch integration enabled will provide "
1895	"correlation analysis between developer activity and industry coverage, "
1896	"highlighting press-driven hype versus organic growth patterns."
1897	)
1898
1899	# Try to re-render from raw data using reader_mode=True so the narrative
1900	# divergence format (from PR #136) is used instead of the AI-prompt format.
1901	stem = press_context_path.stem # e.g. "2026-W21-press-context"
1902	week = stem.replace("-press-context", "") # e.g. "2026-W21"
1903	data_dir = press_context_path.parent.parent # data/analyzed/ -> data/
1904	external_path = data_dir / "raw" / f"{week}-external-news.json"
1905	legacy_path = data_dir / "raw" / f"{week}-techcrunch.json"
1906	tc_path = external_path if external_path.exists() else legacy_path
1907	corr_path = data_dir / "analyzed" / f"{week}-correlations.json"
1908
1909	if tc_path.exists():
1910	from scripts.render_press_context import load_json as rpc_load_json
1911	from scripts.render_press_context import render_press_context
1912
1913	tc_data = rpc_load_json(tc_path)
1914	corr_data = rpc_load_json(corr_path) if corr_path.exists() else {}
1915	if tc_data is not None:
1916	return render_press_context(tc_data, corr_data or {}, week, reader_mode=True)
1917
1918	# Fallback: strip AI instructions from the pre-rendered file.
1919	content = press_context_path.read_text(encoding="utf-8").strip()
1920	return _strip_ai_instructions(content)
1921
1922
1923	def generate_no_ai_summary(
1924	raw_json_path: Path, current_datetime: str, press_context_path: Path \| None = None
1925	) -> str:
1926	"""Generate a valid summary from raw JSON without any AI API calls."""
1927	payload = sanitize_repo_payload(load_json(raw_json_path))
1928	week = payload["week"]
1929	new_repos = payload.get("new_repos", [])
1930	trending_repos = payload.get("trending_repos", [])
1931	signals = payload.get("signals", {})
1932	raw_topics = signals.get("top_topics", [])
1933	top_topics = [t["topic"] if isinstance(t, dict) else str(t) for t in raw_topics]
1934
1935	total_stars = sum(r.get("stars", 0) for r in new_repos + trending_repos)
1936	repos_featured = len(new_repos) + len(trending_repos)
1937
1938	all_repos = sorted(new_repos + trending_repos, key=lambda r: r.get("stars", 0), reverse=True)
1939	top_repo = all_repos[0]["full_name"] if all_repos else "unknown/unknown"
1940
1941	tags = (
1942	top_topics[:5] if len(top_topics) >= 3 else ["open-source", "developer-tools", "automation"]
1943	)
1944
1945	# Notable new repos
1946	notable_new = sorted(new_repos, key=lambda r: r.get("stars", 0), reverse=True)[:10]
1947	notable_lines = []
1948	for repo in notable_new:
1949	desc = repo.get("description") or "No description provided"
1950	lang = repo.get("language") or "Unknown"
1951	notable_lines.append(
1952	f"- [{repo['full_name']}]({repo.get('url', '#')}) ({lang}, "
1953	f"{repo.get('stars', 0):,} stars): {desc}"
1954	)
1955	notable_section = (
1956	"\n".join(notable_lines)
1957	if notable_lines
1958	else "No new repositories were captured this week."
1959	)
1960
1961	# Language breakdown
1962	lang_counts: dict[str, int] = {}
1963	for repo in all_repos:
1964	lang = repo.get("language")
1965	if lang:
1966	lang_counts[lang] = lang_counts.get(lang, 0) + 1
1967	top_langs = sorted(lang_counts.items(), key=lambda x: x[1], reverse=True)[:5]
1968	lang_summary = (
1969	", ".join(f"{lang} ({count})" for lang, count in top_langs)
1970	if top_langs
1971	else "diverse mix of languages"
1972	)
1973
1974	year_str = week.split("-W")[0]
1975	week_num = week.split("-W")[1]
1976	topics_str = ", ".join(top_topics[:8]) if top_topics else "not available from this crawl"
1977	title_topics = [topic.replace("-", " ").title() for topic in top_topics[:2] if topic]
1978	if len(title_topics) == 2:
1979	fallback_title = f"{title_topics[0]}, {title_topics[1]}, and This Week's Repo Signals"
1980	elif len(title_topics) == 1:
1981	fallback_title = f"{title_topics[0]} Leads This Week's Repo Signals"
1982	else:
1983	fallback_title = f"{top_repo.split('/')[-1]} Leads This Week's Repo Signals"
1984
1985	markdown = f'''---
1986	title: "{fallback_title}"
1987	date: {current_datetime}
1988	week: "{week}"
1989	year: {int(year_str)}
1990	tags: [{", ".join(tags)}]
1991	categories: [weekly]
1992	repos_featured: {repos_featured}
1993	stars_tracked: {total_stars}
1994	top_repo: "{top_repo}"
1995	quality_score: {NO_AI_DIAGNOSTIC_QUALITY_SCORE}
1996	summary: "Automated data-only summary for {week}. AI analysis was unavailable; this report presents raw crawl statistics and top repositories without editorial commentary."
1997	---
1998
1999	## This Week's Trends
2000
2001	Without AI-powered analysis, this section reports observed patterns from crawl data rather than synthesized editorial trends. The top community topics this week are {topics_str}, and the dominant languages are {lang_summary}. These signals point to where developer attention is concentrated, though qualitative interpretation of which patterns are durable versus incidental requires a full AI-enabled analysis run.
2002
2003	The crawler captured {repos_featured} repositories this week ({len(new_repos)} new, {len(trending_repos)} trending) with {total_stars:,} cumulative stars. The top repository by star count is [{top_repo}](https://github.com/{top_repo}). Raw patterns suggest continued investment in {lang_summary}, but without editorial judgment these should be treated as directional rather than conclusive.
2004
2005	## Where Industry Meets Code
2006
2007	{_render_press_section_no_ai(press_context_path)}
2008
2009	## Signal & Noise
2010
2011	The primary observable signal this week comes from language and topic distribution. The top languages are {lang_summary}. The top community topics are {topics_str}. These patterns indicate where developer attention is concentrating and what categories are gaining traction relative to prior weeks.
2012
2013	Without AI-powered filtering, distinguishing signal from noise requires manual review. Some repositories in the crawl may represent low-quality forks, exploit tools, or promotional projects that inflate topic counts without contributing meaningful innovation. Future AI-enabled runs will provide better noise filtering and critical editorial judgment.
2014
2015	## Blind Spots
2016
2017	This automated summary lacks the editorial judgment that AI analysis would normally provide. Specific blind spots in this report include: comparative trend analysis against prior weeks, qualitative assessment of repository significance, identification of emerging ecosystem patterns not visible from raw metrics, and filtering of low-signal entries that inflate topic counts. The raw data is preserved for future re-analysis when AI capabilities become available.
2018
2019	## The Week Ahead
2020
2021	Week {week_num} of {year_str} captured {repos_featured} repositories with {total_stars:,} cumulative stars tracked. The top repository is [{top_repo}](https://github.com/{top_repo}). This summary was generated without AI assistance and presents factual crawl statistics only. A full analytical run should be attempted when AI model access is restored to provide trend synthesis and editorial judgment.
2022
2023	## Key References
2024
2025	### Notable Projects
2026
2027	{notable_section}
2028
2029	### Press & Industry
2030
2031	{_render_press_section_no_ai(press_context_path) if press_context_path else "No press data was provided this week."}
2032	'''
2033	return markdown.strip() + "\n"
2034
2035
2036	def main(argv: list[str] \| None = None) -> int:
2037	args = parse_args(argv)
2038	wisdom_file = args.wisdom_file
2039	skills_dir = args.skills_dir
2040	continuity_file = args.continuity_file
2041	if (
2042	wisdom_file == DEFAULT_WISDOM_FILE
2043	and skills_dir == DEFAULT_SKILLS_DIR
2044	and continuity_file == DEFAULT_CONTINUITY_FILE
2045	):
2046	wisdom_file, skills_dir, continuity_file = resolve_analysis_context_paths()
2047
2048	# Step 1: Run synthesis if requested
2049	if args.run_synthesis:
2050	payload = load_json(args.raw_json)
2051	sanitized_payload = sanitize_repo_payload(payload)
2052	current_week = sanitized_payload["week"]
2053	previous_summary_path = find_previous_summary(current_week, args.analyzed_dir)
2054	# Render synthesis prompt to file (Copilot CLI will process it)
2055	narrative_or_prompt = render_synthesis_prompt(
2056	press_context_path=args.press_context,
2057	content_root=args.content_root,
2058	continuity_file=continuity_file,
2059	current_datetime=args.current_datetime,
2060	current_week=current_week,
2061	previous_summary_path=previous_summary_path,
2062	prompt_token_budget=args.prompt_token_budget,
2063	)
2064	if not narrative_or_prompt:
2065	print(
2066	"::warning::No meaningful content for synthesis (no press or historical context).",
2067	file=sys.stderr,
2068	)
2069	return 1
2070	output_path = args.synthesis_output or args.output
2071	output_path.parent.mkdir(parents=True, exist_ok=True)
2072	output_path.write_text(narrative_or_prompt, encoding="utf-8")
2073	print(
2074	f"::notice::Synthesis prompt rendered: {estimate_tokens(narrative_or_prompt)} tokens written to {output_path}",
2075	file=sys.stderr,
2076	)
2077	return 0
2078
2079	# Load synthesis narrative from Step 1 output if provided
2080	synthesis_narrative: str \| None = None
2081	if (
2082	args.synthesis_input
2083	and args.synthesis_input.exists()
2084	and args.synthesis_input.stat().st_size > 0
2085	):
2086	synthesis_narrative = args.synthesis_input.read_text(encoding="utf-8").strip()
2087	if synthesis_narrative:
2088	# Escape boundary markers — synthesis output is untrusted LLM content
2089	from scripts.sanitize_repo_content import _escape_untrusted_boundaries
2090
2091	synthesis_narrative = _escape_untrusted_boundaries(synthesis_narrative)
2092	print(
2093	f"::notice::Using synthesis narrative ({estimate_tokens(synthesis_narrative)} tokens) from {args.synthesis_input}",
2094	file=sys.stderr,
2095	)
2096
2097	prompt, preflight = _build_prompt(
2098	prompt_template_path=args.prompt_template,
2099	raw_json_path=args.raw_json,
2100	output_path=args.output,
2101	current_datetime=args.current_datetime,
2102	analyzed_dir=args.analyzed_dir,
2103	content_root=args.content_root,
2104	wisdom_file=wisdom_file,
2105	skills_dir=skills_dir,
2106	continuity_file=continuity_file,
2107	press_context_path=args.press_context,
2108	prompt_token_budget=args.prompt_token_budget,
2109	allow_compaction=True,
2110	synthesis_narrative=synthesis_narrative,
2111	)
2112	write_preflight_reports(preflight, args.preflight_report_json, args.preflight_report_md)
2113
2114	if not preflight.prompt_within_budget:
2115	print(
2116	"::error::Rendered analysis prompt exceeds explicit budget after deterministic compaction: "
2117	f"{preflight.prompt_tokens}/{preflight.prompt_token_budget} tokens.",
2118	file=sys.stderr,
2119	)
2120	return 1
2121
2122	if args.print_prompt:
2123	sys.stdout.write(prompt)
2124	return 0
2125
2126	if args.no_ai:
2127	markdown = generate_no_ai_summary(args.raw_json, args.current_datetime, args.press_context)
2128	else:
2129	print(
2130	"::error::GitHub Models/OpenAI analysis fallback is disabled; use Copilot CLI or --no-ai for staged diagnostics.",
2131	file=sys.stderr,
2132	)
2133	return 1
2134
2135	args.output.parent.mkdir(parents=True, exist_ok=True)
2136	args.output.write_text(markdown, encoding="utf-8")
2137	return 0
2138
2139
2140	if __name__ == "__main__":
2141	raise SystemExit(main())