last30days.py - last30days-skill

last30days.py

1	#!/usr/bin/env python3
2	# fmt: off
3	# ruff: noqa: E402
4	"""last30days CLI."""
5
6	from __future__ import annotations
7
8	import argparse
9	import atexit
10	import datetime
11	import hashlib
12	import json
13	import os
14	import re
15	import signal
16	import sqlite3
17	import sys
18	import threading
19	from collections.abc import Callable
20	from pathlib import Path
21
22	MIN_PYTHON = (3, 12)
23
24
25	def ensure_supported_python(version_info: tuple[int, int, int] \| object \| None = None) -> None:
26	if version_info is None:
27	version_info = sys.version_info
28	major, minor, micro = tuple(version_info[:3])
29	if (major, minor) >= MIN_PYTHON:
30	return
31	req = f"{MIN_PYTHON[0]}.{MIN_PYTHON[1]}"
32	sys.stderr.write(
33	f"last30days v3 requires Python {req}+.\n"
34	f"Detected Python {major}.{minor}.{micro}.\n"
35	f"Install with:\n"
36	f" Mac: brew install python@{req}\n"
37	f" Windows: winget install Python.Python.{req}\n"
38	f" Linux: sudo apt install python{req} (or pyenv install {req})\n"
39	f"Then rerun: python{req} <path-to-script> setup\n"
40	)
41	raise SystemExit(1)
42
43
44	ensure_supported_python()
45
46	if os.name == "nt":
47	for stream in (sys.stdout, sys.stderr):
48	if hasattr(stream, "reconfigure"):
49	stream.reconfigure(encoding="utf-8", errors="replace")
50
51	SCRIPT_DIR = Path(__file__).parent.resolve()
52	sys.path.insert(0, str(SCRIPT_DIR))
53
54	from lib import competitors as competitors_mod, corpus, dates, discovery_handoff, env, freshness, html_render, http, permission_preflight, pipeline, registers, render, schema, ui, x_envelope
55
56	_child_pids: set[int] = set()
57	_child_pids_lock = threading.Lock()
58
59
60	def register_child_pid(pid: int) -> None:
61	with _child_pids_lock:
62	_child_pids.add(pid)
63
64
65	def unregister_child_pid(pid: int) -> None:
66	with _child_pids_lock:
67	_child_pids.discard(pid)
68
69
70	def _cleanup_children() -> None:
71	with _child_pids_lock:
72	pids = list(_child_pids)
73	for pid in pids:
74	try:
75	if hasattr(os, "killpg"):
76	os.killpg(os.getpgid(pid), signal.SIGTERM)
77	else:
78	os.kill(pid, signal.SIGTERM)
79	except (ProcessLookupError, PermissionError, OSError):
80	continue
81
82
83	atexit.register(_cleanup_children)
84
85
86	def parse_meta_ads_page(raw: str) -> str:
87	"""Extract an Ad Library page id from a flag value, or "" if there is none.
88
89	Accepts a bare numeric id or any Ad Library URL carrying
90	``view_all_page_id``. A ``facebook.com/<vanity>`` URL is deliberately
91	rejected rather than guessed at: a vanity handle is not a page id, and one
92	live check resolved a brand-looking handle to a private person's profile.
93	"""
94	value = str(raw or "").strip()
95	if not value:
96	return ""
97	if re.fullmatch(r"\d{5,20}", value):
98	return value
99	match = re.search(r"view_all_page_id=(\d{5,20})", value)
100	return match.group(1) if match else ""
101
102
103	def parse_search_flag(raw: str, flag_name: str = "--search") -> list[str]:
104	sources = []
105	for source in raw.split(","):
106	source = source.strip().lower()
107	if not source:
108	continue
109	normalized = pipeline.SEARCH_ALIAS.get(source, source)
110	if normalized not in pipeline.MOCK_AVAILABLE_SOURCES:
111	raise SystemExit(f"Unknown search source in {flag_name}: {source}")
112	if normalized not in sources:
113	sources.append(normalized)
114	if not sources:
115	raise SystemExit(f"{flag_name} requires at least one source.")
116	return sources
117
118	def parse_as_of_date_arg(value: str) -> str:
119	try:
120	parsed = dates.parse_as_of_date(value)
121	except ValueError as exc:
122	raise argparse.ArgumentTypeError(str(exc)) from exc
123	return parsed
124
125	def resolve_requested_sources(args_search: str \| None, config: dict) -> list[str] \| None:
126	"""Resolve the requested source set: explicit --search wins, then the
127	LAST30DAYS_DEFAULT_SEARCH config key (env var or .env file), then None
128	(per-query default behavior). The config fallback lets users pin a fixed
129	source set that survives upgrades without patching SKILL.md (#442).
130	"""
131	if args_search:
132	return parse_search_flag(args_search)
133	default_search = (config.get("LAST30DAYS_DEFAULT_SEARCH") or "").strip()
134	if default_search:
135	return parse_search_flag(default_search, flag_name="LAST30DAYS_DEFAULT_SEARCH")
136	return None
137
138
139	def add_deep_research_source(
140	requested_sources: list[str] \| None,
141	) -> list[str] \| None:
142	"""Add Perplexity without replacing the default-source sentinel.
143
144	``None`` means that the planner can use the normal configured source set.
145	Deep Research enables Perplexity through ``INCLUDE_SOURCES`` separately, so
146	converting this sentinel to ``["perplexity"]`` would suppress every normal
147	source.
148	"""
149	if requested_sources is None:
150	return None
151	if "perplexity" in requested_sources:
152	return requested_sources
153	return [*requested_sources, "perplexity"]
154
155
156	def enable_deep_research_source(config: dict) -> None:
157	"""Enable the exact Perplexity token or reject a hard exclusion."""
158	excluded = {
159	token.strip().lower()
160	for token in str(config.get("EXCLUDE_SOURCES") or "").split(",")
161	if token.strip()
162	}
163	if "perplexity" in excluded:
164	raise ValueError(
165	"--deep-research conflicts with EXCLUDE_SOURCES=perplexity"
166	)
167
168	include = str(config.get("INCLUDE_SOURCES") or "")
169	tokens = [token.strip() for token in include.split(",") if token.strip()]
170	if "perplexity" not in {token.lower() for token in tokens}:
171	tokens.append("perplexity")
172	config["INCLUDE_SOURCES"] = ",".join(tokens)
173
174
175	def plan_has_explicit_trustpilot_domain(comp_plan: dict \| None) -> bool:
176	"""True when any --competitors-plan entry pins a trustpilot_domain."""
177	if not comp_plan:
178	return False
179	for entry in comp_plan.values():
180	if not isinstance(entry, dict):
181	continue
182	domain = entry.get("trustpilot_domain")
183	if isinstance(domain, str) and domain.strip():
184	return True
185	return False
186
187
188	def activate_trustpilot_for_explicit_domain(
189	config: dict,
190	requested_sources: list[str] \| None,
191	*,
192	reason: str,
193	) -> list[str] \| None:
194	"""Activate the opt-in Trustpilot source when the user pinned a domain.
195
196	Passing ``--trustpilot-domain`` (or a plan-level ``trustpilot_domain``) is
197	unambiguous intent — silently ignoring it when Trustpilot is not in
198	``INCLUDE_SOURCES`` / ``--search`` is the #873 failure mode. Auto-resolve
199	hints must not call this helper.
200
201	``EXCLUDE_SOURCES=trustpilot`` still wins. Mutates ``config`` in place and
202	returns the (possibly extended) ``requested_sources`` list.
203	"""
204	excluded = {
205	token.strip().lower()
206	for token in str(config.get("EXCLUDE_SOURCES") or "").split(",")
207	if token.strip()
208	}
209	if "trustpilot" in excluded:
210	sys.stderr.write(
211	f"[Trustpilot] {reason} ignored: trustpilot is in EXCLUDE_SOURCES\n"
212	)
213	return requested_sources
214
215	include = str(config.get("INCLUDE_SOURCES") or "")
216	tokens = [token.strip() for token in include.split(",") if token.strip()]
217	if "trustpilot" not in {token.lower() for token in tokens}:
218	tokens.append("trustpilot")
219	config["INCLUDE_SOURCES"] = ",".join(tokens)
220	sys.stderr.write(
221	f"[Trustpilot] {reason} activated trustpilot source "
222	"(add to INCLUDE_SOURCES permanently to skip this auto-enable)\n"
223	)
224
225	if requested_sources is not None and "trustpilot" not in requested_sources:
226	requested_sources = [*requested_sources, "trustpilot"]
227	return requested_sources
228
229
230	def activate_telegram_for_explicit_sources(
231	config: dict,
232	requested_sources: list[str] \| None,
233	*,
234	channels: str,
235	) -> list[str] \| None:
236	"""Activate the opt-in Telegram source when the user pinned channel(s).
237
238	Passing ``--telegram-sources`` is unambiguous intent — silently ignoring it
239	when Telegram is not in ``INCLUDE_SOURCES`` / ``--search`` is the same
240	failure mode as #873 (Trustpilot). Auto-activate the source.
241
242	``EXCLUDE_SOURCES=telegram`` still wins. Mutates ``config`` in place and
243	returns the (possibly extended) ``requested_sources`` list.
244	"""
245	excluded = {
246	token.strip().lower()
247	for token in str(config.get("EXCLUDE_SOURCES") or "").split(",")
248	if token.strip()
249	}
250	if "telegram" in excluded:
251	sys.stderr.write(
252	f"[Telegram] --telegram-sources={channels} ignored: telegram is in EXCLUDE_SOURCES\n"
253	)
254	return requested_sources
255
256	config["TELEGRAM_SOURCES"] = channels
257
258	include = str(config.get("INCLUDE_SOURCES") or "")
259	tokens = [token.strip() for token in include.split(",") if token.strip()]
260	if "telegram" not in {token.lower() for token in tokens}:
261	tokens.append("telegram")
262	config["INCLUDE_SOURCES"] = ",".join(tokens)
263	sys.stderr.write(
264	f"[Telegram] --telegram-sources={channels} activated telegram source "
265	"(add to INCLUDE_SOURCES permanently to skip this auto-enable)\n"
266	)
267
268	if requested_sources is not None and "telegram" not in requested_sources:
269	requested_sources = [*requested_sources, "telegram"]
270	return requested_sources
271
272
273	def slugify(value: str, max_length: int = 180) -> str:
274	slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
275	if len(slug) > max_length:
276	# Filenames built from long topics can exceed the OS 255-byte limit
277	# (macOS errno 63). Truncate and append a hash of the full value so
278	# distinct long topics still get distinct, deterministic names.
279	digest = hashlib.sha1(slug.encode("utf-8")).hexdigest()[:10]
280	slug = f"{slug[:max_length].rstrip('-')}-{digest}"
281	return slug or "last30days"
282
283
284	def sanitize_suffix(suffix: str) -> str:
285	"""Sanitize a user-provided ``--save-suffix`` into a path-safe token.
286
287	The suffix is glued directly into the saved-report filename, so restrict it
288	to the same ``[a-z0-9-]`` class as the topic slug. This neutralizes path
289	separators and parent refs (``/``, ``..``) so a suffix can never escape the
290	save directory, while leaving ordinary values ('v3', 'gemini', a client
291	slug) unchanged. Unlike ``slugify`` there is no fallback token: a suffix
292	that sanitizes to nothing simply drops, yielding no suffix part.
293	"""
294	return re.sub(r"[^a-z0-9]+", "-", suffix.lower()).strip("-")
295
296
297	def _report_has_private_corpus(report: schema.Report) -> bool:
298	items_by_source = getattr(report, "items_by_source", {})
299	if isinstance(items_by_source, dict) and items_by_source.get("corpus"):
300	return True
301	candidates = getattr(report, "ranked_candidates", ())
302	if not isinstance(candidates, (list, tuple)):
303	return False
304	return any(
305	candidate.source == "corpus"
306	or any(item.source == "corpus" for item in candidate.source_items)
307	for candidate in candidates
308	)
309
310
311	def _ensure_output_directory(path: Path, *, private: bool) -> None:
312	if not private:
313	path.mkdir(parents=True, exist_ok=True)
314	return
315	missing: list[Path] = []
316	current = path
317	while not current.exists():
318	missing.append(current)
319	current = current.parent
320	path.mkdir(parents=True, exist_ok=True, mode=0o700)
321	for directory in missing:
322	directory.chmod(0o700)
323
324
325	def save_output(
326	report: schema.Report,
327	emit: str,
328	save_dir: str,
329	suffix: str = "",
330	synthesis_md: str \| None = None,
331	topic_override: str \| None = None,
332	rendered_content: str \| None = None,
333	json_profile: str = "agent",
334	register: str = "default",
335	private: bool \| None = None,
336	render_fn: Callable[[Path], str] \| None = None,
337	) -> Path:
338	from datetime import datetime
339	path = Path(save_dir).expanduser().resolve()
340	slug = slugify(topic_override or report.topic)
341	extension = "json" if emit == "json" else "html" if emit == "html" else "md"
342	raw_label = "raw-html" if emit == "html" else "raw"
343	safe_suffix = sanitize_suffix(suffix)
344	suffix_part = f"-{safe_suffix}" if safe_suffix else ""
345	base = path / f"{slug}-{raw_label}{suffix_part}.{extension}"
346	date_str = datetime.now().strftime('%Y-%m-%d')
347	candidates = [base]
348	candidates.append(path / f"{slug}-{raw_label}{suffix_part}-{date_str}.{extension}")
349	for i in range(1, 100):
350	candidates.append(path / f"{slug}-{raw_label}{suffix_part}-{date_str}-{i}.{extension}")
351	# Markdown saves keep the complete debug artifact. JSON and HTML preserve
352	# their requested wire format so file extensions match their content.
353	# When render_fn is supplied, content is produced after O_EXCL allocates
354	# the candidate. This lets the footer cite the file actually written
355	# without racing a separate filesystem probe.
356	if render_fn is None:
357	if rendered_content is not None:
358	static_content = rendered_content
359	elif emit in {"json", "html"}:
360	static_content = emit_output(
361	report,
362	emit,
363	synthesis_md=synthesis_md,
364	json_profile=json_profile,
365	register=register,
366	)
367	else:
368	static_content = render.render_full(report)
369	private_corpus = _report_has_private_corpus(report) or bool(private)
370	_ensure_output_directory(path, private=private_corpus)
371	for candidate in candidates:
372	try:
373	fd = os.open(
374	candidate,
375	os.O_CREAT \| os.O_EXCL \| os.O_WRONLY,
376	0o600 if private_corpus else 0o644,
377	)
378	except FileExistsError:
379	continue
380	try:
381	with os.fdopen(fd, "wb") as f:
382	content = render_fn(candidate) if render_fn is not None else static_content
383	f.write(content.encode("utf-8"))
384	except BaseException:
385	# Deferred rendering happens after the candidate is reserved. Do
386	# not leave an empty or partial report if rendering or writing fails.
387	try:
388	candidate.unlink(missing_ok=True)
389	except OSError:
390	pass
391	raise
392	if candidate.suffix.lower() == ".md":
393	try:
394	from lib import library, library_index
395
396	save_root = candidate.parent.resolve()
397	if save_root == Path(library.DEFAULT_MEMORY_DIR).expanduser().resolve():
398	library_index.sync_library(save_root)
399	else:
400	# A scoped save must sync a per-directory index with the
401	# same paths scoped search uses; syncing the shared DB
402	# from one scope's scan would prune other scopes' rows.
403	library_index.sync_library(
404	save_root,
405	save_root / "briefings",
406	db_path=save_root / ".last30days-library.db",
407	)
408	except (library_index.LibrarySearchUnavailable, OSError, sqlite3.DatabaseError):
409	# Saving research must not depend on the optional local index;
410	# `library search` reports a clear capability error on demand.
411	pass
412	return candidate
413	# Fallback: all 101 candidates existed (extremely unlikely).
414	raise RuntimeError(
415	f"save_output: could not find a unique filename after 101 attempts in {path}"
416	)
417
418
419	def save_rendered_output(
420	rendered_content: str,
421	output_file: str,
422	*,
423	private: bool = False,
424	) -> Path:
425	out_path = Path(output_file).expanduser().resolve()
426	_ensure_output_directory(out_path.parent, private=private)
427	if private and out_path.exists():
428	out_path.chmod(0o600)
429	fd = os.open(
430	out_path,
431	os.O_CREAT \| os.O_TRUNC \| os.O_WRONLY,
432	0o600 if private else 0o644,
433	)
434	with os.fdopen(fd, "w", encoding="utf-8") as handle:
435	handle.write(rendered_content)
436	if private:
437	out_path.chmod(0o600)
438	return out_path
439
440
441	def _publish_metadata_path(html_path: Path) -> Path:
442	return html_path.with_name(f"{html_path.name}.publish.json")
443
444
445	def _write_publish_metadata(html_path: Path, publish_result: dict[str, object]) -> None:
446	payload = {
447	"url": publish_result.get("url"),
448	"site_id": publish_result.get("site_id"),
449	"status": publish_result.get("status"),
450	"published_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
451	}
452	_publish_metadata_path(html_path).write_text(json.dumps(payload, indent=2), encoding="utf-8")
453
454
455	def publish_rendered_html(
456	rendered: str,
457	*,
458	password: str \| None = None,
459	companion_paths: list[Path] \| None = None,
460	) -> dict[str, object]:
461	from lib import html_publish
462
463	result = html_publish.publish_html(rendered, password=password)
464	metadata_errors: list[str] = []
465	for path in companion_paths or []:
466	try:
467	_write_publish_metadata(path, result)
468	except OSError as exc:
469	metadata_errors.append(f"{path}: {exc}")
470	if metadata_errors:
471	result = dict(result)
472	result["_metadata_errors"] = metadata_errors
473	return result
474
475
476	def _publish_password_for_args(
477	args: argparse.Namespace,
478	config: dict[str, object] \| None = None,
479	) -> str \| None:
480	return (
481	args.publish_password
482	or env.read_secret_env("LAST30DAYS_PUBLISH_PASSWORD")
483	or (config or {}).get("LAST30DAYS_PUBLISH_PASSWORD")
484	or None
485	)
486
487
488	def emit_output(
489	report: schema.Report,
490	emit: str,
491	fun_level: str = "medium",
492	save_path: str \| None = None,
493	synthesis_md: str \| None = None,
494	json_profile: str = "agent",
495	register: str = "default",
496	) -> str:
497	if emit == "json":
498	payload = (
499	schema.to_dict(report)
500	if json_profile == "raw"
501	else schema.to_agent_export(report)
502	)
503	return json.dumps(payload, indent=2, sort_keys=True)
504	if emit == "html":
505	return html_render.render_html(
506	report,
507	fun_level=fun_level,
508	save_path=save_path,
509	synthesis_md=synthesis_md,
510	register=register,
511	)
512	if emit in {"compact", "md"}:
513	return render.render_compact(
514	report,
515	fun_level=fun_level,
516	save_path=save_path,
517	register=register,
518	)
519	if emit == "context":
520	return render.render_context(report)
521	if emit == "brief":
522	return render.render_brief(report)
523	raise SystemExit(f"Unsupported emit mode: {emit}")
524
525
526	def emit_comparison_output(
527	entity_reports: list[tuple[str, schema.Report]],
528	emit: str,
529	fun_level: str = "medium",
530	save_path: str \| None = None,
531	synthesis_md: str \| None = None,
532	json_profile: str = "agent",
533	) -> str:
534	if emit == "json":
535	payload = {
536	"comparison": True,
537	"entities": [label for label, _ in entity_reports],
538	"reports": [
539	{
540	"entity": label,
541	"report": (
542	schema.to_dict(report)
543	if json_profile == "raw"
544	else schema.to_agent_export(report)
545	),
546	}
547	for label, report in entity_reports
548	],
549	}
550	if json_profile == "agent":
551	payload["schema_version"] = schema.AGENT_EXPORT_SCHEMA_VERSION
552	return json.dumps(payload, indent=2, sort_keys=True)
553	if emit == "html":
554	return html_render.render_html_comparison(
555	entity_reports,
556	fun_level=fun_level,
557	save_path=save_path,
558	synthesis_md=synthesis_md,
559	)
560	if emit in {"compact", "md"}:
561	return render.render_comparison_multi(
562	entity_reports, fun_level=fun_level, save_path=save_path,
563	)
564	if emit == "context":
565	return render.render_comparison_multi_context(entity_reports)
566	raise SystemExit(f"Unsupported emit mode: {emit}")
567
568
569	def comparison_topic(entity_reports: list[tuple[str, schema.Report]]) -> str:
570	return " vs ".join(label for label, _ in entity_reports)
571
572
573	def comparison_label_key(label: str) -> str:
574	"""Normalize an entity label for duplicate detection.
575
576	Comparison labels double as keys in the fan-out's results dict, so two
577	entities differing only in case, surrounding space, or a repeated space
578	collide there while still looking distinct on the command line. Spaces
579	are collapsed, never stripped: "Open AI" stays distinct from "OpenAI".
580	"""
581	return " ".join(label.split()).casefold()
582
583
584	def compute_save_path_display(save_dir: str, topic: str, suffix: str, emit: str) -> str:
585	"""Compute the user-friendly save path string that will be shown in the footer.
586
587	Uses ~ when the saved file is under the user's home directory; otherwise
588	returns the absolute path.
589	"""
590	from pathlib import Path as _Path
591	path = _Path(save_dir).expanduser().resolve()
592	slug = slugify(topic)
593	extension = "json" if emit == "json" else "html" if emit == "html" else "md"
594	raw_label = "raw-html" if emit == "html" else "raw"
595	safe_suffix = sanitize_suffix(suffix)
596	suffix_part = f"-{safe_suffix}" if safe_suffix else ""
597	raw = path / f"{slug}-{raw_label}{suffix_part}.{extension}"
598	try:
599	home = _Path.home().resolve()
600	relative = raw.relative_to(home)
601	return f"~/{relative.as_posix()}"
602	except ValueError:
603	return raw.as_posix()
604
605
606	def compute_output_path_display(output_file: str) -> str:
607	"""Compute the user-friendly explicit output path shown in render footers."""
608	raw = Path(output_file).expanduser().resolve()
609	try:
610	home = Path.home().resolve()
611	relative = raw.relative_to(home)
612	return f"~/{relative.as_posix()}"
613	except ValueError:
614	return raw.as_posix()
615
616
617	def read_synthesis_file(path: str) -> str:
618	try:
619	return Path(path).expanduser().read_text(encoding="utf-8")
620	except OSError as exc:
621	sys.stderr.write(f"[last30days] Cannot read --synthesis-file: {exc}\n")
622	raise SystemExit(2)
623
624
625	def _scoped_store_db(args: argparse.Namespace) -> Path \| None:
626	"""Scoped runs write findings inside the save dir, matching scoped reads."""
627	save_dir = getattr(args, "save_dir", None)
628	if save_dir:
629	return Path(save_dir).expanduser().resolve() / "research.db"
630	return None
631
632
633	def persist_report(report: schema.Report, store_db: Path \| None = None) -> dict[str, int]:
634	import store
635
636	private_corpus = _report_has_private_corpus(report)
637	with store.scoped_db(store_db):
638	if private_corpus:
639	store.ensure_private_db_files()
640	store.init_db()
641	if private_corpus:
642	store.ensure_private_db_files()
643	topic_row = store.add_topic(report.topic)
644	topic_id = topic_row["id"]
645	source_mode = ",".join(sorted(report.items_by_source)) or "v3"
646	run_id = store.record_run(topic_id, source_mode=source_mode, status="running")
647	try:
648	findings = store.findings_from_report(report)
649	if private_corpus:
650	store.ensure_private_db_files()
651	counts = store.store_findings(run_id, topic_id, findings)
652	store.update_run(
653	run_id,
654	status="completed",
655	findings_new=counts["new"],
656	findings_updated=counts["updated"],
657	)
658	return counts
659	except Exception as exc:
660	store.update_run(run_id, status="failed", error_message=str(exc)[:500])
661	raise
662	finally:
663	if private_corpus:
664	store.ensure_private_db_files()
665
666
667	def build_parser() -> argparse.ArgumentParser:
668	parser = argparse.ArgumentParser(
669	description="Research a topic across live social, market, and grounded web sources.",
670	allow_abbrev=False,
671	)
672	parser.add_argument("topic", nargs="*", help="Research topic")
673	parser.add_argument("--emit", default="compact", choices=["compact", "json", "context", "md", "html", "brief"])
674	parser.add_argument(
675	"--register",
676	choices=registers.REGISTER_NAMES,
677	default=None,
678	help="Audience synthesis preset for the standard brief (default, exec, dev, creator, eli5)",
679	)
680	parser.add_argument(
681	"--json-profile",
682	default="agent",
683	choices=["agent", "raw"],
684	help="JSON export profile for --emit=json (default: agent)",
685	)
686	parser.add_argument("--search", help="Comma-separated source list")
687	parser.add_argument("--quick", action="store_true", help="Lower-latency retrieval profile")
688	parser.add_argument("--deep", action="store_true", help="Higher-recall retrieval profile")
689	freshness_group = parser.add_mutually_exclusive_group()
690	freshness_group.add_argument(
691	"--verify-freshness",
692	action="store_true",
693	default=None,
694	help="Re-check source-grounded claims after research, or verify the cached report when no topic is supplied",
695	)
696	freshness_group.add_argument(
697	"--no-verify-freshness",
698	dest="verify_freshness",
699	action="store_false",
700	help="Disable freshness verification configured by LAST30DAYS_VERIFY_FRESHNESS",
701	)
702	parser.add_argument(
703	"--drill",
704	metavar="TARGET",
705	help="Deep follow-up on a cluster from the fresh last-report.json cache",
706	)
707	parser.add_argument(
708	"--discover",
709	metavar="DOMAIN",
710	nargs="?",
711	const="",
712	default=None,
713	help=(
714	"Sweep river listings and rank the topics accelerating in a domain; "
715	"each survivor gets a full research pass. Bare --discover (no domain) "
716	"runs global trending across every feed's hot list"
717	),
718	)
719	parser.add_argument(
720	"--discover-shallow",
721	action="store_true",
722	help=(
723	"Skip the per-topic research pass during --discover: rank on listing "
724	"evidence only (faster, thinner; the confidence floor still applies)"
725	),
726	)
727	parser.add_argument(
728	"--nominate-only",
729	action="store_true",
730	help=(
731	"Leg 1 of the host-judged discovery protocol: sweep, write the "
732	"nominations bundle for host judgment, and stop (no judging, no "
733	"enrichment). Requires --discover"
734	),
735	)
736	parser.add_argument(
737	"--judgments",
738	metavar="PATH",
739	help=(
740	"Leg 2 of the discovery protocol: resume from the nominations "
741	"bundle, applying the host judgments file at PATH. Requires "
742	"--discover"
743	),
744	)
745	parser.add_argument(
746	"--finalize",
747	action="store_true",
748	help=(
749	"Leg 3 of the discovery protocol: apply host angles, render the "
750	"final discovery brief, and record the topic queue. Requires "
751	"--discover"
752	),
753	)
754	parser.add_argument(
755	"--angles",
756	metavar="PATH",
757	help=(
758	"Optional host angles file for --discover --finalize (omitting it "
759	"ships the brief without angle lines)"
760	),
761	)
762	parser.add_argument("--debug", action="store_true", help="Enable HTTP debug logging")
763	parser.add_argument("--mock", action="store_true", help="Use mock retrieval fixtures")
764	parser.add_argument(
765	"--record-fixtures",
766	metavar="DIR",
767	help=argparse.SUPPRESS,
768	)
769	parser.add_argument("--diagnose", action="store_true", help="Print provider and source availability")
770	parser.add_argument("--preflight", action="store_true",
771	help="Print a safe human-readable permission preflight")
772	parser.add_argument("--welcome", action="store_true",
773	help="Print the first-run welcome text (engine-owned; relay verbatim)")
774	parser.add_argument("--preflight-report-on-save-dir", help=argparse.SUPPRESS)
775	parser.add_argument("--no-browser-cookies", action="store_true",
776	help="Disable browser-cookie extraction even when FROM_BROWSER is configured")
777	parser.add_argument("--save-dir", help="Optional directory for saving the rendered output")
778	parser.add_argument(
779	"--corpus",
780	action="append",
781	default=[],
782	metavar="DIR",
783	help="Add a local .md/.txt/.pdf directory as a private ranked source (repeatable)",
784	)
785	parser.add_argument(
786	"--corpus-all-time",
787	action="store_true",
788	help="Include matching corpus files older than the research window",
789	)
790	parser.add_argument("--output", help="Optional exact file path for saving the rendered output")
791	parser.add_argument("--synthesis-file", help="Markdown synthesis to embed in --emit=html output")
792	parser.add_argument("--publish-html", action="store_true",
793	help="Publish --emit=html output to ht-ml.app (explicit opt-in; public by default)")
794	parser.add_argument("--publish", action="store_true",
795	help="With 'library feed', publish the HTML index and briefs (explicit opt-in; public by default); feed.xml remains local")
796	parser.add_argument("--publish-password",
797	help="Optional shared password for --publish-html or 'library feed --publish'; prefer LAST30DAYS_PUBLISH_PASSWORD to avoid exposing secrets in process lists")
798	parser.add_argument("--store", action="store_true", help="Persist ranked findings to the SQLite research store")
799	parser.add_argument("--x-handle", help="X handle for targeted supplemental search")
800	parser.add_argument("--x-related", help="Comma-separated related X handles (searched with lower weight)")
801	parser.add_argument(
802	"--x-posts",
803	dest="x_posts",
804	metavar="PATH",
805	help=(
806	"Path to a last30days-x-posts/1 JSON envelope of posts the hosting "
807	"model fetched through its X connector; replaces the engine's X "
808	"fetch for this run. A file path only (never inline JSON); on a "
809	"comparison run use the per-entity x_posts field of --competitors-plan."
810	),
811	)
812	parser.add_argument("--web-backend", default="auto",
813	choices=["auto", "brave", "exa", "serper", "parallel", "parallel-mcp", "keyless", "none"],
814	help="Web search backend (default: auto; parallel-mcp explicitly opts into the "
815	"anonymous hosted MCP; keyless forces the zero-key floor)")
816	parser.add_argument("--deep-research", action="store_true",
817	help="Use at most one Perplexity Deep Research run. Direct PERPLEXITY_API_KEY uses the Agent API background path; OPENROUTER_API_KEY keeps the synchronous Sonar fallback; cannot be combined with competitor or vs-mode.")
818	parser.add_argument("--hiring-signals", action="store_true",
819	help="Analyze public jobs/careers postings as evidence-backed company focus signals.")
820	parser.add_argument("--plan", help="JSON query plan (skips internal LLM planner). Can be a JSON string or a file path.")
821	parser.add_argument("--save-suffix", help="Suffix for saved output filename (e.g., 'gemini' → kanye-west-raw-gemini.md)")
822	parser.add_argument("--subreddits", help="Comma-separated broad/category subreddit names to search (e.g., SaaS,Entrepreneur)")
823	parser.add_argument("--dedicated-subreddits", help="Comma-separated entity-home subreddit names (e.g., Kanye,WestSubEver). Pulled in full (top+hot+new) and exempt from the relevance floor since the whole sub is the topic.")
824	parser.add_argument("--tiktok-hashtags", help="Comma-separated TikTok hashtags without # (e.g., tella,screenrecording)")
825	parser.add_argument("--tiktok-creators", help="Comma-separated TikTok creator handles (e.g., TellaHQ,taborplace)")
826	parser.add_argument("--ig-creators", help="Comma-separated Instagram creator handles (e.g., tella.tv,laborstories)")
827	parser.add_argument(
828	"--days",
829	"--lookback-days",
830	dest="lookback_days",
831	type=int,
832	default=None,
833	help="Number of days to look back for research (default: 30, watchlist uses 90)",
834	)
835	parser.add_argument(
836	"--as-of",
837	dest="as_of_date",
838	type=parse_as_of_date_arg,
839	help=(
840	"End date for the lookback window in YYYY-MM-DD format. "
841	"When set, --days looks back from this date instead of today."
842	),
843	)
844	parser.add_argument("--max-results", dest="max_results", type=int,
845	help="Override the final ranked-pool cap (pool_limit/rerank_limit) from the depth profile. "
846	"Use for high-volume topics where the default (deep=60) under-covers. See issue #716.")
847	parser.add_argument("--max-per-source", dest="max_per_source", type=int,
848	help="Override the per-stream cap (per_stream_limit) applied to each (source, subquery) before "
849	"pooling. Raising it increases unique-item yield when one source has many relevant items. "
850	"See issue #716.")
851	parser.add_argument("--max-source-fetches", dest="max_source_fetches", type=int,
852	help="Override the per-source fetch cap (MAX_SOURCE_FETCHES, default x=2) that limits how many "
853	"subqueries actually fetch a capped source. Raise it so every X subquery in a multi-angle "
854	"--plan runs instead of just the first two. See issue #716.")
855	parser.add_argument("--auto-resolve", action="store_true",
856	help="Use web search to discover subreddits/handles before planning (for platforms without WebSearch)")
857	parser.add_argument("--github-user", help="GitHub username for person-mode search (e.g., steipete)")
858	parser.add_argument("--github-repo", help="Comma-separated owner/repo for project-mode search (e.g., openclaw/openclaw,paperclipai/paperclip)")
859	parser.add_argument(
860	"--trustpilot-domain",
861	help=(
862	"Trustpilot review-page domain for the topic (e.g., www.thriftbooks.com). "
863	"Used verbatim, bypasses the brand-shape gate, and auto-activates the "
864	"opt-in Trustpilot source for this run (unless EXCLUDE_SOURCES=trustpilot). "
865	"Find the domain with `trustpilot-pp-cli search '<name>'`."
866	),
867	)
868	parser.add_argument(
869	"--amazon-query",
870	help=(
871	"Product keyword the amazon source searches, when that source is active. "
872	"Defaults to the topic. Supply it whenever the topic is not the product: "
873	"a person topic searches their company's product line "
874	"(--amazon-query='June Oven'), and a brand searches brand-plus-category "
875	"(--amazon-query='Weber grill', not 'Weber' -- a bare brand keyword lands "
876	"on an ad-heavy page that can miss the brand's own bestsellers). "
877	"Requires the brightdata CLI on PATH and logged in."
878	),
879	)
880	parser.add_argument(
881	"--meta-ads-page",
882	help=(
883	"Meta Ad Library page id for the topic's advertiser, when the meta_ads "
884	"source is active. Skips name-based page resolution and its discovery "
885	"credit. Accepts a bare numeric page id (e.g. 123456789012345) or an Ad "
886	"Library URL carrying view_all_page_id. A facebook.com vanity URL is not "
887	"a page id and is rejected. Use it when a brand advertises under product "
888	"names, or when resolution picked the wrong company."
889	),
890	)
891	parser.add_argument(
892	"--telegram-sources",
893	help=(
894	"Comma-separated list of public Telegram channel handles or t.me URLs. "
895	"Auto-activates the opt-in Telegram source for this run. "
896	"Accepts: bare handle (aipost), @handle (@aipost), "
897	"t.me URL (https://t.me/aipost), or preview URL (https://t.me/s/aipost). "
898	"Rejects joinchat links and numeric -100 supergroup IDs."
899	),
900	)
901	parser.add_argument(
902	"--competitors",
903	nargs="?",
904	const=2,
905	type=int,
906	default=None,
907	metavar="N",
908	help="Auto-discover N competitor entities and fan out last30days across all of them as a comparison (default N=2 → 3-way: original + 2 peers; range 1..6). Use --competitors-list to override discovery.",
909	)
910	parser.add_argument(
911	"--competitors-list",
912	dest="competitors_list",
913	help="Comma-separated competitor entities to skip discovery (e.g., 'Anthropic,xAI,Google Gemini'). Implies --competitors.",
914	)
915	parser.add_argument(
916	"--polymarket-keywords",
917	dest="polymarket_keywords",
918	help=(
919	"Comma-separated keywords that Polymarket market titles must match "
920	"to be included. Use for ambiguous single-token topics like 'Warriors' "
921	"(nba,gsw,golden-state) to filter out Glasgow Warriors rugby, Honor "
922	"of Kings Rogue Warriors, etc. When omitted, Polymarket returns all "
923	"matching markets — so expect cross-entity noise on generic topics."
924	),
925	)
926	parser.add_argument(
927	"--competitors-plan",
928	dest="competitors_plan",
929	help=(
930	"JSON mapping of per-entity Step 0.55 targeting for competitor / vs-mode "
931	"sub-runs. Schema: {entity_name: {x_handle?, x_related?, subreddits?, "
932	"github_user?, github_repos?, context?}}. Accepts inline JSON or a file "
933	"path. Implies --competitors. Preferred over --competitors-list when the "
934	"hosting model has already resolved per-entity handles and subs."
935	),
936	)
937	return parser
938
939
940	def parse_competitors_plan(raw: str \| None) -> dict[str, dict]:
941	"""Parse a --competitors-plan argument into a {entity_name_lower: plan_entry} dict.
942
943	Accepts inline JSON or a file path (matches --plan). Returns {} on None/empty.
944	Validation: top-level must be a dict; each value must be a dict. Unknown fields
945	in entry values log a warning but do not abort. Invalid JSON or non-dict shape
946	raises SystemExit(2) with a clear stderr message.
947	"""
948	if not raw:
949	return {}
950	plan_str = raw
951	if os.path.isfile(plan_str):
952	try:
953	with open(plan_str, encoding="utf-8") as f:
954	plan_str = f.read()
955	except (OSError, UnicodeDecodeError) as exc:
956	sys.stderr.write(f"[CompetitorsPlan] Cannot read plan file: {exc}\n")
957	raise SystemExit(2)
958	try:
959	parsed = json.loads(plan_str)
960	except json.JSONDecodeError as exc:
961	sys.stderr.write(f"[CompetitorsPlan] Invalid JSON: {exc}\n")
962	raise SystemExit(2)
963	if not isinstance(parsed, dict):
964	sys.stderr.write(
965	f"[CompetitorsPlan] Top-level must be a dict of "
966	f"{{entity: {{targeting}}}}, got {type(parsed).__name__}\n"
967	)
968	raise SystemExit(2)
969	known_fields = {
970	"x_handle", "x_related", "subreddits",
971	"github_user", "github_repos", "trustpilot_domain", "context",
972	"x_posts",
973	}
974	normalized: dict[str, dict] = {}
975	for entity, entry in parsed.items():
976	if not isinstance(entry, dict):
977	sys.stderr.write(
978	f"[CompetitorsPlan] Entry for {entity!r} must be a dict, "
979	f"got {type(entry).__name__}; skipping.\n"
980	)
981	continue
982	unknown = set(entry.keys()) - known_fields
983	if unknown:
984	sys.stderr.write(
985	f"[CompetitorsPlan] Unknown fields in {entity!r}: "
986	f"{sorted(unknown)}; ignoring.\n"
987	)
988	normalized[entity.strip().lower()] = {
989	**{k: v for k, v in entry.items() if k in known_fields},
990	"_name": entity.strip(),
991	}
992	return normalized
993
994
995	def subrun_kwargs_for(
996	entity: str,
997	plan_entry: dict,
998	*,
999	resolved: dict,
1000	) -> dict:
1001	"""Build an explicit per-entity kwargs dict for pipeline.run().
1002
1003	Plan values win over auto_resolve values. Returns keys for all per-entity
1004	targeting flags so callers never fall through to closure defaults.
1005
1006	This helper is the single source of truth for sub-run kwargs — main-topic
1007	flags can only leak if a caller bypasses it.
1008	"""
1009	def _choose(plan_key: str, resolved_key: str \| None = None):
1010	if plan_key in plan_entry and plan_entry[plan_key]:
1011	return plan_entry[plan_key]
1012	if resolved_key is not None and resolved.get(resolved_key):
1013	return resolved[resolved_key]
1014	return None
1015
1016	x_handle = _choose("x_handle", "x_handle")
1017	if isinstance(x_handle, str):
1018	x_handle = x_handle.lstrip("@") or None
1019
1020	subreddits = _choose("subreddits", "subreddits")
1021	if isinstance(subreddits, list):
1022	subreddits = [s.strip().removeprefix("r/") for s in subreddits if s.strip()] or None
1023
1024	x_related = plan_entry.get("x_related")
1025	if isinstance(x_related, list):
1026	x_related = [h.strip().lstrip("@") for h in x_related if h.strip()] or None
1027	else:
1028	x_related = None
1029
1030	github_user = _choose("github_user", "github_user")
1031	if isinstance(github_user, str):
1032	github_user = github_user.lstrip("@").lower() or None
1033
1034	github_repos = _choose("github_repos", "github_repos")
1035	if isinstance(github_repos, list):
1036	github_repos = [r.strip() for r in github_repos if r.strip() and "/" in r.strip()] or None
1037
1038	trustpilot_domain = _choose("trustpilot_domain", "trustpilot_domain")
1039	if isinstance(trustpilot_domain, str):
1040	trustpilot_domain = trustpilot_domain.strip() or None
1041	# Provenance: a plan-supplied domain is user-set (verbatim-final); one that
1042	# only came from auto_resolve is a hint that retries via search on a miss.
1043	trustpilot_domain_is_hint = bool(
1044	trustpilot_domain and not plan_entry.get("trustpilot_domain")
1045	)
1046
1047	context = plan_entry.get("context") or resolved.get("context") or ""
1048
1049	return {
1050	"x_handle": x_handle,
1051	"x_related": x_related,
1052	"subreddits": subreddits,
1053	"github_user": github_user,
1054	"github_repos": github_repos,
1055	"trustpilot_domain": trustpilot_domain,
1056	"_trustpilot_domain_is_hint": trustpilot_domain_is_hint,
1057	"_context": context,
1058	}
1059
1060
1061	COMPETITORS_MIN = competitors_mod.COMPETITORS_MIN
1062	COMPETITORS_MAX = competitors_mod.COMPETITORS_MAX
1063	COMPETITORS_DEFAULT = competitors_mod.COMPETITORS_DEFAULT
1064
1065
1066	def truncate_comparison_entities(entities: list[str], *, warn: bool = True) -> list[str]:
1067	"""Cap a vs-entity list at COMPARISON_ENTITY_MAX; optionally warn on stderr."""
1068	ceiling = competitors_mod.COMPARISON_ENTITY_MAX
1069	if len(entities) <= ceiling:
1070	return list(entities)
1071	kept = entities[:ceiling]
1072	dropped = entities[ceiling:]
1073	if warn:
1074	sys.stderr.write(
1075	f"[Competitors] vs-topic has {len(entities)} entities; "
1076	f"using first {ceiling}, dropped: {', '.join(dropped)}\n"
1077	)
1078	return kept
1079
1080
1081	def apply_vs_competitor_routing(
1082	topic: str,
1083	*,
1084	competitors_flag: int \| None,
1085	comp_enabled: bool,
1086	comp_count: int,
1087	comp_explicit: list[str],
1088	comp_plan: dict[str, dict] \| None = None,
1089	) -> tuple[str, bool, int, list[str]]:
1090	"""Apply vs-string / plan routing on top of resolve_competitors_args.
1091
1092	Precedence for who runs:
1093	1. ``--competitors-list`` (explicit peers; topic unchanged)
1094	2. Pure discover-N (``--competitors`` without list or plan) — topic
1095	unchanged, even if it contains ``vs``
1096	3. vs-string split (first entity becomes main topic) — used for bare
1097	vs-topics and vs-topic + ``--competitors-plan``
1098	4. ``--competitors-plan`` keys as peers when there is no vs-string
1099	(including when ``--competitors N`` is also set)
1100	"""
1101	from lib import planner as _planner
1102
1103	if comp_explicit:
1104	return topic, True, len(comp_explicit), list(comp_explicit)
1105
1106	# Preserve discover-N semantics: numeric flag without plan/list must not
1107	# rewrite a vs-string into named peers.
1108	if competitors_flag is not None and not comp_plan:
1109	return topic, True, comp_count, []
1110
1111	vs_entities = truncate_comparison_entities(
1112	_planner._comparison_entities(topic, uncapped=True),
1113	warn=True,
1114	)
1115	if len(vs_entities) >= 2:
1116	main, peers = vs_entities[0], vs_entities[1:]
1117	sys.stderr.write(
1118	f"[Competitors] vs-mode: routing to N-pass fanout: "
1119	f"{main} vs {' vs '.join(peers)}\n"
1120	)
1121	return main, True, len(peers), peers
1122
1123	if comp_plan:
1124	plan_peers = [
1125	(entry.get("_name") or key)
1126	for key, entry in comp_plan.items()
1127	]
1128	plan_peers = [name for name in plan_peers if name]
1129	if len(plan_peers) > COMPETITORS_MAX:
1130	sys.stderr.write(
1131	f"[Competitors] --competitors-plan has {len(plan_peers)} entries, "
1132	f"clamping to {COMPETITORS_MAX}.\n"
1133	)
1134	plan_peers = plan_peers[:COMPETITORS_MAX]
1135	return topic, True, len(plan_peers), plan_peers
1136
1137	return topic, comp_enabled, comp_count, comp_explicit
1138
1139
1140	def resolve_competitors_args(args: argparse.Namespace) -> tuple[bool, int, list[str]]:
1141	"""Normalize competitors flags into (enabled, count, explicit_list).
1142
1143	- (False, 0, []) when neither flag, list, nor plan is set.
1144	- An explicit ``--competitors-list`` always wins; count is derived from list length.
1145	- ``--competitors-plan`` alone enables mode with an empty peer list; vs-routing
1146	fills peers from the vs-string or plan keys.
1147	- A numeric count outside [1, 6] is clamped with a stderr warning.
1148	- count <= 0 (explicit) raises SystemExit(2).
1149	"""
1150	explicit_list: list[str] = []
1151	list_flag_provided = args.competitors_list is not None
1152	if list_flag_provided:
1153	explicit_list = [
1154	entity.strip()
1155	for entity in args.competitors_list.split(",")
1156	if entity.strip()
1157	]
1158	if not explicit_list:
1159	sys.stderr.write("[Competitors] --competitors-list is empty.\n")
1160	raise SystemExit(2)
1161
1162	competitors_flag = args.competitors
1163	list_present = bool(explicit_list)
1164	flag_present = competitors_flag is not None
1165	plan_present = bool(getattr(args, "competitors_plan", None))
1166
1167	if not list_present and not flag_present and not plan_present:
1168	return False, 0, []
1169
1170	if list_present:
1171	count = len(explicit_list)
1172	if flag_present and competitors_flag != count:
1173	sys.stderr.write(
1174	f"[Competitors] --competitors={competitors_flag} ignored; using "
1175	f"{count} entries from --competitors-list.\n"
1176	)
1177	if count > COMPETITORS_MAX:
1178	sys.stderr.write(
1179	f"[Competitors] --competitors-list has {count} entries, clamping to {COMPETITORS_MAX}.\n"
1180	)
1181	explicit_list = explicit_list[:COMPETITORS_MAX]
1182	count = COMPETITORS_MAX
1183	return True, count, explicit_list
1184
1185	if flag_present:
1186	count = competitors_flag
1187	if count < COMPETITORS_MIN:
1188	sys.stderr.write(
1189	f"[Competitors] --competitors must be >= {COMPETITORS_MIN} (got {count}).\n"
1190	)
1191	raise SystemExit(2)
1192	if count > COMPETITORS_MAX:
1193	sys.stderr.write(
1194	f"[Competitors] --competitors={count} exceeds max {COMPETITORS_MAX}; clamping.\n"
1195	)
1196	count = COMPETITORS_MAX
1197	return True, count, []
1198
1199	# plan_present alone: enable; peers filled by apply_vs_competitor_routing.
1200	return True, 0, []
1201
1202
1203	def _missing_sources_for_promo(diag: dict[str, object]) -> str \| None:
1204	available = set(diag.get("available_sources") or [])
1205	missing = []
1206	if "reddit" not in available:
1207	missing.append("reddit")
1208	# X is optional. A successful run without X must reach the research output
1209	# without an authentication or browser-cookie promo in front of it.
1210	# The web promo nudges toward a paid backend for higher-quality web search.
1211	# Grounding is now available keyless on non-native hosts, so key the promo on
1212	# the absence of a paid backend, not on grounding availability. Suppress it
1213	# entirely on native-search hosts, where the model's own search is better and
1214	# setting a paid engine key would be the wrong advice.
1215	if not diag.get("native_web_backend") and not diag.get("native_search"):
1216	missing.append("web")
1217	if not missing:
1218	return None
1219	return missing[0]
1220
1221
1222	def _optional_x_omission_text(
1223	diag: dict[str, object],
1224	requested_sources: list[str] \| None,
1225	) -> str \| None:
1226	"""Return a non-blocking post-result note for a default run without X.
1227
1228	Explicit ``--search`` runs already define their intended source boundary,
1229	so they do not need an omission note. Doctor/diagnose remains the place for
1230	X setup or repair instructions.
1231	"""
1232	if requested_sources is not None:
1233	return None
1234	available = set(diag.get("available_sources") or [])
1235	if "x" in available:
1236	return None
1237	return (
1238	"Optional source omitted: X/Twitter was not enabled; research "
1239	"continued with the available sources."
1240	)
1241
1242
1243	def _show_runtime_ui(
1244	report: schema.Report,
1245	progress: ui.ProgressDisplay,
1246	diag: dict[str, object],
1247	suppress_web_promo: bool = False,
1248	) -> None:
1249	counts = {source: len(items) for source, items in report.items_by_source.items()}
1250	display_sources = list(
1251	dict.fromkeys(
1252	[
1253	*report.query_plan.source_weights.keys(),
1254	*report.items_by_source.keys(),
1255	*report.errors_by_source.keys(),
1256	]
1257	)
1258	)
1259	progress.end_processing()
1260	progress.show_complete(
1261	source_counts=counts,
1262	display_sources=display_sources,
1263	)
1264	promo = _missing_sources_for_promo(diag)
1265	# The `web` promo nudges users to set BRAVE_API_KEY / SERPER_API_KEY, which
1266	# is wrong advice when a hosting reasoning model (Claude Code, Codex,
1267	# Hermes, Gemini) is driving — those already have WebSearch and can
1268	# pre-resolve Step 0.55 themselves. Suppress the web promo when a hosting
1269	# model signal is present (--plan or --competitors-plan was passed).
1270	if promo:
1271	if suppress_web_promo and promo == "web":
1272	return
1273	if suppress_web_promo and promo == "both":
1274	# "both" means reddit + web both missing; still nudge reddit but
1275	# skip the web line. show_promo has a per-source variant.
1276	progress.show_promo("reddit", diag=diag)
1277	return
1278	progress.show_promo(promo, diag=diag)
1279
1280
1281	REPORT_CACHE_VERSION = "last30days-report-cache/v1"
1282	DEFAULT_REPORT_CACHE_TTL_SECONDS = 3600
1283
1284
1285	def _last_report_cache_path() -> Path \| None:
1286	if env.CONFIG_DIR is None:
1287	return None
1288	return env.CONFIG_DIR / "last-report.json"
1289
1290
1291	def _report_cache_ttl_seconds(config: dict[str, object]) -> int:
1292	raw = os.environ.get("LAST30DAYS_REPORT_CACHE_TTL_SECONDS")
1293	if raw is None:
1294	raw = config.get("LAST30DAYS_REPORT_CACHE_TTL_SECONDS")
1295	if raw is None or raw == "":
1296	return DEFAULT_REPORT_CACHE_TTL_SECONDS
1297	try:
1298	return max(0, int(raw))
1299	except (TypeError, ValueError):
1300	return DEFAULT_REPORT_CACHE_TTL_SECONDS
1301
1302
1303	def _is_report_cache_fresh(timestamp: object, ttl_seconds: int) -> bool:
1304	return env.is_timestamp_fresh(timestamp, ttl_seconds)
1305
1306
1307	def _write_last_run(
1308	topic: str,
1309	report: "schema.Report",
1310	entity_reports: list[tuple[str, schema.Report]] \| None = None,
1311	*,
1312	x_envelope_sha256: str \| None = None,
1313	) -> bool:
1314	# ``x_envelope_sha256`` binds the cached report to the --x-posts file it
1315	# was built from; _load_last_report_cache misses on any mismatch.
1316	try:
1317	if env.CONFIG_DIR is None:
1318	return False
1319	target = env.CONFIG_DIR
1320	cached_reports = entity_reports or [(report.topic, report)]
1321	has_private_corpus = any(
1322	cached_report.items_by_source.get("corpus")
1323	for _, cached_report in cached_reports
1324	)
1325	_ensure_output_directory(target, private=has_private_corpus)
1326	counts = {source: len(items) for source, items in report.items_by_source.items()}
1327	payload = {
1328	"topic": topic,
1329	"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
1330	"sources": counts,
1331	"total": sum(counts.values()),
1332	"report_cache": str(target / "last-report.json"),
1333	"comparison": bool(entity_reports),
1334	}
1335	(target / "last-run.json").write_text(json.dumps(payload, indent=2))
1336	cache_payload = {
1337	"schema": REPORT_CACHE_VERSION,
1338	"topic": topic,
1339	"timestamp": payload["timestamp"],
1340	"comparison": bool(entity_reports),
1341	"x_envelope_sha256": x_envelope_sha256 or None,
1342	"reports": [
1343	{"entity": label, "report": schema.to_dict(cached_report)}
1344	for label, cached_report in cached_reports
1345	],
1346	}
1347	report_cache_path = target / "last-report.json"
1348	report_cache_path.write_text(json.dumps(cache_payload, indent=2))
1349	if has_private_corpus:
1350	report_cache_path.chmod(0o600)
1351	return True
1352	except Exception as exc:
1353	# Never fatal, but never silent either (#787's lesson): callers that
1354	# promise cache state (drill chaining) branch on the return value.
1355	sys.stderr.write(f"[last30days] warning: could not write run cache: {exc}\n")
1356	return False
1357
1358
1359	def _load_last_report_cache(
1360	topic: str \| None,
1361	ttl_seconds: int = DEFAULT_REPORT_CACHE_TTL_SECONDS,
1362	*,
1363	x_envelope_sha256: str \| None = None,
1364	) -> tuple[schema.Report, list[tuple[str, schema.Report]] \| None, Path] \| None:
1365	cache_path = _last_report_cache_path()
1366	if cache_path is None or not cache_path.exists():
1367	return None
1368	try:
1369	payload = json.loads(cache_path.read_text(encoding="utf-8"))
1370	if not isinstance(payload, dict):
1371	raise TypeError("report cache payload must be a JSON object")
1372	if payload.get("schema") != REPORT_CACHE_VERSION:
1373	return None
1374	if not _is_report_cache_fresh(payload.get("timestamp"), ttl_seconds):
1375	return None
1376	# A report built from a --x-posts envelope is only reusable with the
1377	# same envelope content; a digest on either side that does not match
1378	# the other is a miss.
1379	cached_digest = payload.get("x_envelope_sha256") or None
1380	if (cached_digest or x_envelope_sha256) and cached_digest != x_envelope_sha256:
1381	return None
1382	cached_topic = str(payload.get("topic") or "").strip().lower()
1383	if topic is not None and cached_topic != topic.strip().lower():
1384	return None
1385	reports_payload = payload.get("reports") or []
1386	if not reports_payload:
1387	return None
1388	entity_reports = [
1389	(str(item.get("entity") or ""), schema.report_from_dict(item["report"]))
1390	for item in reports_payload
1391	if isinstance(item, dict) and isinstance(item.get("report"), dict)
1392	]
1393	if not entity_reports:
1394	return None
1395	if payload.get("comparison"):
1396	if len(entity_reports) < 2:
1397	return None
1398	if len(entity_reports) != len(reports_payload):
1399	return None
1400	return entity_reports[0][1], entity_reports, cache_path
1401	return entity_reports[0][1], None, cache_path
1402	except (OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc:
1403	sys.stderr.write(
1404	f"[last30days] Could not read report cache {cache_path}: "
1405	f"{type(exc).__name__}: {exc}\n"
1406	)
1407	return None
1408
1409
1410	def _config_truthy(value: object) -> bool:
1411	return str(value or "").strip().lower() in {"1", "true", "yes", "on"}
1412
1413
1414	def _freshness_enabled(args: argparse.Namespace, config: dict[str, object]) -> bool:
1415	if args.verify_freshness is not None:
1416	return bool(args.verify_freshness)
1417	return _config_truthy(config.get("LAST30DAYS_VERIFY_FRESHNESS"))
1418
1419
1420	def _update_cached_freshness(
1421	cache_path: Path,
1422	report: schema.Report,
1423	entity_reports: list[tuple[str, schema.Report]] \| None,
1424	) -> bool:
1425	"""Rewrite cached report bodies without extending the research-cache TTL."""
1426	try:
1427	payload = json.loads(cache_path.read_text(encoding="utf-8"))
1428	if not isinstance(payload, dict) or payload.get("schema") != REPORT_CACHE_VERSION:
1429	return False
1430	existing = payload.get("reports") or []
1431	if entity_reports:
1432	cached_reports = entity_reports
1433	else:
1434	label = (
1435	str(existing[0].get("entity") or report.topic)
1436	if existing and isinstance(existing[0], dict)
1437	else report.topic
1438	)
1439	cached_reports = [(label, report)]
1440	payload["reports"] = [
1441	{"entity": label, "report": schema.to_dict(cached_report)}
1442	for label, cached_report in cached_reports
1443	]
1444	cache_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
1445	return True
1446	except (OSError, json.JSONDecodeError, TypeError, ValueError) as exc:
1447	sys.stderr.write(
1448	f"[last30days] warning: could not update freshness cache: {exc}\n"
1449	)
1450	return False
1451
1452
1453	def _verify_report_set(
1454	report: schema.Report,
1455	entity_reports: list[tuple[str, schema.Report]] \| None,
1456	*,
1457	allow_network: bool,
1458	) -> None:
1459	reports = [item for _, item in entity_reports] if entity_reports else [report]
1460	for current_report in reports:
1461	freshness.verify_report(current_report, allow_network=allow_network)
1462	if not any(current_report.freshness_verdicts for current_report in reports):
1463	# An empty verdict list is a legitimate outcome, but a silent one has
1464	# already misled operators once; say why there is nothing to show.
1465	sys.stderr.write(
1466	"[last30days] Freshness verification found no re-checkable claims"
1467	" in this report; the verdict list is empty.\n"
1468	)
1469
1470
1471	def _run_cached_freshness(
1472	args: argparse.Namespace,
1473	config: dict[str, object],
1474	) -> int:
1475	cached = _load_last_report_cache(
1476	None,
1477	ttl_seconds=_report_cache_ttl_seconds(config),
1478	)
1479	if cached is None:
1480	sys.stderr.write("[last30days] No fresh cached report; run a research pass first.\n")
1481	return 2
1482	report, entity_reports, cache_path = cached
1483	_verify_report_set(report, entity_reports, allow_network=not args.mock)
1484	if _update_cached_freshness(cache_path, report, entity_reports):
1485	sys.stderr.write(f"[last30days] Updated freshness verdicts in {cache_path}\n")
1486	else:
1487	sys.stderr.write("[last30days] warning: freshness cache update failed\n")
1488	return _render_save_and_print(args, report, entity_reports, None, config)
1489
1490
1491	def _drill_config(config: dict[str, object], sources: list[str]) -> dict[str, object]:
1492	"""Enable configured comment enrichments for a deep follow-up."""
1493	drill_config = dict(config)
1494	include = {
1495	value.strip().lower()
1496	for value in str(config.get("INCLUDE_SOURCES") or "").split(",")
1497	if value.strip()
1498	}
1499	comment_flags = {
1500	"youtube": "youtube_comments",
1501	"tiktok": "tiktok_comments",
1502	"instagram": "instagram_comments",
1503	}
1504	include.update(comment_flags[source] for source in sources if source in comment_flags)
1505	if include:
1506	drill_config["INCLUDE_SOURCES"] = ",".join(sorted(include))
1507	drill_config["_drill_mode"] = True
1508	return drill_config
1509
1510
1511	def _run_drill(
1512	args: argparse.Namespace,
1513	config: dict[str, object],
1514	) -> int:
1515	from lib import planner
1516
1517	cached = _load_last_report_cache(
1518	None,
1519	ttl_seconds=_report_cache_ttl_seconds(config),
1520	)
1521	if cached is None:
1522	sys.stderr.write(
1523	"[last30days] No fresh cached report; run a research pass first.\n"
1524	)
1525	return 2
1526	report, entity_reports, cache_path = cached
1527	if entity_reports:
1528	sys.stderr.write(
1529	"[last30days] Drill mode needs a single-topic cached report; "
1530	"run a research pass for one entity first.\n"
1531	)
1532	return 2
1533
1534	lookback_days = args.lookback_days
1535	if lookback_days is None:
1536	range_from = datetime.date.fromisoformat(report.range_from)
1537	range_to = datetime.date.fromisoformat(report.range_to)
1538	lookback_days = (range_to - range_from).days
1539	as_of_date = args.as_of_date or report.range_to
1540
1541	try:
1542	matched_clusters = planner.resolve_drill_clusters(report, args.drill)
1543	drill_plan = planner.build_drill_plan(
1544	report,
1545	args.drill,
1546	clusters=matched_clusters,
1547	)
1548	except planner.DrillTargetError as exc:
1549	sys.stderr.write(f"[last30days] {exc}\n")
1550	return 2
1551
1552	sources = list(drill_plan.source_weights)
1553	drill_config = _drill_config(config, sources)
1554	diag = pipeline.diagnose(drill_config, sources, safe=False)
1555	progress = ui.ProgressDisplay(
1556	f"{report.topic} — drill: {args.drill}",
1557	show_banner=True,
1558	)
1559	progress.start_processing()
1560	resolved = report.artifacts.get("resolved") or {}
1561	try:
1562	drill_report = pipeline.run(
1563	# Keep source gating anchored to the cached entity (for example,
1564	# StockTwits needs the original cashtag/finance context). The
1565	# external drill plan below remains cluster-focused.
1566	topic=report.topic,
1567	config=drill_config,
1568	depth="deep",
1569	requested_sources=sources,
1570	mock=args.mock,
1571	x_handle=(
1572	(args.x_handle or resolved.get("x_handle") or None)
1573	if "x" in sources else None
1574	),
1575	x_related=(
1576	[value.strip() for value in args.x_related.split(",") if value.strip()]
1577	if (args.x_related and "x" in sources) else None
1578	),
1579	web_backend=args.web_backend,
1580	external_plan=schema.to_dict(drill_plan),
1581	subreddits=(
1582	([value.strip().removeprefix("r/") for value in args.subreddits.split(",") if value.strip()]
1583	if args.subreddits else list(resolved.get("subreddits") or []) or None)
1584	if "reddit" in sources else None
1585	),
1586	tiktok_hashtags=(
1587	[value.strip().lstrip("#") for value in args.tiktok_hashtags.split(",") if value.strip()]
1588	if args.tiktok_hashtags else None
1589	),
1590	tiktok_creators=(
1591	[value.strip().lstrip("@") for value in args.tiktok_creators.split(",") if value.strip()]
1592	if args.tiktok_creators else None
1593	),
1594	ig_creators=(
1595	[value.strip().lstrip("@") for value in args.ig_creators.split(",") if value.strip()]
1596	if args.ig_creators else None
1597	),
1598	lookback_days=lookback_days,
1599	as_of_date=as_of_date,
1600	github_user=(
1601	(args.github_user or resolved.get("github_user") or None)
1602	if "github" in sources else None
1603	),
1604	github_repos=(
1605	([value.strip() for value in args.github_repo.split(",") if value.strip()]
1606	if args.github_repo else list(resolved.get("github_repos") or []) or None)
1607	if "github" in sources else None
1608	),
1609	trustpilot_domain=(
1610	(args.trustpilot_domain or resolved.get("trustpilot_domain") or None)
1611	if "trustpilot" in sources else None
1612	),
1613	internal_subrun=True,
1614	corpus_dirs=args.corpus,
1615	corpus_all_time=args.corpus_all_time,
1616	)
1617	except Exception:
1618	progress.end_processing()
1619	raise
1620
1621	_show_runtime_ui(drill_report, progress, diag, suppress_web_promo=True)
1622	merged = pipeline.merge_drill_report(
1623	report,
1624	drill_report,
1625	matched_clusters,
1626	target=args.drill,
1627	)
1628	if _freshness_enabled(args, config):
1629	_verify_report_set(merged, None, allow_network=not args.mock)
1630	else:
1631	merged.freshness_verdicts = []
1632	if _write_last_run(report.topic, merged):
1633	sys.stderr.write(f"[last30days] Updated drill cache in {cache_path}\n")
1634	else:
1635	sys.stderr.write(
1636	"[last30days] warning: drill cache update failed; the next drill "
1637	"will see the pre-drill report\n"
1638	)
1639
1640	store_default = str(
1641	os.environ.get("LAST30DAYS_STORE")
1642	or config.get("LAST30DAYS_STORE")
1643	or ""
1644	).lower()
1645	if args.store or store_default in {"1", "true", "yes"}:
1646	counts = persist_report(merged, store_db=_scoped_store_db(args))
1647	sys.stderr.write(
1648	f"[last30days] Stored {counts['new']} new, "
1649	f"{counts['updated']} updated findings\n"
1650	)
1651
1652	synthesis_md = None
1653	if args.synthesis_file:
1654	if args.emit == "html":
1655	synthesis_md = read_synthesis_file(args.synthesis_file)
1656	else:
1657	sys.stderr.write(
1658	"[last30days] Warning: --synthesis-file is only used with "
1659	"--emit=html; ignoring.\n"
1660	)
1661	return _render_save_and_print(args, merged, None, synthesis_md, config)
1662
1663
1664	def _save_discovery_output(
1665	rendered: str,
1666	*,
1667	domain: str,
1668	emit: str,
1669	save_dir: str,
1670	suffix: str = "",
1671	) -> Path:
1672	directory = Path(save_dir).expanduser().resolve()
1673	directory.mkdir(parents=True, exist_ok=True)
1674	extension = "json" if emit == "json" else "md"
1675	safe_suffix = sanitize_suffix(suffix)
1676	suffix_part = f"-{safe_suffix}" if safe_suffix else ""
1677	stem = f"{slugify(domain)}-discover-raw{suffix_part}"
1678	date_str = datetime.datetime.now().strftime("%Y-%m-%d")
1679	candidates = [directory / f"{stem}.{extension}", directory / f"{stem}-{date_str}.{extension}"]
1680	candidates.extend(directory / f"{stem}-{date_str}-{index}.{extension}" for index in range(1, 100))
1681	encoded = rendered.encode("utf-8")
1682	for candidate in candidates:
1683	try:
1684	fd = os.open(candidate, os.O_CREAT \| os.O_EXCL \| os.O_WRONLY, 0o644)
1685	except FileExistsError:
1686	continue
1687	with os.fdopen(fd, "wb") as output:
1688	output.write(encoded)
1689	return candidate
1690	raise RuntimeError("Could not find a unique discovery output filename")
1691
1692
1693	def _pre_run_prior_state(
1694	prior: dict[str, object] \| None, run_ref: str
1695	) -> dict[str, object] \| None:
1696	"""Reconstruct the queue state a topic had BEFORE this run identity
1697	recorded it.
1698
1699	A row whose last_run_ref equals THIS run's run_ref was stamped by this
1700	very run's own earlier attempt (a finalize retry), so its surface_count
1701	already includes this run's surfacing: subtract it and keep the prior's
1702	covered state (covered_at intact) so the retry renders exactly like the
1703	first attempt did. Only when nothing remains after the subtraction AND
1704	the row was never covered is the topic genuinely first-ever (no prior).
1705	"""
1706	if not prior or prior.get("last_run_ref") != run_ref:
1707	return prior
1708	previously = max(0, int(prior["surface_count"]) - 1)
1709	if previously == 0 and prior["status"] != "covered":
1710	return None
1711	adjusted = dict(prior)
1712	adjusted["surface_count"] = previously
1713	return adjusted
1714
1715
1716	def _annotate_and_record_discovery_queue(
1717	report: schema.DiscoveryReport,
1718	args: argparse.Namespace,
1719	config: dict[str, object],
1720	run_ref: str \| None = None,
1721	) -> schema.DiscoveryReport:
1722	"""Stamp queue annotations onto report topics, then record this surfacing.
1723
1724	Order matters: annotations describe the queue state BEFORE this run, so
1725	each topic is matched first and recorded second. The queue is on by
1726	default; the resolved config value LAST30DAYS_DISCOVERY_QUEUE == "off"
1727	(env var or .env, via env.get_config) disables it. Scoped runs
1728	(--save-dir) write the scoped research.db, never the global one. Runs
1729	synchronously after the pipeline returns - this writes disk, so the
1730	abandon-on-timeout daemon-thread pattern is forbidden here.
1731
1732	``run_ref`` overrides the run identity: the finalize leg passes the
1733	pending report's leg-2 run_ref through so a finalize retry records (and
1734	annotates) as the SAME run - store.record_discovery_surfacing skips the
1735	double-count, and rows this very run identity stamped are not "prior"
1736	state, so retries render identically instead of claiming a resurfacing.
1737	"""
1738	queue_setting = str(config.get("LAST30DAYS_DISCOVERY_QUEUE") or "").strip().lower()
1739	if queue_setting == "off" or not report.topics:
1740	return report
1741
1742	import dataclasses
1743
1744	import store
1745
1746	run_ref = run_ref or f"discover:{report.domain or 'trending'}:{report.generated_at}"
1747	as_of = (report.generated_at or "")[:10] or report.range_to
1748	annotated: list[schema.DiscoveryTopic] = []
1749	with store.scoped_db(_scoped_store_db(args)):
1750	store.init_db()
1751	# Phase 1: match EVERY topic before recording ANY. Interleaving
1752	# match+record in one loop lets topic N fuzzy-match a same-anchor
1753	# sibling row this very run recorded seconds earlier, falsely
1754	# annotating a first-ever topic as "surfaced 2nd time".
1755	# A row stamped by THIS run identity is this run's own earlier
1756	# attempt (finalize retry), not prior state: reconstruct the pre-run
1757	# state (count minus this run's own surfacing, covered state kept)
1758	# so retries render identically for topics WITH history too.
1759	priors = [
1760	_pre_run_prior_state(prior, run_ref)
1761	for prior in (
1762	store.match_discovery_topic(topic.name) for topic in report.topics
1763	)
1764	]
1765	# Phase 2: record this run's surfacings. A topic whose (possibly
1766	# fuzzy) prior row is covered inherits that covered state, so a
1767	# user's covered mark survives judge naming drift instead of
1768	# silently forking into a fresh uncovered row.
1769	for topic, prior in zip(report.topics, priors):
1770	inherit_covered_at = None
1771	if prior and prior["status"] == "covered":
1772	inherit_covered_at = prior["covered_at"] or prior["last_surfaced"]
1773	store.record_discovery_surfacing(
1774	topic.name,
1775	domain=report.domain,
1776	run_ref=run_ref,
1777	as_of=as_of,
1778	inherit_covered_at=inherit_covered_at,
1779	)
1780	for topic, prior in zip(report.topics, priors):
1781	if prior:
1782	topic = dataclasses.replace(
1783	topic,
1784	previously_surfaced_count=prior["surface_count"],
1785	last_surfaced=prior["last_surfaced"],
1786	covered=prior["status"] == "covered",
1787	)
1788	annotated.append(topic)
1789	return dataclasses.replace(report, topics=annotated)
1790
1791
1792	def _record_discovery_queue_safely(
1793	report: schema.DiscoveryReport,
1794	args: argparse.Namespace,
1795	config: dict[str, object],
1796	run_ref: str \| None = None,
1797	) -> schema.DiscoveryReport:
1798	"""Annotate + record the discovery queue, degrading a broken research.db
1799	(locked, read-only dir, corrupt) to a stderr warning: a queue failure
1800	must never destroy a finished pipeline run or the protocol's final
1801	brief. Shared verbatim by the one-shot and finalize paths."""
1802	try:
1803	return _annotate_and_record_discovery_queue(
1804	report, args, config, run_ref=run_ref,
1805	)
1806	except (sqlite3.Error, OSError) as exc:
1807	sys.stderr.write(
1808	f"[last30days] Warning: discovery queue unavailable ({exc}); "
1809	"continuing without queue annotations.\n"
1810	)
1811	return report
1812
1813
1814	def _emit_and_save_discovery_report(
1815	report: schema.DiscoveryReport,
1816	args: argparse.Namespace,
1817	domain: str,
1818	) -> None:
1819	"""Render a discovery report per --emit, honor --output/--save-dir, and
1820	print it. Shared verbatim by the one-shot and finalize paths."""
1821	if args.emit == "json":
1822	payload = schema.to_dict(report) if args.json_profile == "raw" else schema.to_discovery_export(report)
1823	rendered = json.dumps(payload, indent=2, sort_keys=True)
1824	else:
1825	rendered = render.render_discovery(report)
1826
1827	if args.output:
1828	output_path = save_rendered_output(rendered, args.output)
1829	sys.stderr.write(f"[last30days] Saved output to {output_path}\n")
1830	if args.save_dir:
1831	save_path = _save_discovery_output(
1832	rendered,
1833	domain=domain or "trending",
1834	emit=args.emit,
1835	save_dir=args.save_dir,
1836	suffix=args.save_suffix or "",
1837	)
1838	sys.stderr.write(f"[last30days] Saved output to {save_path}\n")
1839	print(rendered)
1840
1841
1842	def _discovery_strict_exit_code(
1843	source_status: dict[str, schema.SourceOutcome],
1844	config: dict[str, object],
1845	) -> int:
1846	"""The ONE LAST30DAYS_STRICT_EXIT evaluation for every discovery
1847	invocation - the one-shot and all three protocol legs (issue #384's
1848	discovery counterpart). Rendering/output already happened by the time
1849	this runs; only the exit code shifts to 3 when strict exit is on and any
1850	source outcome is neither clean nor an expected skip."""
1851	strict = str(config.get("LAST30DAYS_STRICT_EXIT") or "").strip().lower()
1852	if strict not in {"1", "true", "yes", "on"}:
1853	return 0
1854	degraded = sorted(
1855	source for source, outcome in (source_status or {}).items()
1856	if outcome.state not in _STRICT_EXIT_OK_STATES
1857	)
1858	if not degraded:
1859	return 0
1860	sys.stderr.write(
1861	f"[last30days] strict-exit: degraded sources: {', '.join(degraded)}\n"
1862	)
1863	sys.stderr.flush()
1864	return 3
1865
1866
1867	def _require_discover_mock_parity(
1868	loaded_mock: bool,
1869	args_mock: bool,
1870	*,
1871	label: str,
1872	path: Path \| None,
1873	) -> None:
1874	"""A protocol leg's --mock flag must match the loaded handoff file's
1875	stamped provenance: mock-born state finalized by a real run would fake a
1876	real brief from fixture data, and real state finalized by --mock would
1877	silently drop the round's queue write. Mismatch is a contract failure
1878	(exit 2 via HandoffContractError)."""
1879	if bool(loaded_mock) == bool(args_mock):
1880	return
1881	location = str(path) if path is not None else "(unknown path)"
1882	if loaded_mock:
1883	raise discovery_handoff.HandoffContractError(
1884	f"{label} {location} is mock-born (a --mock leg wrote it): "
1885	"mock-born state cannot be finalized by a real run. Re-run this "
1886	"leg with --mock, or start a fresh real `--discover "
1887	"--nominate-only` sweep."
1888	)
1889	raise discovery_handoff.HandoffContractError(
1890	f"{label} {location} was written by a real run: real state "
1891	"cannot be finalized by a --mock run. Drop --mock, or start a fresh "
1892	"`--discover --nominate-only --mock` sweep."
1893	)
1894
1895
1896	def _run_queue_list(args: argparse.Namespace, config: dict[str, object]) -> int:
1897	"""List uncovered surfaced topics from the persistent discovery queue."""
1898	import store
1899
1900	db_path = _scoped_store_db(args)
1901	if not Path(db_path or store.DB_PATH).exists():
1902	print("Discovery queue is empty - no discovery run has recorded topics yet.")
1903	return 0
1904	with store.scoped_db(db_path):
1905	rows = store.list_discovery_queue(status="surfaced")
1906	if not rows:
1907	# An existing db with zero queue rows (e.g. created via --store)
1908	# means no discovery run has recorded anything - only claim
1909	# "every topic is covered" when covered rows actually exist.
1910	if store.list_discovery_queue():
1911	print("Discovery queue is empty - every surfaced topic is marked covered.")
1912	else:
1913	print("Discovery queue is empty - no discovery run has recorded topics yet.")
1914	return 0
1915
1916	headers = ("name", "domain", "surface_count", "last_surfaced", "status")
1917	table = [
1918	(
1919	str(row["name"]),
1920	str(row["domain"] or "-"),
1921	str(row["surface_count"]),
1922	str(row["last_surfaced"]),
1923	str(row["status"]),
1924	)
1925	for row in rows
1926	]
1927	widths = [
1928	max(len(headers[column]), *(len(row[column]) for row in table))
1929	for column in range(len(headers))
1930	]
1931	lines = [
1932	" ".join(header.ljust(widths[i]) for i, header in enumerate(headers)).rstrip(),
1933	" ".join("-" * widths[i] for i in range(len(headers))),
1934	]
1935	lines.extend(
1936	" ".join(row[i].ljust(widths[i]) for i in range(len(headers))).rstrip()
1937	for row in table
1938	)
1939	print("\n".join(lines))
1940	return 0
1941
1942
1943	def _run_queue_cover(
1944	args: argparse.Namespace,
1945	config: dict[str, object],
1946	name: str,
1947	) -> int:
1948	"""Mark a queued discovery topic covered; unknown names error loudly."""
1949	import store
1950
1951	if not name:
1952	sys.stderr.write(
1953	"[last30days] queue cover requires a topic name: "
1954	'queue cover "<topic name>".\n'
1955	)
1956	return 2
1957	db_path = _scoped_store_db(args)
1958	if not Path(db_path or store.DB_PATH).exists():
1959	sys.stderr.write(
1960	f"[last30days] No queued topic named {name!r}: the discovery queue "
1961	"is empty (no discovery run has recorded topics yet).\n"
1962	)
1963	return 2
1964	with store.scoped_db(db_path):
1965	row = store.mark_discovery_covered(
1966	name, as_of=datetime.date.today().isoformat()
1967	)
1968	if row is None:
1969	sys.stderr.write(
1970	f"[last30days] No queued topic named {name!r}. Covering requires "
1971	"the exact topic name; run 'queue list' to see queued names.\n"
1972	)
1973	return 2
1974	print(f"Marked covered: {row['name']} (covered {row['covered_at']})")
1975	return 0
1976
1977
1978	def _resolve_discovery_source_boundary(
1979	args: argparse.Namespace, config: dict[str, object],
1980	) -> tuple[list[str] \| None, list[str] \| None] \| None:
1981	"""Resolve the discovery sweep's source lists from the user's boundary.
1982
1983	Returns ``(listing_sources, enrichment_boundary)`` - the discovery-capable
1984	subset for the sweep, and the user's ORIGINAL boundary honored by the
1985	per-topic research passes (which reach beyond the listing feeds - e.g.
1986	Techmeme, arXiv, YouTube, Polymarket); both None mean every available
1987	source. Returns None (after writing the exit-2 error) when the configured
1988	boundary leaves nothing to sweep: silently widening to all feeds would
1989	query sources the user filtered out.
1990	"""
1991	requested_sources = resolve_requested_sources(args.search, config)
1992	enrich_requested_sources = list(requested_sources) if requested_sources else None
1993	if requested_sources:
1994	discovery_sources = [
1995	source for source in requested_sources
1996	if source in pipeline.DISCOVERY_SOURCES
1997	]
1998	if not discovery_sources:
1999	origin = "--search" if args.search is not None else "LAST30DAYS_DEFAULT_SEARCH"
2000	sys.stderr.write(
2001	f"[last30days] {origin} has no discovery-capable sources "
2002	f"(unsupported: {', '.join(requested_sources)}); discovery "
2003	f"sweeps use: {', '.join(pipeline.DISCOVERY_SOURCES)}. Pass "
2004	"--search with one of those (or clear the source filter) to "
2005	"run a sweep.\n"
2006	)
2007	return None
2008	requested_sources = discovery_sources
2009	return requested_sources, enrich_requested_sources
2010
2011
2012	def _discover_subreddits(args: argparse.Namespace) -> list[str] \| None:
2013	return (
2014	[value.strip().removeprefix("r/") for value in args.subreddits.split(",") if value.strip()]
2015	if args.subreddits else None
2016	)
2017
2018
2019	def _discover_domain(args: argparse.Namespace) -> str:
2020	"""The whitespace-normalized discovery domain; empty = global trending."""
2021	return " ".join(str(args.discover or "").split())
2022
2023
2024	def _run_discover(args: argparse.Namespace, config: dict[str, object]) -> int:
2025	domain = _discover_domain(args)
2026	# Empty domain = global trending: sweep every river feed's hot list with no
2027	# keyword gate. The confidence floor is what keeps junk out, not a keyword.
2028	# (--as-of and HTML rejection live in _main's shared --discover dispatch,
2029	# so every leg - one-shot or protocol - applies the same guards.)
2030	if args.synthesis_file:
2031	sys.stderr.write("[last30days] Warning: --synthesis-file is not used by discovery mode.\n")
2032
2033	boundary = _resolve_discovery_source_boundary(args, config)
2034	if boundary is None:
2035	return 2
2036	requested_sources, enrich_requested_sources = boundary
2037	subreddits = _discover_subreddits(args)
2038	depth = "deep" if args.deep else "quick" if args.quick else "default"
2039	try:
2040	report = pipeline.run_discover(
2041	domain=domain,
2042	config=config,
2043	depth=depth,
2044	requested_sources=requested_sources,
2045	mock=args.mock,
2046	subreddits=subreddits,
2047	lookback_days=args.lookback_days or 30,
2048	as_of_date=args.as_of_date,
2049	enrich=not args.discover_shallow,
2050	enrich_requested_sources=enrich_requested_sources,
2051	)
2052	except ValueError as exc:
2053	sys.stderr.write(f"[last30days] {exc}\n")
2054	return 2
2055
2056	# Persistent topic queue: annotate this report from prior surfacings, then
2057	# record this run's surfacings - BEFORE rendering/export so the Pipeline
2058	# line and the JSON queue fields see the annotations. Mock runs stay 100%
2059	# side-effect-free.
2060	if not args.mock:
2061	report = _record_discovery_queue_safely(report, args, config)
2062
2063	_emit_and_save_discovery_report(report, args, domain)
2064	return _discovery_strict_exit_code(report.source_status, config)
2065
2066
2067	def _discover_handoff_state_dir(args: argparse.Namespace) -> Path \| None:
2068	"""One resolver for every protocol leg's handoff files: the save dir when
2069	given (mirroring _scoped_store_db's scoping), else the config dir - the
2070	same base _last_report_cache_path uses. args.save_dir is read AFTER the
2071	LAST30DAYS_MEMORY_DIR fallback in _main resolved it."""
2072	return discovery_handoff.handoff_state_dir(
2073	getattr(args, "save_dir", None), env.CONFIG_DIR
2074	)
2075
2076
2077	def _run_discover_nominate(args: argparse.Namespace, config: dict[str, object]) -> int:
2078	"""Protocol leg 1: sweep the listings, build the full judge pool, write
2079	the nominations bundle, and print the host-facing judging digest.
2080
2081	No stage-1 judge, enrichment, confidence floor, or queue writes happen on
2082	this leg - the host judges from the bundle and leg 2 (--judgments)
2083	resumes from it. A zero-nomination sweep short-circuits to the existing
2084	nothing-solid brief with NO bundle written: there is nothing to judge.
2085	Writing a fresh bundle starts a NEW protocol round, so any pending
2086	report left by a prior round is deleted alongside it.
2087	"""
2088	domain = _discover_domain(args)
2089	boundary = _resolve_discovery_source_boundary(args, config)
2090	if boundary is None:
2091	return 2
2092	requested_sources, enrich_requested_sources = boundary
2093	lookback_days = args.lookback_days or 30
2094	try:
2095	result = pipeline.run_discover_nominate(
2096	domain=domain,
2097	config=config,
2098	depth="deep" if args.deep else "quick" if args.quick else "default",
2099	requested_sources=requested_sources,
2100	mock=args.mock,
2101	subreddits=_discover_subreddits(args),
2102	lookback_days=lookback_days,
2103	as_of_date=args.as_of_date,
2104	)
2105	except ValueError as exc:
2106	sys.stderr.write(f"[last30days] {exc}\n")
2107	return 2
2108
2109	if not result.pool:
2110	print(render.render_discovery(pipeline.nominate_nothing_solid_report(result)))
2111	return _discovery_strict_exit_code(result.source_status, config)
2112
2113	entries = [
2114	discovery_handoff.PoolEntry(
2115	nomination=nomination,
2116	cluster_id=cluster_id,
2117	# No provider runs on this leg, so the nomination's name and junk
2118	# flag ARE the topic_shape heuristics - stored on the row as
2119	# leg 2's fallback for anything the host leaves unjudged.
2120	heuristic_name=nomination.name,
2121	heuristic_junk=nomination.junk_shape,
2122	)
2123	for nomination, cluster_id in result.pool
2124	]
2125	bundle = discovery_handoff.write_nominations_bundle(
2126	entries,
2127	domain=result.plan.domain,
2128	tier="shallow" if args.discover_shallow else "deep",
2129	from_date=result.from_date,
2130	to_date=result.to_date,
2131	lookback_days=lookback_days,
2132	enrichment_source_boundary=enrich_requested_sources,
2133	requested_sources=requested_sources,
2134	# The sweep's finalized per-source outcomes ride the bundle so legs
2135	# 2-3 report degraded coverage instead of silently reading clean; the
2136	# mock stamp keeps mock-born and real state from cross-finalizing.
2137	source_status=result.source_status,
2138	mock=args.mock,
2139	# Same resolution as _discover_handoff_state_dir: save dir when
2140	# given, else the config dir.
2141	save_dir=getattr(args, "save_dir", None),
2142	config_dir=env.CONFIG_DIR,
2143	)
2144	# A fresh bundle starts a NEW protocol round: a pending report left by a
2145	# prior round is cross-round state a bare --finalize could silently
2146	# consume - delete it (missing file is a no-op).
2147	state_dir = _discover_handoff_state_dir(args)
2148	if state_dir is not None:
2149	discovery_handoff.pending_report_path(state_dir).unlink(missing_ok=True)
2150	print(discovery_handoff.build_host_digest(bundle))
2151	print(
2152	"\nJudgments file schema (leg 2): "
2153	f'{{"bundle_id": "{bundle.bundle_id}", "judgments": '
2154	'[{"id": "n1", "name": "<short topic name>", "junk": false, '
2155	'"worthiness": 0-100}, ...]}. '
2156	"Then resume with: --discover --judgments <path>."
2157	)
2158	return _discovery_strict_exit_code(result.source_status, config)
2159
2160
2161	def _run_discover_resume(args: argparse.Namespace, config: dict[str, object]) -> int:
2162	"""Protocol leg 2: resume from the nominations bundle, apply the host
2163	judgments file, run the deep per-topic research pass, and persist the
2164	ranked result as the pending report for leg 3 (--finalize).
2165
2166	Contract failures (missing/stale bundle, judgments not bound to it, a
2167	bundle whose mock provenance disagrees with this run's --mock flag, an
2168	unwritable pending-report path) raise HandoffContractError and map to
2169	exit 2 in _run_discover_protocol_leg. Zero floor survivors renders the
2170	nothing-solid brief right here (clearing any stale prior-round pending
2171	file): no pending file, no leg 3. No queue writes and no artifact saves
2172	happen on this leg - the topic queue and the rendered brief belong to
2173	leg 3.
2174	"""
2175	save_dir = getattr(args, "save_dir", None)
2176	bundle = discovery_handoff.read_nominations_bundle(
2177	save_dir=save_dir, config_dir=env.CONFIG_DIR,
2178	)
2179	_require_discover_mock_parity(
2180	bundle.mock, args.mock,
2181	label="Nominations bundle", path=bundle.path,
2182	)
2183	judgments = discovery_handoff.read_judgments(
2184	args.judgments, bundle, save_dir=save_dir, config_dir=env.CONFIG_DIR,
2185	)
2186	result = pipeline.run_discover_resume(
2187	bundle, judgments, config=config, mock=args.mock,
2188	)
2189	report = result.report
2190
2191	if not report.topics:
2192	# Nothing cleared the floor: the honest brief ends the protocol here.
2193	# This round wrote no pending file, so a stale one from an earlier
2194	# round must not survive to feed a bare --finalize (missing file is
2195	# a no-op).
2196	state_dir = _discover_handoff_state_dir(args)
2197	if state_dir is not None:
2198	discovery_handoff.pending_report_path(state_dir).unlink(missing_ok=True)
2199	print(render.render_discovery(report))
2200	return _discovery_strict_exit_code(report.source_status, config)
2201
2202	state_dir = _discover_handoff_state_dir(args)
2203	if state_dir is None:
2204	# Unreachable in practice - reading the bundle above required one of
2205	# these locations - but kept as a loud contract error, not an assert.
2206	raise discovery_handoff.HandoffContractError(
2207	"No handoff location available to write the pending report: "
2208	"pass --save-dir or configure ~/.config/last30days/."
2209	)
2210	pending_path = discovery_handoff.pending_report_path(state_dir)
2211	payload = {
2212	"kind": schema.DISCOVERY_PENDING_KIND,
2213	"schema_version": schema.DISCOVERY_PENDING_SCHEMA_VERSION,
2214	"bundle_id": bundle.bundle_id,
2215	# Fresh TTL clock: leg 3 measures staleness from THIS resume run,
2216	# not from the leg-1 sweep.
2217	"generated_at": report.generated_at,
2218	# Same run_ref format the queue records (leg 3 replays it verbatim).
2219	"run_ref": f"discover:{report.domain or 'trending'}:{report.generated_at}",
2220	# Leg-2 provenance: leg 3 refuses to finalize across the mock/real
2221	# boundary in either direction.
2222	"mock": bool(args.mock),
2223	# Full schema round-trip (the _write_last_run precedent): leg 3
2224	# rebuilds the report from this dict instead of re-running anything.
2225	"report": schema.to_dict(report),
2226	"angle_inputs": result.angle_inputs,
2227	}
2228	# ONE post-loop write from the main thread; enrichment workers are daemon
2229	# threads and never touch disk.
2230	try:
2231	state_dir.mkdir(parents=True, exist_ok=True)
2232	pending_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
2233	except OSError as exc:
2234	# A locked/read-only/full disk is the protocol's clean exit-2 path,
2235	# never a traceback (same contract as the bundle write).
2236	raise discovery_handoff.HandoffContractError(
2237	f"Could not write pending discovery report {pending_path}: {exc}"
2238	) from exc
2239
2240	print(
2241	f"Judged discovery resume: {len(report.topics)} topic"
2242	f"{'s' if len(report.topics) != 1 else ''} cleared the floor "
2243	f"(bundle_id {bundle.bundle_id})."
2244	)
2245	print(f"Pending report: {pending_path}")
2246	print("\nAngle inputs by nomination id:")
2247	print(json.dumps(result.angle_inputs, indent=2))
2248	print(
2249	"\nWrite the angles file (leg 3): "
2250	f'{{"bundle_id": "{bundle.bundle_id}", "angles": '
2251	'[{"id": "n1", "podcast": "<one-sentence hook>", '
2252	'"x_article": "<one-sentence hook>"}, ...]} - one row per topic id '
2253	"above.\n"
2254	"Then finalize with: --discover --finalize --angles <path>."
2255	)
2256	return _discovery_strict_exit_code(report.source_status, config)
2257
2258
2259	def _run_discover_finalize(args: argparse.Namespace, config: dict[str, object]) -> int:
2260	"""Protocol leg 3: load the leg-2 pending report, apply host angles,
2261	render the final brief, save discovery artifacts, and record the topic
2262	queue. The cheap offline leg - no sweep, no enrichment, no providers,
2263	no network; everything renders from the pending report. (HTML/--as-of
2264	rejection lives in _main's shared --discover dispatch.)
2265
2266	Contract failures (missing/stale/mismatched pending report or angles)
2267	raise HandoffContractError and map to exit 2 in
2268	_run_discover_protocol_leg. The pending file is deliberately LEFT IN
2269	PLACE on success: a finalize retry with a corrected angles file must
2270	keep working within the TTL, and the queue records under the pending
2271	report's leg-2 run_ref, so retries never double-count a surfacing.
2272	Mock finalize renders identically but writes no queue rows.
2273	"""
2274	import dataclasses
2275
2276	save_dir = getattr(args, "save_dir", None)
2277	pending = discovery_handoff.read_pending_report(
2278	save_dir=save_dir, config_dir=env.CONFIG_DIR,
2279	)
2280	_require_discover_mock_parity(
2281	pending.mock, args.mock,
2282	label="Pending discovery report", path=pending.path,
2283	)
2284	angles = discovery_handoff.read_angles(
2285	args.angles, pending, save_dir=save_dir, config_dir=env.CONFIG_DIR,
2286	)
2287	try:
2288	report = schema.discovery_report_from_dict(pending.report)
2289	except (KeyError, TypeError, ValueError) as exc:
2290	# The envelope validated but the report body is structurally
2291	# incomplete: a contract failure with the resume remedy, never a
2292	# traceback out of the finalize leg.
2293	raise discovery_handoff.HandoffContractError(
2294	f"Pending discovery report {pending.path} carries a malformed "
2295	f"report body ({type(exc).__name__}: {exc}). "
2296	f"{discovery_handoff._RESUME_REMEDY}"
2297	) from exc
2298
2299	if angles:
2300	# Host angles are keyed by nomination id; the pending report's
2301	# angle_inputs mapping carries each surviving id's applied topic
2302	# name, which is how angles land on the right DiscoveryTopic.
2303	angles_by_name = {
2304	name: host
2305	for nomination_id, host in angles.items()
2306	if (name := (pending.angle_inputs.get(nomination_id) or {}).get("name"))
2307	}
2308	report = dataclasses.replace(report, topics=[
2309	dataclasses.replace(
2310	topic,
2311	podcast_angle=host.podcast,
2312	x_article_angle=host.x_article,
2313	)
2314	if (host := angles_by_name.get(topic.name)) is not None
2315	else topic
2316	for topic in report.topics
2317	])
2318
2319	# Persistent topic queue: the protocol's ONE queue write happens here,
2320	# under the leg-2 run identity (pending.run_ref) so finalize retries are
2321	# idempotent. Mock runs stay 100% side-effect-free.
2322	if not args.mock:
2323	report = _record_discovery_queue_safely(
2324	report, args, config, run_ref=pending.run_ref or None,
2325	)
2326
2327	_emit_and_save_discovery_report(report, args, report.domain)
2328	return _discovery_strict_exit_code(report.source_status, config)
2329
2330
2331	def _run_discover_protocol_leg(
2332	args: argparse.Namespace, config: dict[str, object]
2333	) -> int:
2334	"""Route one validated protocol invocation to its leg. Contract failures
2335	(unreadable/stale/mismatched handoff files) map to stderr + exit 2 here,
2336	so the leg bodies (U3-U5) raise HandoffContractError freely."""
2337	try:
2338	if args.nominate_only:
2339	return _run_discover_nominate(args, config)
2340	# --judgments dispatch keys on flag presence (is not None), matching
2341	# the --discover convention: never on the path string's truthiness.
2342	if args.judgments is not None:
2343	return _run_discover_resume(args, config)
2344	return _run_discover_finalize(args, config)
2345	except discovery_handoff.HandoffContractError as exc:
2346	sys.stderr.write(f"[last30days] {exc.message}\n")
2347	return 2
2348
2349
2350	_STRICT_EXIT_OK_STATES = {"ok", "no-results", "skipped-unconfigured"}
2351
2352
2353	def _strict_exit_code(
2354	report: schema.Report,
2355	entity_reports: list[tuple[str, schema.Report]] \| None,
2356	config: dict[str, object],
2357	) -> int:
2358	"""Opt-in machine-detectable degraded-run signal (issue #384).
2359
2360	When LAST30DAYS_STRICT_EXIT is truthy, a run whose report carries any
2361	source outcome that is neither clean nor a plain no-results exits 3 so
2362	cron/CI wrappers can distinguish degraded coverage from success. Default
2363	behavior (exit 0, warning rendered in the report footer) is unchanged.
2364	"""
2365	raw = str(config.get("LAST30DAYS_STRICT_EXIT") or "").strip().lower()
2366	if raw not in {"1", "true", "yes", "on"}:
2367	return 0
2368	reports = [report] + [rep for _, rep in (entity_reports or [])]
2369	degraded = sorted({
2370	name
2371	for rep in reports
2372	for name, outcome in (rep.source_status or {}).items()
2373	if outcome.state not in _STRICT_EXIT_OK_STATES
2374	})
2375	if not degraded:
2376	return 0
2377	sys.stderr.write(
2378	f"[last30days] strict-exit: degraded sources: {', '.join(degraded)}\n"
2379	)
2380	sys.stderr.flush()
2381	return 3
2382
2383
2384	def _audience_register_for_run(
2385	args: argparse.Namespace,
2386	config: dict[str, object],
2387	entity_reports: list[tuple[str, schema.Report]] \| None,
2388	) -> registers.AudienceRegister:
2389	"""Resolve CLI > config for single-topic standard brief renderers."""
2390
2391	from lib import planner
2392
2393	topic = " ".join(getattr(args, "topic", [])).strip()
2394	comparison_topic_requested = bool(
2395	len(planner._comparison_entities(topic)) >= 2
2396	or args.competitors is not None
2397	or args.competitors_list
2398	or args.competitors_plan
2399	)
2400	if (
2401	entity_reports
2402	or comparison_topic_requested
2403	or args.drill
2404	or args.emit not in {"compact", "md", "html"}
2405	):
2406	return registers.get_register()
2407	explicit = getattr(args, "register", None)
2408	configured = config.get("LAST30DAYS_REGISTER")
2409	name = explicit or (str(configured) if configured else "default")
2410	# Preserve configs written by the pre-register ELI5 follow-up command.
2411	legacy_eli5 = str(config.get("ELI5_MODE") or "").strip().lower()
2412	if not explicit and not configured and legacy_eli5 in {"1", "true", "yes", "on"}:
2413	name = "eli5"
2414	return registers.get_register(name)
2415
2416
2417	def _render_save_and_print(
2418	args: argparse.Namespace,
2419	report: schema.Report,
2420	entity_reports: list[tuple[str, schema.Report]] \| None,
2421	synthesis_md: str \| None,
2422	config: dict[str, object],
2423	) -> int:
2424	fun_level = str(config.get("FUN_LEVEL", "medium")).lower()
2425	try:
2426	audience = _audience_register_for_run(args, config, entity_reports)
2427	except ValueError as exc:
2428	sys.stderr.write(f"[last30days] {exc}\n")
2429	return 2
2430	if audience.name != "default":
2431	sys.stderr.write(f"[last30days] Audience register: {audience.name}\n")
2432	sys.stderr.flush()
2433	# Comparison HTML is the one case where the saved file's title and content
2434	# have to be overridden away from the leading entity's report. Compute the
2435	# gate once so the footer-display and save-output paths can't disagree.
2436	is_comparison_html = bool(entity_reports) and args.emit == "html"
2437	footer_save_path = None
2438	if args.output:
2439	footer_save_path = compute_output_path_display(args.output)
2440	elif args.save_dir:
2441	save_topic_for_display = comparison_topic(entity_reports) if is_comparison_html else report.topic
2442	footer_save_path = compute_save_path_display(
2443	args.save_dir, save_topic_for_display, args.save_suffix or "", args.emit
2444	)
2445
2446	if entity_reports:
2447	rendered = emit_comparison_output(
2448	entity_reports,
2449	args.emit,
2450	fun_level=fun_level,
2451	save_path=footer_save_path,
2452	synthesis_md=synthesis_md,
2453	json_profile=args.json_profile,
2454	)
2455	else:
2456	rendered = emit_output(
2457	report,
2458	args.emit,
2459	fun_level=fun_level,
2460	save_path=footer_save_path,
2461	synthesis_md=synthesis_md,
2462	json_profile=args.json_profile,
2463	register=audience.name,
2464	)
2465	has_private_corpus = _report_has_private_corpus(report) or bool(
2466	entity_reports
2467	and any(_report_has_private_corpus(entity) for _label, entity in entity_reports)
2468	)
2469	private_saved_format = has_private_corpus
2470	publish_companion_paths: list[Path] = []
2471	if args.output:
2472	output_path = save_rendered_output(
2473	rendered,
2474	args.output,
2475	private=private_saved_format,
2476	)
2477	if args.emit == "html":
2478	publish_companion_paths.append(output_path)
2479	sys.stderr.write(f"[last30days] Saved output to {output_path}\n")
2480	sys.stderr.flush()
2481	if args.save_dir:
2482	# Save the main topic's raw file (single-entity or comparison main).
2483	# Bind the render to the path save_output actually allocates so the
2484	# saved report and stdout agree even when collision fallback is used.
2485	def _render_with_actual_path(actual_path: Path) -> str:
2486	nonlocal rendered
2487	display = compute_output_path_display(str(actual_path))
2488	if entity_reports:
2489	rendered = emit_comparison_output(
2490	entity_reports,
2491	args.emit,
2492	fun_level=fun_level,
2493	save_path=display,
2494	synthesis_md=synthesis_md,
2495	json_profile=args.json_profile,
2496	)
2497	else:
2498	rendered = emit_output(
2499	report,
2500	args.emit,
2501	fun_level=fun_level,
2502	save_path=display,
2503	synthesis_md=synthesis_md,
2504	json_profile=args.json_profile,
2505	register=audience.name,
2506	)
2507	if args.emit not in {"json", "html"} and not entity_reports:
2508	# Markdown saves keep the complete debug artifact (all clusters
2509	# and per-source items), matching the render_fn-less path in
2510	# save_output and the comparison peer saves. Saving the compact
2511	# stdout render instead made most collected evidence
2512	# unrecoverable from the raw file (#923). The stdout re-render
2513	# above still runs so the visible footer cites the real path,
2514	# and the saved artifact carries the same citation.
2515	return render.render_full(report, save_path=display)
2516	return rendered
2517
2518	save_path = save_output(
2519	report,
2520	args.emit,
2521	args.save_dir,
2522	suffix=args.save_suffix or "",
2523	synthesis_md=synthesis_md,
2524	topic_override=comparison_topic(entity_reports) if is_comparison_html else None,
2525	json_profile=args.json_profile,
2526	register=audience.name,
2527	private=private_saved_format,
2528	render_fn=_render_with_actual_path,
2529	)
2530	if args.emit == "html":
2531	publish_companion_paths.append(save_path)
2532	sys.stderr.write(f"[last30days] Saved output to {save_path}\n")
2533	comparison_peer_paths: list[Path] = []
2534	# Competitor / vs-mode: also save a per-entity raw file for each peer.
2535	# Matches historical vs-mode behavior (N passes -> N save files).
2536	if entity_reports and len(entity_reports) > 1:
2537	for label, entity_report in entity_reports[1:]:
2538	peer_path = save_output(
2539	entity_report, args.emit, args.save_dir,
2540	suffix=args.save_suffix or "",
2541	synthesis_md=synthesis_md,
2542	json_profile=args.json_profile,
2543	private=_report_has_private_corpus(entity_report),
2544	)
2545	comparison_peer_paths.append(peer_path)
2546	sys.stderr.write(f"[last30days] Saved output to {peer_path}\n")
2547	peers_display = ", ".join(str(path) for path in comparison_peer_paths)
2548	sys.stderr.write(
2549	f"[last30days] Comparison artifact set: main={save_path}; "
2550	f"peers={peers_display}\n"
2551	)
2552	sys.stderr.flush()
2553	if args.publish_html:
2554	try:
2555	has_private_corpus = "corpus" in report.source_status or bool(
2556	entity_reports
2557	and any("corpus" in entity.source_status for _label, entity in entity_reports)
2558	)
2559	publish_rendered = rendered
2560	if has_private_corpus:
2561	sys.stderr.write(
2562	"[last30days] Excluding local corpus evidence and synthesis from published HTML.\n"
2563	)
2564	if entity_reports:
2565	publish_rendered = emit_comparison_output(
2566	[
2567	(label, schema.without_sources(entity, {"corpus"}))
2568	for label, entity in entity_reports
2569	],
2570	"html",
2571	fun_level=fun_level,
2572	save_path=footer_save_path,
2573	synthesis_md=None,
2574	json_profile=args.json_profile,
2575	)
2576	else:
2577	publish_rendered = emit_output(
2578	schema.without_sources(report, {"corpus"}),
2579	"html",
2580	fun_level=fun_level,
2581	save_path=footer_save_path,
2582	synthesis_md=None,
2583	json_profile=args.json_profile,
2584	register=audience.name,
2585	)
2586	publish_result = publish_rendered_html(
2587	publish_rendered,
2588	password=_publish_password_for_args(args, config),
2589	companion_paths=publish_companion_paths,
2590	)
2591	sys.stderr.write(f"[last30days] Published HTML to {publish_result['url']}\n")
2592	for warning in publish_result.get("_metadata_errors") or []:
2593	sys.stderr.write(f"[last30days] Publish metadata warning: {warning}\n")
2594	if publish_result.get("update_key"):
2595	sys.stderr.write(
2596	"[last30days] ht-ml.app returned an update key; not writing it "
2597	"to stdout, HTML, or publish metadata.\n"
2598	)
2599	sys.stderr.flush()
2600	except Exception as exc:
2601	sys.stderr.write(f"[last30days] HTML publish failed: {exc}\n")
2602	sys.stderr.flush()
2603	print(rendered)
2604	return _strict_exit_code(report, entity_reports, config)
2605
2606
2607	def _propagate_config_to_environ(config: dict[str, object]) -> None:
2608	"""Push relevant env keys to os.environ so provider modules can read them.
2609
2610	The env.get_config() function reads from a .env file, but providers.py
2611	reads from os.environ directly. Without this, OPENAI_BASE_URL and
2612	XAI_BASE_URL overrides are silently ignored. This is a no-op for
2613	keys that are already set in process env.
2614	"""
2615	for key in ("OPENAI_BASE_URL", "XAI_BASE_URL", "OPENROUTER_BASE_URL"):
2616	val = config.get(key)
2617	if val and not os.environ.get(key):
2618	os.environ[key] = val
2619
2620
2621	def _setup_allows_browser_cookies(args: argparse.Namespace, extra_argv: list[str]) -> bool:
2622	return (
2623	not args.no_browser_cookies
2624	and not args.diagnose
2625	and not args.preflight
2626	and "--allow-browser-cookies" in extra_argv
2627	)
2628
2629
2630	SETUP_PASSTHROUGH_FLAGS = {
2631	"--allow-browser-cookies",
2632	"--device-auth",
2633	"--github",
2634	"--github-start",
2635	"--github-poll",
2636	"--openclaw",
2637	"--store-key",
2638	}
2639
2640	STORE_KEY_FLAG = "--store-key"
2641
2642
2643	def _split_store_key(extra_argv: list[str]) -> tuple[bool, str, list[str]]:
2644	"""Pull ``--store-key <NAME>`` / ``--store-key=<NAME>`` out of ``extra_argv``.
2645
2646	Returns ``(present, name, remaining)``. ``name`` is "" when the flag has
2647	no value; ``remaining`` is every other passthrough token, so the regular
2648	allowlist check still applies to them.
2649	"""
2650	present = False
2651	name = ""
2652	remaining: list[str] = []
2653	i = 0
2654	while i < len(extra_argv):
2655	arg = extra_argv[i]
2656	if arg == STORE_KEY_FLAG:
2657	present = True
2658	if i + 1 < len(extra_argv) and not extra_argv[i + 1].startswith("-"):
2659	name = extra_argv[i + 1]
2660	i += 2
2661	continue
2662	i += 1
2663	continue
2664	if arg.startswith(STORE_KEY_FLAG + "="):
2665	present = True
2666	name = arg[len(STORE_KEY_FLAG) + 1:]
2667	i += 1
2668	continue
2669	remaining.append(arg)
2670	i += 1
2671	return present, name, remaining
2672
2673
2674	# One credential line: longer than any real token, short enough that a
2675	# misdirected stream on stdin cannot grow memory.
2676	STORE_KEY_MAX_BYTES = 64 * 1024
2677
2678
2679	def _run_store_key(name: str) -> int:
2680	"""``setup --store-key <NAME>``: persist one allowlisted credential from stdin.
2681
2682	Reads exactly one line from stdin (bounded to ``STORE_KEY_MAX_BYTES``),
2683	strips whitespace, and writes it to the global ``.env`` as a 0o600 secret
2684	through ``setup_wizard.write_api_key``. An existing line for the same
2685	name is replaced, so a rejected credential can be rotated by running the
2686	command again. The value never reaches stdout or stderr: stdout carries
2687	``NAME=****`` plus a JSON line ``{"persisted": bool, "key": NAME}``. A
2688	name outside ``env.KEYCHAIN_KEYS`` or an empty value exits 2 without
2689	echoing anything.
2690	"""
2691	from lib import setup_wizard
2692
2693	if name not in env.KEYCHAIN_KEYS:
2694	# Do not enumerate the allowlist here: on an official-only host a
2695	# failure hint must not name the legacy credential keys.
2696	sys.stderr.write(
2697	"[last30days] setup --store-key: unknown or missing key name "
2698	"(must be a credential name the engine loads from its .env; "
2699	"see CONFIGURATION.md).\n"
2700	)
2701	return 2
2702	value = sys.stdin.readline(STORE_KEY_MAX_BYTES).strip()
2703	if not value:
2704	sys.stderr.write(
2705	f"[last30days] setup --store-key {name}: empty value on stdin; "
2706	"pipe the credential as a single line.\n"
2707	)
2708	return 2
2709	persisted = bool(
2710	setup_wizard.write_api_key(env.CONFIG_FILE, value, key_name=name, replace=True)
2711	)
2712	print(f"{name}=****")
2713	print(json.dumps({"persisted": persisted, "key": name}))
2714	return 0 if persisted else 1
2715
2716	SKILL_ONLY_FLAGS = {
2717	"--agent",
2718	}
2719
2720	# Doctor passthrough: `doctor --json` / `doctor --cached` mirror the setup
2721	# passthrough pattern (neither is a global parser flag; they only mean
2722	# something to doctor). `--cached` serves the stored doctor-cache.json report
2723	# within its TTL and falls through to a live run otherwise.
2724	DOCTOR_PASSTHROUGH_FLAGS = {
2725	"--json",
2726	"--cached",
2727	"--postmortem",
2728	"--probe",
2729	}
2730
2731
2732	def _looks_inline_json(value: str) -> bool:
2733	"""True when a --x-posts argument is JSON text rather than a path."""
2734	stripped = value.strip()
2735	return stripped.startswith(("{", "[")) or "\n" in value
2736
2737
2738	def _comparison_requested(args: argparse.Namespace, topic: str) -> bool:
2739	"""Whether this invocation is a comparison run (vs-topic or competitor flags)."""
2740	from lib import planner as _planner
2741
2742	return any(
2743	value is not None
2744	for value in (args.competitors, args.competitors_list, args.competitors_plan)
2745	) or len(_planner._comparison_entities(topic, uncapped=True)) >= 2
2746
2747
2748	def _read_x_envelope(
2749	path: str,
2750	topic: str,
2751	args: argparse.Namespace,
2752	*,
2753	x_handle: str \| None,
2754	x_related: list[str] \| None,
2755	) -> x_envelope.Envelope:
2756	"""Validate a host-fetched X envelope against this run's window and topic."""
2757	from_date, to_date = dates.get_date_range(
2758	args.lookback_days or 30, as_of_date=args.as_of_date
2759	)
2760	return x_envelope.read(
2761	path,
2762	(from_date, to_date),
2763	topic,
2764	handles=[x_handle] if x_handle else [],
2765	related=[h for h in (x_related or []) if h and h.strip()],
2766	)
2767
2768
2769	def _attach_entity_envelopes(comp_plan: dict[str, dict], args: argparse.Namespace) -> None:
2770	"""Validate every per-entity ``x_posts`` path in a --competitors-plan.
2771
2772	Each envelope is checked against its own entity (topic) and that entry's
2773	``x_handle``/``x_related`` handles, and stored on the entry as
2774	``_x_envelope`` for the entity sub-run. Raises EnvelopeContractError.
2775	"""
2776	for entry in comp_plan.values():
2777	raw = entry.get("x_posts")
2778	if not raw:
2779	continue
2780	if not isinstance(raw, str) or _looks_inline_json(raw):
2781	raise x_envelope.EnvelopeContractError(
2782	f"--competitors-plan entry {entry.get('_name', '')!r}: x_posts must "
2783	"be a file path to a last30days-x-posts/1 envelope, never inline JSON. "
2784	"Rewrite the plan entry, or drop its x_posts field."
2785	)
2786	related = entry.get("x_related") if isinstance(entry.get("x_related"), list) else None
2787	entry["_x_envelope"] = _read_x_envelope(
2788	raw, str(entry.get("_name") or ""), args,
2789	x_handle=entry.get("x_handle") if isinstance(entry.get("x_handle"), str) else None,
2790	x_related=[str(h) for h in related] if related else None,
2791	)
2792
2793
2794	def _combine_envelope_digests(main_sha256: str \| None, entity_sha256: dict[str, str]) -> str \| None:
2795	"""One digest binding the last-report cache to every envelope a run uses.
2796
2797	A single top-level envelope is bound by its own file digest; per-entity
2798	comparison envelopes are folded, name-sorted, into one digest. Both the
2799	cache write (validated envelopes) and the cache lookup (planned paths)
2800	must go through here so a comparison cache can be reused.
2801	"""
2802	parts: list[str] = []
2803	if main_sha256:
2804	parts.append(main_sha256)
2805	for name in sorted(entity_sha256):
2806	parts.append(f"{name}:{entity_sha256[name]}")
2807	if not parts:
2808	return None
2809	if len(parts) == 1 and main_sha256:
2810	return main_sha256
2811	return hashlib.sha256("\|".join(parts).encode("utf-8")).hexdigest()
2812
2813
2814	def _x_envelope_digest(
2815	main: x_envelope.Envelope \| None, comp_plan: dict[str, dict] \| None
2816	) -> str \| None:
2817	"""Digest of the validated envelopes this run used (cache write side)."""
2818	entity_sha256 = {
2819	name: entry["_x_envelope"].sha256
2820	for name, entry in (comp_plan or {}).items()
2821	if entry.get("_x_envelope") is not None
2822	}
2823	return _combine_envelope_digests(main.sha256 if main is not None else None, entity_sha256)
2824
2825
2826	def _validate_extra_argv(parser: argparse.ArgumentParser, topic: str, extra_argv: list[str]) -> None:
2827	if not extra_argv:
2828	return
2829	if topic.lower() == "setup":
2830	# --store-key carries a value token; the name itself is allowlisted
2831	# later in _run_store_key, not here.
2832	_, _, extra_argv = _split_store_key(extra_argv)
2833	unsupported = [arg for arg in extra_argv if arg not in SETUP_PASSTHROUGH_FLAGS]
2834	if unsupported:
2835	parser.error(
2836	"unsupported setup argument(s): "
2837	+ ", ".join(unsupported)
2838	+ f"; supported setup passthrough flags are {', '.join(sorted(SETUP_PASSTHROUGH_FLAGS))}"
2839	)
2840	return
2841	if topic.lower() == "doctor":
2842	unsupported = [arg for arg in extra_argv if arg not in DOCTOR_PASSTHROUGH_FLAGS]
2843	if unsupported:
2844	parser.error(
2845	"unsupported doctor argument(s): "
2846	+ ", ".join(unsupported)
2847	+ f"; supported doctor passthrough flags are {', '.join(sorted(DOCTOR_PASSTHROUGH_FLAGS))}"
2848	)
2849	return
2850	skill_only = [arg for arg in extra_argv if arg in SKILL_ONLY_FLAGS]
2851	other_unknown = [arg for arg in extra_argv if arg not in SKILL_ONLY_FLAGS]
2852	if skill_only:
2853	message = (
2854	"unsupported Python CLI argument(s): "
2855	+ ", ".join(skill_only)
2856	+ "; these are skill arguments and must not be forwarded to scripts/last30days.py"
2857	)
2858	if other_unknown:
2859	message += "; also unsupported: " + ", ".join(other_unknown)
2860	parser.error(message)
2861	parser.error("unsupported Python CLI argument(s): " + ", ".join(extra_argv))
2862
2863
2864	def _config_policy_for_args(args: argparse.Namespace, topic: str, extra_argv: list[str]) -> env.ConfigLoadPolicy:
2865	normalized_topic = topic.lower()
2866	is_library_command = (
2867	normalized_topic == "library feed"
2868	or normalized_topic == "library search"
2869	or normalized_topic.startswith("library search ")
2870	)
2871	# Queue commands are local SQLite reads/writes: like library commands they
2872	# must never trigger browser-cookie extraction or Keychain prompts.
2873	is_queue_command = (
2874	normalized_topic == "queue list"
2875	or normalized_topic == "queue cover"
2876	or normalized_topic.startswith("queue cover ")
2877	)
2878	is_cached_verification = bool(getattr(args, "verify_freshness", None)) and not normalized_topic
2879	if args.no_browser_cookies:
2880	browser_mode = "off"
2881	elif (
2882	args.diagnose or args.preflight or normalized_topic == "doctor"
2883	or is_library_command or is_queue_command or is_cached_verification
2884	):
2885	# doctor is plan-only like --diagnose: it must never read cookies.
2886	# Cache-only freshness verification hits only point APIs (Polymarket,
2887	# GitHub, StockTwits) - no cookie-backed source, so no Keychain prompt.
2888	browser_mode = "plan_only"
2889	elif normalized_topic == "setup":
2890	browser_mode = "read" if _setup_allows_browser_cookies(args, extra_argv) else "off"
2891	else:
2892	browser_mode = "read"
2893	return env.ConfigLoadPolicy(
2894	browser_cookies=browser_mode,
2895	inspect_ignored_project_config=args.diagnose or args.preflight or normalized_topic == "doctor",
2896	)
2897
2898
2899	def _run_library_feed(args: argparse.Namespace, config: dict[str, object]) -> int:
2900	"""Generate the local research index/feed and optionally publish it."""
2901	from lib import feed, html_publish, library
2902
2903	if args.publish_html:
2904	sys.stderr.write(
2905	"[last30days] library feed uses --publish, not --publish-html.\n"
2906	)
2907	return 2
2908	if args.output:
2909	sys.stderr.write(
2910	"[last30days] library feed writes index.html and feed.xml to --save-dir; "
2911	"--output is not supported.\n"
2912	)
2913	return 2
2914
2915	memory_dir = Path(args.save_dir).expanduser() if args.save_dir else library.DEFAULT_MEMORY_DIR
2916	output_dir = memory_dir.resolve()
2917	# Scoped libraries (--save-dir) must not mix in the global briefing
2918	# archive: a client-specific or publishable feed pulling unrelated default
2919	# briefings could publish them publicly. The default library keeps the
2920	# archive; a scoped one reads only its own directory.
2921	briefs_dir = (
2922	library.DEFAULT_BRIEFS_DIR if not args.save_dir else memory_dir / "briefings"
2923	)
2924	entries, notes = library.scan_library(memory_dir, briefs_dir)
2925	feed_author = str(
2926	config.get("LAST30DAYS_LIBRARY_OWNER") or "last30days research library"
2927	)
2928	output_dir.mkdir(parents=True, exist_ok=True)
2929	library_id = library.get_or_create_library_id(output_dir)
2930	rendered_briefs_dir = output_dir / "briefs"
2931	has_private_entries = any(
2932	render.PRIVATE_CORPUS_START in entry.content for entry in entries
2933	)
2934	_ensure_output_directory(rendered_briefs_dir, private=has_private_entries)
2935
2936	def _preserve_hand_written_page(existing_path: Path, generated_marker: str) -> None:
2937	"""Back up any page library feed did not generate before overwriting it."""
2938	if not existing_path.exists():
2939	return
2940	try:
2941	marker_found = generated_marker in existing_path.read_text(encoding="utf-8")
2942	except (OSError, UnicodeDecodeError):
2943	marker_found = False
2944	if marker_found:
2945	return
2946	backup = existing_path.with_suffix(existing_path.suffix + ".bak")
2947	counter = 1
2948	while backup.exists():
2949	backup = existing_path.with_suffix(f"{existing_path.suffix}.bak{counter}")
2950	counter += 1
2951	existing_path.replace(backup)
2952	sys.stderr.write(
2953	f"[last30days] {existing_path.name} was not generated by "
2954	f"library feed; preserved the original at {backup.name}\n"
2955	)
2956
2957	publishable_brief_documents: dict[str, str] = {}
2958	for entry in entries:
2959	rendered = html_render.render_library_brief(entry)
2960	target = rendered_briefs_dir / entry.output_name
2961	_preserve_hand_written_page(target, html_render.LIBRARY_BRIEF_MARKER)
2962	save_rendered_output(
2963	rendered,
2964	str(target),
2965	private=render.PRIVATE_CORPUS_START in entry.content,
2966	)
2967	publishable_brief_documents[entry.entry_id] = html_render.render_library_brief(
2968	entry, include_private=False
2969	)
2970
2971	current_brief_names = {entry.output_name for entry in entries}
2972	for path in rendered_briefs_dir.glob("*.html"):
2973	is_orphan = path.name not in current_brief_names
2974	if not (is_orphan and library.is_generated_brief_name(path.name)):
2975	continue
2976	# A generated-looking name is not proof of ownership; only prune
2977	# pages that carry the renderer's own marker.
2978	try:
2979	generated = html_render.LIBRARY_BRIEF_MARKER in path.read_text(
2980	encoding="utf-8"
2981	)
2982	except (OSError, UnicodeDecodeError):
2983	generated = False
2984	if generated:
2985	path.unlink()
2986
2987	feed_xml = feed.render_atom(entries, library_id=library_id, author=feed_author)
2988	index_html = html_render.render_library_index(entries)
2989	feed_path = output_dir / "feed.xml"
2990	index_path = output_dir / "index.html"
2991	_preserve_hand_written_page(feed_path, "urn:last30days:research-library")
2992	_preserve_hand_written_page(
2993	index_path, "Generated locally by <strong>last30days</strong>"
2994	)
2995	feed_path.write_text(feed_xml, encoding="utf-8")
2996	index_path.write_text(index_html, encoding="utf-8")
2997
2998	for note in notes:
2999	sys.stderr.write(f"[last30days] Library note: {note}\n")
3000	sys.stderr.write(
3001	f"[last30days] Library feed generated {len(entries)} brief(s): "
3002	f"{index_path} and {feed_path}\n"
3003	)
3004
3005	if args.publish:
3006	password = _publish_password_for_args(args, config)
3007	entry_urls: dict[str, str] = {}
3008	try:
3009	brief_results = html_publish.publish_html_documents(
3010	publishable_brief_documents,
3011	password=password,
3012	)
3013	entry_urls = {
3014	entry_id: str(result["url"])
3015	for entry_id, result in brief_results.items()
3016	}
3017	if batch_error := getattr(brief_results, "error", None):
3018	raise batch_error
3019	published_index = html_render.render_library_index(
3020	entries,
3021	entry_urls=entry_urls,
3022	feed_url=None,
3023	)
3024	index_result = html_publish.publish_html(published_index, password=password)
3025	index_url = str(index_result["url"])
3026	except (html_publish.HtmlPublishError, KeyError, OSError) as exc:
3027	sys.stderr.write(f"[last30days] Library publish failed: {exc}\n")
3028	if entry_urls:
3029	sys.stderr.write(
3030	f"[last30days] Partial publish: {len(entry_urls)} public brief "
3031	"page(s) were created before the failure.\n"
3032	)
3033	return 1
3034
3035	# Keep the local artifacts useful as a record of the live publication.
3036	feed_path.write_text(
3037	feed.render_atom(
3038	entries,
3039	library_id=library_id,
3040	entry_urls=entry_urls,
3041	author=feed_author,
3042	),
3043	encoding="utf-8",
3044	)
3045	index_path.write_text(
3046	html_render.render_library_index(entries, entry_urls=entry_urls),
3047	encoding="utf-8",
3048	)
3049	sys.stderr.write(f"[last30days] Published library to {index_url}\n")
3050	sys.stderr.write(f"[last30days] Local Atom feed: {feed_path}\n")
3051	print(
3052	f"Library: {index_url}\nFeed: {feed_path}\n"
3053	"Atom feed is local; host feed.xml on any static host (for example, GitHub Pages) "
3054	"to make it subscribable."
3055	)
3056	return 0
3057
3058	print(
3059	f"Library: {index_path}\nFeed: {feed_path}\n"
3060	"Atom feed is local; host feed.xml on any static host (for example, GitHub Pages) "
3061	"to make it subscribable."
3062	)
3063	return 0
3064
3065
3066	def _run_library_search(
3067	args: argparse.Namespace,
3068	config: dict[str, object],
3069	query: str,
3070	) -> int:
3071	"""Search saved briefs and store sightings without network access."""
3072	from lib import library, library_index
3073
3074	if not query.strip():
3075	sys.stderr.write("[last30days] library search requires a non-empty query.\n")
3076	return 2
3077	if args.publish or args.publish_html:
3078	sys.stderr.write("[last30days] library search does not publish output.\n")
3079	return 2
3080	if args.emit != "compact":
3081	sys.stderr.write("[last30days] library search currently supports text output only.\n")
3082	return 2
3083	if args.output:
3084	sys.stderr.write(
3085	"[last30days] library search prints to stdout; --output is not supported.\n"
3086	)
3087	return 2
3088
3089	memory_dir = Path(args.save_dir).expanduser() if args.save_dir else library.DEFAULT_MEMORY_DIR
3090	try:
3091	matches, synced = library_index.sync_and_search(
3092	query,
3093	memory_dir=memory_dir,
3094	briefs_dir=(
3095	memory_dir / "briefings" if args.save_dir else library.DEFAULT_BRIEFS_DIR
3096	),
3097	db_path=(
3098	memory_dir.resolve() / ".last30days-library.db"
3099	if args.save_dir else library_index.DEFAULT_LIBRARY_DB
3100	),
3101	# A scoped search must never merge in the shared store: one
3102	# client's sightings would leak into another client's scope. A
3103	# scoped store is read only if it exists inside the save dir.
3104	store_db_path=(
3105	memory_dir.resolve() / "research.db"
3106	if args.save_dir else library_index.DEFAULT_STORE_DB
3107	),
3108	)
3109	except library_index.LibrarySearchUnavailable as exc:
3110	sys.stderr.write(f"[last30days] Library search unavailable: {exc}.\n")
3111	return 2
3112	except (OSError, sqlite3.DatabaseError) as exc:
3113	sys.stderr.write(f"[last30days] Library search failed: {exc}.\n")
3114	return 1
3115	for note in synced.notes:
3116	sys.stderr.write(f"[last30days] Library note: {note}\n")
3117	if synced.rebuilt:
3118	sys.stderr.write("[last30days] Rebuilt a corrupt library search index.\n")
3119	print(render.render_library_search(query, matches), end="")
3120	return 0
3121
3122
3123	def _looks_like_entity_topic(topic: str) -> bool:
3124	"""Whether a topic names a person, company, or product rather than a theme.
3125
3126	Keys on brevity, not capitalization. People type lowercase: "bentgo",
3127	"peter steinberger" and "getenergy.com" are entity searches every bit as
3128	much as their title-cased forms, and requiring a capital meant the most
3129	common real-world spelling never resolved a handle.
3130
3131	A short topic is an entity search; a longer one is a theme. "Peter
3132	Steinberger", "bentgo" and "getenergy.com" qualify; "best AI coding tools
3133	2026" and "how to build agents that scale" do not. Question-shaped topics
3134	are themes regardless of length.
3135
3136	Used only to decide whether resolving an X handle is worth one web search,
3137	so a false negative costs the old behavior and a false positive costs a
3138	single search.
3139	"""
3140	text = (topic or "").strip()
3141	if not text or text.endswith("?"):
3142	return False
3143	words = [w for w in re.findall(r"[A-Za-z0-9_.@'-]+", text) if w]
3144	if not words or len(words) > 4:
3145	return False
3146	if any(w.startswith("@") for w in words):
3147	return True
3148	# A theme reads as a phrase built from common words; an entity does not.
3149	common = {
3150	"best", "top", "how", "why", "what", "when", "vs", "versus", "guide",
3151	"tips", "review", "reviews", "news", "latest", "update", "updates",
3152	"trends", "tools", "and", "or", "for", "the", "with", "about",
3153	}
3154	return not any(w.lower() in common for w in words)
3155
3156
3157	def main() -> int:
3158	parser = build_parser()
3159	# Use parse_known_args so setup sub-flags (--device-auth, --github,
3160	# --openclaw) pass through without argparse hard-exiting.
3161	args, extra_argv = parser.parse_known_args()
3162	if args.record_fixtures:
3163	with http.recording_requests(Path(args.record_fixtures)):
3164	return _main(parser, args, extra_argv)
3165	return _main(parser, args, extra_argv)
3166
3167
3168	def _main(
3169	parser: argparse.ArgumentParser,
3170	args: argparse.Namespace,
3171	extra_argv: list[str],
3172	) -> int:
3173	if args.debug:
3174	os.environ["LAST30DAYS_DEBUG"] = "1"
3175
3176	if args.welcome:
3177	from lib import setup_wizard
3178	print(setup_wizard.render_welcome())
3179	return 0
3180
3181	topic = " ".join(args.topic).strip()
3182	original_topic = topic
3183	_validate_extra_argv(parser, topic, extra_argv)
3184	if args.x_posts is not None and _looks_inline_json(args.x_posts):
3185	sys.stderr.write(
3186	"[last30days] --x-posts accepts a file path only (inline JSON is not "
3187	"accepted); write the envelope to a .json file and pass its path.\n"
3188	)
3189	return 2
3190	if args.publish and topic.lower() != "library feed":
3191	sys.stderr.write(
3192	"[last30days] --publish is only supported by the 'library feed' command.\n"
3193	)
3194	return 2
3195	if topic.lower() == "setup":
3196	# Persisting a credential needs no config load (no Keychain / pass
3197	# probes, no cookie policy), so it dispatches before get_config.
3198	store_key_present, store_key_name, _ = _split_store_key(extra_argv)
3199	if store_key_present:
3200	return _run_store_key(store_key_name)
3201
3202	config = env.get_config(policy=_config_policy_for_args(args, topic, extra_argv))
3203	# One memo per command: comparison mode runs pipeline.run per entity in
3204	# parallel, so the reset must not live inside the pipeline.
3205	http.reset_reddit_keyless_memo()
3206	resolved_corpus_dirs = corpus.resolve_directories(
3207	args.corpus, config.get("LAST30DAYS_CORPUS_DIRS")
3208	)
3209	# EXCLUDE_SOURCES=corpus disables corpus retrieval entirely; the hosted
3210	# privacy bypass below must use the same predicate, or hosted users with
3211	# configured-but-excluded dirs silently lose the remote backend.
3212	excluded_sources = {
3213	value.strip().lower()
3214	for value in str(config.get("EXCLUDE_SOURCES") or "").split(",")
3215	if value.strip()
3216	}
3217	if "corpus" in excluded_sources:
3218	resolved_corpus_dirs = []
3219	if resolved_corpus_dirs:
3220	config["_CORPUS_DIRS"] = [str(path) for path in resolved_corpus_dirs]
3221	if _config_truthy(config.get("LAST30DAYS_CORPUS_IN_EXPORT")):
3222	config["_CORPUS_IN_EXPORT"] = True
3223	_propagate_config_to_environ(config)
3224
3225	# Env-var fallback for --save-dir, mirroring the LAST30DAYS_STORE pattern below.
3226	# Uses `is None` / `is not None` checks (not truthy `or`) at every layer so that
3227	# `--save-dir ""`, `LAST30DAYS_MEMORY_DIR=""` (shell-export-empty), and explicit
3228	# absence each correctly suppress save. An `or` chain would collapse the empty
3229	# shell-export into the same path as unset, silently falling through to .env.
3230	if args.save_dir is None:
3231	env_val = os.environ.get("LAST30DAYS_MEMORY_DIR")
3232	args.save_dir = env_val if env_val is not None else config.get("LAST30DAYS_MEMORY_DIR")
3233
3234	# Surface SSH-routing config as an env var so library modules (e.g.
3235	# youtube_yt) can read it without taking a config dependency. This
3236	# routes yt-dlp through `ssh <host>` to bypass YouTube's bot-wall on
3237	# datacenter IPs (see lib/youtube_yt.py for details).
3238	if config.get("LAST30DAYS_YOUTUBE_SSH_HOST") and "LAST30DAYS_YOUTUBE_SSH_HOST" not in os.environ:
3239	os.environ["LAST30DAYS_YOUTUBE_SSH_HOST"] = config["LAST30DAYS_YOUTUBE_SSH_HOST"]
3240
3241	if args.preflight:
3242	requested_sources = resolve_requested_sources(args.search, config)
3243	diag = pipeline.diagnose(config, requested_sources, safe=True)
3244	if args.save_dir or args.preflight_report_on_save_dir:
3245	preflight = permission_preflight.build(
3246	config,
3247	diag,
3248	planned_save_dir=args.save_dir,
3249	report_on_save_dir=args.preflight_report_on_save_dir,
3250	)
3251	else:
3252	preflight = diag["permission_preflight"]
3253	if args.emit == "json":
3254	print(json.dumps(preflight, indent=2, sort_keys=True))
3255	else:
3256	print(permission_preflight.render_text(preflight), end="")
3257	return 0
3258
3259	# Handle doctor subcommand: topic-word dispatch mirroring setup (exact
3260	# match only, so multi-word research topics containing "doctor" still
3261	# research normally). Aggregates probes/descriptors/prescriptions into
3262	# one grouped health surface; always exits 0.
3263	if topic.lower() == "doctor":
3264	from lib import doctor
3265	return doctor.run(
3266	config,
3267	emit_json=(args.emit == "json" or "--json" in extra_argv),
3268	cached="--cached" in extra_argv,
3269	postmortem="--postmortem" in extra_argv,
3270	probe="--probe" in extra_argv,
3271	)
3272
3273	if topic.lower() == "library feed":
3274	return _run_library_feed(args, config)
3275	if topic.lower() == "library search" or topic.lower().startswith("library search "):
3276	return _run_library_search(args, config, topic[len("library search") :].strip())
3277
3278	if topic.lower() == "queue list":
3279	return _run_queue_list(args, config)
3280	if topic.lower() == "queue cover" or topic.lower().startswith("queue cover "):
3281	return _run_queue_cover(args, config, topic[len("queue cover") :].strip())
3282
3283	# Handle setup subcommand
3284	if topic.lower() == "setup":
3285	from lib import setup_wizard
3286	if "--openclaw" in extra_argv:
3287	results = setup_wizard.run_openclaw_setup(config)
3288	print(json.dumps(results))
3289	return 0
3290	if any(f in extra_argv for f in ("--github", "--device-auth", "--github-start", "--github-poll")):
3291	if "--github-start" in extra_argv:
3292	results = setup_wizard.run_github_start()
3293	elif "--github-poll" in extra_argv:
3294	results = setup_wizard.run_github_poll()
3295	elif "--github" in extra_argv:
3296	results = setup_wizard.run_github_auth()
3297	else:
3298	results = setup_wizard.run_full_device_auth()
3299	# Persist the returned key so the paid sources activate on the next
3300	# run, and mask it in stdout so the secret never lands in the host
3301	# model's captured Bash output.
3302	api_key = results.get("api_key")
3303	status = results.get("status")
3304	if api_key:
3305	if status == "success":
3306	results["persisted"] = setup_wizard.write_api_key(env.CONFIG_FILE, api_key)
3307	elif status == "already_registered":
3308	results["persisted"] = True # key was already saved
3309	else:
3310	results.setdefault("persisted", False)
3311	# Mask for EVERY status that carries a key, not just success, so
3312	# the raw secret never reaches the host model's captured stdout.
3313	results["api_key"] = setup_wizard.mask_api_key(api_key)
3314	else:
3315	results["persisted"] = False
3316	print(json.dumps(results))
3317	return 0
3318	sys.stderr.write("Running auto-setup...\n")
3319	results = setup_wizard.run_auto_setup(
3320	config,
3321	allow_browser_cookies=_setup_allows_browser_cookies(args, extra_argv),
3322	)
3323	# Persist FROM_BROWSER only when every service's cookies came from the
3324	# SAME single browser — then we can fast-path future runs to it. If
3325	# different services matched different browsers, or none matched, leave
3326	# FROM_BROWSER unset so the safe default remains no browser-cookie
3327	# reads. We deliberately do NOT pin "auto" here (it would re-probe
3328	# Chrome and re-trigger the prompt) nor a single browser (it would
3329	# silently skip the service that used the other one).
3330	found_browsers = set(results.get("cookies_found", {}).values())
3331	from_browser = found_browsers.pop() if len(found_browsers) == 1 else None
3332	# Pin only a silent winner (firefox/safari). Pinning a Chromium browser
3333	# would make every steady-state run re-read its Keychain-encrypted store
3334	# and can re-trigger the "Always Allow" prompt, so Chrome is used for the
3335	# first-run scan but never pinned.
3336	if from_browser in {"chrome", "brave", "edge", "vivaldi", "opera", "arc", "chromium"}:
3337	from_browser = None
3338	setup_wizard.write_setup_config(env.CONFIG_FILE, from_browser=from_browser)
3339	results["env_written"] = True
3340	sys.stderr.write(setup_wizard.get_setup_status_text(results) + "\n")
3341	return 0
3342
3343	# Bare --discover (no domain) is global trending, so the dispatch keys on
3344	# "flag present" (is not None), never on the domain string's truthiness.
3345	if args.deep_research and not topic:
3346	sys.stderr.write(
3347	"[last30days] --deep-research requires a normal positional topic; "
3348	"it cannot be combined with discovery, drill, or cached-only modes.\n"
3349	)
3350	return 2
3351
3352	if args.discover is not None:
3353	if topic:
3354	sys.stderr.write(
3355	"[last30days] --discover supplies the domain and cannot be combined "
3356	"with a positional topic.\n"
3357	)
3358	return 2
3359	if args.drill:
3360	sys.stderr.write("[last30days] --discover and --drill are mutually exclusive.\n")
3361	return 2
3362	# Shared guards for EVERY discover invocation - the one-shot and all
3363	# three protocol legs - hoisted here so no leg can drift: discovery
3364	# sweeps live listings (never --as-of) and has no HTML pipeline yet.
3365	if args.as_of_date:
3366	sys.stderr.write(
3367	"[last30days] --as-of cannot be used with --discover because discovery "
3368	"sweeps current live listings.\n"
3369	)
3370	return 2
3371	if args.emit == "html" or args.publish_html:
3372	sys.stderr.write("[last30days] discovery mode does not support HTML publishing yet.\n")
3373	return 2
3374	# The three protocol legs are one-leg-per-invocation: each pairing
3375	# below asks for two legs at once, so name the combination and stop.
3376	# (--judgments/--angles dispatch on presence, never path truthiness.)
3377	for first, second, conflict in (
3378	("--nominate-only", "--judgments", args.nominate_only and args.judgments is not None),
3379	("--nominate-only", "--finalize", args.nominate_only and args.finalize),
3380	("--judgments", "--finalize", args.judgments is not None and args.finalize),
3381	):
3382	if conflict:
3383	sys.stderr.write(
3384	f"[last30days] {first} and {second} are mutually exclusive: "
3385	"each runs a different leg of the discovery protocol.\n"
3386	)
3387	return 2
3388	if args.angles is not None and not args.finalize:
3389	sys.stderr.write(
3390	"[last30days] --angles only applies to --discover --finalize "
3391	"runs; add --finalize or drop the flag.\n"
3392	)
3393	return 2
3394	protocol_leg = (
3395	args.nominate_only or args.judgments is not None or args.finalize
3396	)
3397	if protocol_leg and args.mock and not args.save_dir:
3398	# Truthiness is right here: an empty --save-dir/env value means
3399	# "no save dir", and handoff state would land in the real config
3400	# dir - a side effect mock runs must never have.
3401	sys.stderr.write(
3402	"[last30days] mock protocol legs require --save-dir to stay "
3403	"side-effect-free: --mock with --nominate-only/--judgments/"
3404	"--finalize would otherwise write handoff state into the real "
3405	"config dir.\n"
3406	)
3407	return 2
3408	if protocol_leg:
3409	return _run_discover_protocol_leg(args, config)
3410	return _run_discover(args, config)
3411
3412	if args.discover_shallow:
3413	# Without --discover this flag would silently no-op into a full
3414	# research run - reject it instead of ignoring the requested mode.
3415	sys.stderr.write(
3416	"[last30days] --discover-shallow only applies to --discover runs; "
3417	"add --discover [domain] or drop the flag.\n"
3418	)
3419	return 2
3420
3421	# Same orphan rule for every protocol-leg flag: without --discover each
3422	# would silently no-op into a normal research run.
3423	for flag_label, present in (
3424	("--nominate-only", args.nominate_only),
3425	("--judgments", args.judgments is not None),
3426	("--finalize", args.finalize),
3427	):
3428	if present:
3429	sys.stderr.write(
3430	f"[last30days] {flag_label} only applies to --discover runs; "
3431	"add --discover [domain] or drop the flag.\n"
3432	)
3433	return 2
3434	if args.angles is not None:
3435	sys.stderr.write(
3436	"[last30days] --angles only applies to --discover --finalize runs; "
3437	"add --discover --finalize or drop the flag.\n"
3438	)
3439	return 2
3440
3441	if args.drill:
3442	if topic:
3443	sys.stderr.write(
3444	"[last30days] --drill uses the cached topic and cannot be "
3445	"combined with a new topic.\n"
3446	)
3447	return 2
3448	if args.publish_html and args.emit != "html":
3449	sys.stderr.write("[last30days] --publish-html requires --emit=html\n")
3450	return 2
3451	if args.dedicated_subreddits:
3452	config["_dedicated_subreddits"] = [
3453	value.strip().removeprefix("r/")
3454	for value in args.dedicated_subreddits.split(",")
3455	if value.strip()
3456	]
3457	if args.polymarket_keywords:
3458	config["_polymarket_keywords"] = [
3459	value.strip().lower()
3460	for value in args.polymarket_keywords.split(",")
3461	if value.strip()
3462	]
3463	return _run_drill(args, config)
3464
3465	if args.verify_freshness and not topic:
3466	return _run_cached_freshness(args, config)
3467
3468	if args.lookback_days is None:
3469	args.lookback_days = 30
3470
3471	if args.deep_research and not args.diagnose:
3472	from lib import planner as _planner
3473
3474	if not (
3475	config.get("PERPLEXITY_API_KEY")
3476	or config.get("OPENROUTER_API_KEY")
3477	):
3478	print(
3479	"Error: --deep-research requires PERPLEXITY_API_KEY or "
3480	"OPENROUTER_API_KEY",
3481	file=sys.stderr,
3482	)
3483	return 1
3484	comparison_requested = any(
3485	value is not None
3486	for value in (
3487	args.competitors,
3488	args.competitors_list,
3489	args.competitors_plan,
3490	)
3491	) or len(_planner._comparison_entities(topic, uncapped=True)) >= 2
3492	if comparison_requested:
3493	sys.stderr.write(
3494	"Error: --deep-research cannot be combined with competitor or vs-mode. "
3495	"It permits one paid Deep Research run per user action; run each topic "
3496	"separately.\n"
3497	)
3498	return 2
3499	config["_deep_research"] = True
3500	try:
3501	enable_deep_research_source(config)
3502	except ValueError as exc:
3503	print(f"Error: {exc}", file=sys.stderr)
3504	return 2
3505
3506	# Reject a misspelled configured register before remote submission or any
3507	# local source retrieval. Excluded modes resolve to default and remain
3508	# unaffected by the register setting.
3509	try:
3510	_audience_register_for_run(args, config, None)
3511	except ValueError as exc:
3512	sys.stderr.write(f"[last30days] {exc}\n")
3513	return 2
3514
3515	# Remote API path: when BOTH LAST30DAYS_API_KEY and LAST30DAYS_API_BASE are
3516	# set (and --mock is not), the search runs through the configured remote API
3517	# instead of local sources; no local provider keys are needed (see
3518	# lib/hosted.py). With either env var unset, behavior below is byte-identical
3519	# to local-only runs - there is no built-in endpoint.
3520	if (
3521	topic
3522	and resolved_corpus_dirs
3523	and env.read_secret_env("LAST30DAYS_API_KEY")
3524	and os.environ.get("LAST30DAYS_API_BASE")
3525	):
3526	sys.stderr.write(
3527	"[last30days] Local corpus configured; bypassing the hosted backend so files stay on this machine.\n"
3528	)
3529	if (
3530	topic
3531	and not args.diagnose
3532	and not args.mock
3533	and not args.record_fixtures
3534	and env.read_secret_env("LAST30DAYS_API_KEY")
3535	and os.environ.get("LAST30DAYS_API_BASE")
3536	and not resolved_corpus_dirs
3537	and not args.deep_research
3538	):
3539	if _freshness_enabled(args, config):
3540	if args.verify_freshness is True:
3541	sys.stderr.write(
3542	"[last30days] Freshness verification is not supported by the hosted backend; "
3543	"run locally or omit --verify-freshness.\n"
3544	)
3545	return 2
3546	sys.stderr.write(
3547	"hosted backend does not support freshness verification; skipping\n"
3548	)
3549	if args.emit == "json" and args.json_profile == "agent":
3550	sys.stderr.write(
3551	"[last30days] --json-profile=agent requires the local Report; "
3552	"the remote API backend only supports --json-profile=raw.\n"
3553	)
3554	return 2
3555	if args.x_posts is not None:
3556	# The envelope is a local-engine contract; the remote API has no
3557	# lane to receive it.
3558	sys.stderr.write(
3559	"[last30days] --x-posts is not supported by the hosted backend; "
3560	"run locally or omit --x-posts.\n"
3561	)
3562	return 2
3563	from lib import hosted
3564	depth = "deep" if args.deep else "quick" if args.quick else "default"
3565	try:
3566	audience = _audience_register_for_run(args, config, None)
3567	except ValueError as exc:
3568	sys.stderr.write(f"[last30days] {exc}\n")
3569	return 2
3570	hosted_kwargs = {
3571	"emit": args.emit,
3572	"save_dir": args.save_dir,
3573	"save_suffix": args.save_suffix or "",
3574	}
3575	if audience.name != "default":
3576	hosted_kwargs["register"] = audience.name
3577	return hosted.run_hosted(topic, depth, **hosted_kwargs)
3578
3579	requested_sources = resolve_requested_sources(args.search, config)
3580	if args.deep_research:
3581	requested_sources = add_deep_research_source(requested_sources)
3582	# Explicit --trustpilot-domain is user intent: activate the opt-in source
3583	# before diagnose/run so the flag cannot silently no-op (#873). Auto-resolve
3584	# hints are applied later and must not call this path.
3585	cli_trustpilot_domain = (
3586	args.trustpilot_domain.strip() if args.trustpilot_domain else ""
3587	)
3588	if cli_trustpilot_domain:
3589	requested_sources = activate_trustpilot_for_explicit_domain(
3590	config,
3591	requested_sources,
3592	reason=f"--trustpilot-domain={cli_trustpilot_domain}",
3593	)
3594	# Explicit --telegram-sources is user intent: activate the opt-in source
3595	# before diagnose/run so the flag cannot silently no-op (same pattern as
3596	# Trustpilot #873). Sets TELEGRAM_SOURCES in config for pipeline.
3597	cli_telegram_sources = (
3598	args.telegram_sources.strip() if args.telegram_sources else ""
3599	)
3600	if cli_telegram_sources:
3601	requested_sources = activate_telegram_for_explicit_sources(
3602	config,
3603	requested_sources,
3604	channels=cli_telegram_sources,
3605	)
3606	# Host-fetched X envelope: validated before diagnose so a present
3607	# envelope plans X in (available_sources) and a bad one fails closed here.
3608	x_posts_envelope: x_envelope.Envelope \| None = None
3609	if args.x_posts is not None:
3610	if not topic:
3611	sys.stderr.write("[last30days] --x-posts requires a research topic.\n")
3612	return 2
3613	if _comparison_requested(args, topic):
3614	sys.stderr.write(
3615	"[last30days] --x-posts applies to a single-topic run; on a "
3616	"comparison run pass each entity's envelope through the "
3617	"x_posts field of its --competitors-plan entry.\n"
3618	)
3619	return 2
3620	try:
3621	x_posts_envelope = _read_x_envelope(
3622	args.x_posts, topic, args,
3623	x_handle=args.x_handle,
3624	x_related=args.x_related.split(",") if args.x_related else None,
3625	)
3626	except x_envelope.EnvelopeContractError as exc:
3627	sys.stderr.write(f"[last30days] {exc.message}\n")
3628	return 2
3629	diag = pipeline.diagnose(
3630	config, requested_sources, safe=args.diagnose,
3631	x_envelope=x_posts_envelope is not None,
3632	)
3633
3634	if args.diagnose:
3635	print(json.dumps(diag, indent=2, sort_keys=True))
3636	return 0
3637
3638	# Competitor sub-runs shallow-copy this config. The shared object makes the
3639	# paid Perplexity cap command-wide and thread-safe across that fanout. Keep
3640	# this runtime-only object out of the safe diagnose configuration contract.
3641	config["_perplexity_paid_budget"] = pipeline.PaidSourceBudget()
3642
3643	# Per-entity host-fetched X envelopes are validated here, on the main
3644	# thread and BEFORE the report-cache lookup, so a bad or stale one fails
3645	# closed (exit 2) instead of silently dropping that entity inside the
3646	# fan-out or being served from a cache built while it was still valid.
3647	comp_plan = parse_competitors_plan(args.competitors_plan)
3648	try:
3649	_attach_entity_envelopes(comp_plan, args)
3650	except x_envelope.EnvelopeContractError as exc:
3651	sys.stderr.write(f"[last30days] {exc.message}\n")
3652	return 2
3653
3654	if not topic:
3655	parser.print_usage(sys.stderr)
3656	return 2
3657	if args.publish_html and args.emit != "html":
3658	sys.stderr.write("[last30days] --publish-html requires --emit=html\n")
3659	return 2
3660
3661	synthesis_md = None
3662	if args.synthesis_file:
3663	if args.emit == "html":
3664	synthesis_md = read_synthesis_file(args.synthesis_file)
3665	else:
3666	sys.stderr.write("[last30days] Warning: --synthesis-file is only used with --emit=html; ignoring.\n")
3667
3668	if not os.environ.get("LAST30DAYS_SKIP_PREFLIGHT"):
3669	from lib import preflight
3670	refuse_msg = preflight.check_class_1_trap(topic)
3671	if refuse_msg:
3672	sys.stderr.write(refuse_msg)
3673	return 2
3674
3675	if (
3676	args.emit == "html"
3677	and synthesis_md is not None
3678	and not args.deep_research
3679	):
3680	cached = _load_last_report_cache(
3681	topic,
3682	ttl_seconds=_report_cache_ttl_seconds(config),
3683	x_envelope_sha256=_x_envelope_digest(x_posts_envelope, comp_plan),
3684	)
3685	if cached is not None:
3686	cached_report, cached_entity_reports, cache_path = cached
3687	sys.stderr.write(
3688	f"[last30days] Reusing cached report data from {cache_path}\n"
3689	)
3690	sys.stderr.flush()
3691	if _freshness_enabled(args, config):
3692	_verify_report_set(
3693	cached_report,
3694	cached_entity_reports,
3695	allow_network=not args.mock,
3696	)
3697	_update_cached_freshness(
3698	cache_path,
3699	cached_report,
3700	cached_entity_reports,
3701	)
3702	return _render_save_and_print(
3703	args, cached_report, cached_entity_reports, synthesis_md, config
3704	)
3705	sys.stderr.write(
3706	"[last30days] No matching cached report data for "
3707	"--emit=html --synthesis-file; running fresh research.\n"
3708	)
3709	sys.stderr.flush()
3710
3711	progress = ui.ProgressDisplay(topic, show_banner=True)
3712	progress.start_processing()
3713
3714	depth = "deep" if args.deep else "quick" if args.quick else "default"
3715	# CLI overrides for the depth profile's result caps (issue #716). Stashed on
3716	# config so pipeline.run() can apply them without widening its signature; the
3717	# comparison path inherits them via `entity_config = dict(config)`.
3718	if args.max_results is not None:
3719	config["_max_results"] = args.max_results
3720	if args.max_per_source is not None:
3721	config["_max_per_source"] = args.max_per_source
3722	if args.max_source_fetches is not None:
3723	config["_max_source_fetches"] = args.max_source_fetches
3724	try:
3725	x_related = [h.strip() for h in args.x_related.split(",") if h.strip()] if args.x_related else None
3726	subreddits = [s.strip().removeprefix("r/") for s in args.subreddits.split(",") if s.strip()] if args.subreddits else None
3727	dedicated_subreddits = [s.strip().removeprefix("r/") for s in args.dedicated_subreddits.split(",") if s.strip()] if args.dedicated_subreddits else None
3728	tiktok_hashtags = [h.strip().lstrip("#") for h in args.tiktok_hashtags.split(",") if h.strip()] if args.tiktok_hashtags else None
3729	tiktok_creators = [c.strip().lstrip("@") for c in args.tiktok_creators.split(",") if c.strip()] if args.tiktok_creators else None
3730	ig_creators = [c.strip().lstrip("@") for c in args.ig_creators.split(",") if c.strip()] if args.ig_creators else None
3731	# Parse external plan if provided via --plan flag
3732	external_plan = None
3733	if args.plan:
3734	import json as _json
3735	plan_str = args.plan
3736	if os.path.isfile(plan_str):
3737	try:
3738	with open(plan_str, encoding="utf-8") as f:
3739	plan_str = f.read()
3740	except (OSError, UnicodeDecodeError) as exc:
3741	sys.stderr.write(f"[Planner] Cannot read --plan file: {exc}\n")
3742	raise SystemExit(2)
3743	try:
3744	external_plan = _json.loads(plan_str)
3745	except _json.JSONDecodeError as exc:
3746	sys.stderr.write(f"[Planner] Invalid --plan JSON: {exc}\n")
3747	# Fail fast instead of silently dropping to the internal planner
3748	# and burning a paid run the user did not ask for. Mirrors the
3749	# --plan file-read branch above and parse_competitors_plan.
3750	raise SystemExit(2)
3751	from lib import planner as _plan_validator
3752	try:
3753	_plan_validator.validate_external_plan(external_plan)
3754	except ValueError as exc:
3755	sys.stderr.write(f"[Planner] Invalid --plan schema: {exc}.\n")
3756	raise SystemExit(2)
3757
3758	# Auto-resolve: use web search to discover subreddits/handles before planning.
3759	# This is the engine-side equivalent of SKILL.md Steps 0.55/0.75 for platforms
3760	# without WebSearch (OpenClaw, Codex, raw CLI).
3761	repos_from_auto_resolve = False
3762	trustpilot_domain_is_hint = False
3763	# Resolve automatically for entity-shaped topics even without the flag.
3764	# A person or company topic whose handle the user did not supply is the
3765	# case where first-party evidence is hardest to protect: the handle is
3766	# absent from the topic and may never appear in retrieved mentions, so
3767	# nothing downstream can identify the subject's own posts. One web
3768	# search closes that. If it returns nothing, pipeline.run skips the X
3769	# relevance floor entirely — a noisier report beats losing evidence.
3770	# Skipped when a handle was already supplied, when an external plan
3771	# owns resolution, or in mock runs.
3772	if (
3773	not args.auto_resolve
3774	and not external_plan
3775	and not args.x_handle
3776	and not args.mock
3777	and _looks_like_entity_topic(topic)
3778	):
3779	args.auto_resolve = True
3780	sys.stderr.write(
3781	"[AutoResolve] entity-shaped topic with no --x-handle; "
3782	"resolving the subject's handle so its own posts are not pruned\n"
3783	)
3784
3785	if args.auto_resolve and not external_plan:
3786	from lib import resolve
3787	resolution = resolve.auto_resolve(topic, config)
3788	if resolution.get("subreddits") and not subreddits:
3789	subreddits = resolution["subreddits"]
3790	sys.stderr.write(f"[AutoResolve] Subreddits: {', '.join(subreddits)}\n")
3791	if resolution.get("x_handle") and not args.x_handle:
3792	args.x_handle = resolution["x_handle"]
3793	sys.stderr.write(f"[AutoResolve] X handle: @{args.x_handle}\n")
3794	# Empty x_handle is intentional: do not invent a lexical stand-in.
3795	# pipeline.run treats an unidentified subject as "skip the X floor".
3796	if resolution.get("github_user") and not args.github_user:
3797	args.github_user = resolution["github_user"]
3798	sys.stderr.write(f"[AutoResolve] GitHub user: @{args.github_user}\n")
3799	if resolution.get("github_repos") and not args.github_repo:
3800	args.github_repo = ",".join(resolution["github_repos"])
3801	# auto_resolve already canonicalized via canonicalize_github_repos(cap=5);
3802	# mark so we don't re-canonicalize below and clobber its relevance order.
3803	repos_from_auto_resolve = True
3804	sys.stderr.write(f"[AutoResolve] GitHub repos: {args.github_repo}\n")
3805	if resolution.get("trustpilot_domain") and not args.trustpilot_domain:
3806	# Hint provenance matters: only user-set flags are verbatim-final;
3807	# a resolved hint retries via the CLI search when it misses.
3808	args.trustpilot_domain = resolution["trustpilot_domain"]
3809	trustpilot_domain_is_hint = True
3810	sys.stderr.write(f"[AutoResolve] Trustpilot domain: {args.trustpilot_domain} (hint)\n")
3811	if resolution.get("context"):
3812	# Inject context into external_plan metadata for the planner to use
3813	if not external_plan:
3814	external_plan = None # planner will use its own, but with context
3815	# Store context for the planner prompt injection
3816	config["_auto_resolve_context"] = resolution["context"]
3817	sys.stderr.write(f"[AutoResolve] Context: {resolution['context'][:80]}...\n")
3818
3819	github_user = args.github_user.lstrip("@").lower() if args.github_user else None
3820	github_repos = [r.strip() for r in args.github_repo.split(",") if r.strip() and "/" in r.strip()] if args.github_repo else None
3821	trustpilot_domain = args.trustpilot_domain.strip() if args.trustpilot_domain else None
3822
3823	comp_enabled, comp_count, comp_explicit = resolve_competitors_args(args)
3824	# comp_plan was parsed, and its per-entity envelopes validated, before
3825	# the report-cache lookup above.
3826
3827	# Plan-level trustpilot_domain pins are the same user intent as the CLI
3828	# flag (already activated above). Auto-resolve hints must not activate.
3829	if plan_has_explicit_trustpilot_domain(comp_plan):
3830	requested_sources = activate_trustpilot_for_explicit_domain(
3831	config,
3832	requested_sources,
3833	reason="competitors-plan trustpilot_domain",
3834	)
3835
3836	# Only canonicalize when repos came from a user-supplied --github-repo flag.
3837	# When repos_from_auto_resolve is True, auto_resolve already ran
3838	# canonicalize_github_repos(cap=5) and ranked by relevance; re-running here
3839	# with cap=None can re-sort by topic-slug match and lose that ordering.
3840	if github_repos and not repos_from_auto_resolve:
3841	from lib import resolve as resolve_lib
3842	original_github_repos = github_repos[:]
3843	github_repos = resolve_lib.canonicalize_github_repos(topic, github_repos, cap=None)
3844	if github_repos != original_github_repos:
3845	sys.stderr.write(
3846	"[GitHub] Canonicalized repos: "
3847	f"{','.join(original_github_repos)} -> {','.join(github_repos)}\n"
3848	)
3849
3850	# Polymarket disambiguation: if user passed --polymarket-keywords,
3851	# store on config so the polymarket adapter can filter matches.
3852	if args.polymarket_keywords:
3853	keywords = [
3854	k.strip().lower()
3855	for k in args.polymarket_keywords.split(",")
3856	if k.strip()
3857	]
3858	if keywords:
3859	config["_polymarket_keywords"] = keywords
3860
3861	# Product keyword for the amazon source. Carried on config rather than
3862	# threaded through the run signature (the _polymarket_keywords idiom):
3863	# it is one optional string consumed in exactly two places.
3864	if getattr(args, "amazon_query", None):
3865	config["_amazon_query"] = args.amazon_query.strip()
3866	# Unlike --trustpilot-domain, this flag deliberately does NOT
3867	# auto-activate its source: the lane spends metered credits, so
3868	# turning it on stays an explicit request. But silence is the
3869	# wrong failure mode -- a model that resolves the keyword and
3870	# forgets the --search token would otherwise get no signal at
3871	# all that the flag did nothing.
3872	_amazon_requested = (
3873	(requested_sources and "amazon" in requested_sources)
3874	or "amazon" in str(config.get("INCLUDE_SOURCES") or "").lower()
3875	)
3876	if not _amazon_requested:
3877	sys.stderr.write(
3878	"[Amazon] --amazon-query was set but the amazon source was not "
3879	"requested; add it to --search (e.g. --search reddit,x,amazon) "
3880	"or set INCLUDE_SOURCES=amazon. Ignoring the keyword.\n"
3881	)
3882
3883	# Advertiser page override for the meta_ads source. Same shape as
3884	# --amazon-query (config-carried, warn-not-activate) and for the same
3885	# reason: the lane spends metered credits per call.
3886	if getattr(args, "meta_ads_page", None):
3887	page_id = parse_meta_ads_page(args.meta_ads_page)
3888	if not page_id:
3889	sys.stderr.write(
3890	"[Meta Ads] --meta-ads-page must be a numeric Ad Library page id "
3891	"or an Ad Library URL containing view_all_page_id; a facebook.com "
3892	"vanity URL is not a page id. Ignoring the override.\n"
3893	)
3894	else:
3895	config["_meta_ads_page"] = page_id
3896	_meta_ads_requested = (
3897	(requested_sources and "meta_ads" in requested_sources)
3898	or "meta_ads" in str(config.get("INCLUDE_SOURCES") or "").lower()
3899	)
3900	if not _meta_ads_requested:
3901	sys.stderr.write(
3902	"[Meta Ads] --meta-ads-page was set but the meta_ads source "
3903	"was not requested; add it to --search (e.g. --search "
3904	"reddit,x,meta_ads) or set INCLUDE_SOURCES=meta_ads. "
3905	"Ignoring the page.\n"
3906	)
3907
3908	# vs-mode / plan routing: split a vs-topic into main + peers unless
3909	# discover-N or an explicit --competitors-list already decided who runs.
3910	topic, comp_enabled, comp_count, comp_explicit = apply_vs_competitor_routing(
3911	topic,
3912	competitors_flag=args.competitors,
3913	comp_enabled=comp_enabled,
3914	comp_count=comp_count,
3915	comp_explicit=comp_explicit,
3916	comp_plan=comp_plan,
3917	)
3918	if comp_enabled:
3919	config["_perplexity_paid_budget"] = pipeline.PaidSourceBudget(
3920	owner=topic,
3921	)
3922
3923	# Plan alone with zero peers (empty/invalid JSON object, or all entries
3924	# skipped) must not fall through to discover-N with a misleading abort.
3925	if (
3926	comp_enabled
3927	and not comp_explicit
3928	and args.competitors is None
3929	and args.competitors_plan
3930	):
3931	sys.stderr.write(
3932	"[Competitors] --competitors-plan has no usable peer entries "
3933	"(and the topic is not a vs-comparison). Pass a non-empty plan, "
3934	"a vs-topic, --competitors-list, or --competitors N.\n"
3935	)
3936	return 2
3937
3938	# Dedicated subs ride the config dict (already threaded to every source
3939	# fetch) so the keyless Reddit path can pull them floor-exempt without
3940	# widening pipeline.run / _retrieve_stream signatures.
3941	if dedicated_subreddits:
3942	config["_dedicated_subreddits"] = dedicated_subreddits
3943
3944	def _main_runner() -> schema.Report:
3945	r = pipeline.run(
3946	topic=topic,
3947	config=config,
3948	depth=depth,
3949	requested_sources=requested_sources,
3950	mock=args.mock,
3951	x_handle=args.x_handle,
3952	x_related=x_related,
3953	web_backend=args.web_backend,
3954	external_plan=external_plan,
3955	subreddits=subreddits,
3956	tiktok_hashtags=tiktok_hashtags,
3957	tiktok_creators=tiktok_creators,
3958	ig_creators=ig_creators,
3959	lookback_days=args.lookback_days,
3960	as_of_date=args.as_of_date,
3961	github_user=github_user,
3962	github_repos=github_repos,
3963	trustpilot_domain=trustpilot_domain,
3964	trustpilot_domain_is_hint=trustpilot_domain_is_hint,
3965	internal_subrun=comp_enabled,
3966	hiring_signals_mode=args.hiring_signals,
3967	save_dir=args.save_dir,
3968	corpus_dirs=args.corpus,
3969	corpus_all_time=args.corpus_all_time,
3970	x_posts=(
3971	comp_plan.get(topic.strip().lower(), {}).get("_x_envelope")
3972	if comp_enabled else x_posts_envelope
3973	),
3974	)
3975	r.artifacts["resolved"] = {
3976	"entity": topic,
3977	"x_handle": (args.x_handle or "").lstrip("@"),
3978	"subreddits": list(subreddits or []),
3979	"github_user": (github_user or ""),
3980	"github_repos": list(github_repos or []),
3981	"trustpilot_domain": (trustpilot_domain or ""),
3982	"context": config.get("_auto_resolve_context", "") or "",
3983	}
3984	return r
3985
3986	if comp_enabled:
3987	from lib import competitors as competitors_mod
3988	from lib import fanout, resolve as resolve_mod
3989
3990	if comp_explicit:
3991	discovered = comp_explicit
3992	else:
3993	if not resolve_mod._has_backend(config) and not args.mock:
3994	sys.stderr.write(
3995	"[Competitors] Cannot auto-discover peers without help.\n"
3996	"\n"
3997	"RECOMMENDED PATH (hosting reasoning models — Claude Code, Codex, "
3998	"Hermes, Gemini, any agent with a WebSearch tool): YOU have "
3999	"WebSearch. Use it to run full Step 0.55 per entity, then invoke "
4000	"the engine with a vs-topic plus --competitors-plan:\n"
4001	" 1. WebSearch for '{topic} competitors' or '{topic} alternatives'.\n"
4002	" 2. For each peer, WebSearch for handles/subs/github (Step 0.55).\n"
4003	" 3. Re-invoke: /last30days '{topic} vs {peer1} vs {peer2}' "
4004	"--competitors-plan '{\"Peer1\":{\"x_handle\":\"h1\",\"subreddits\":"
4005	"[\"s1\"],...},\"Peer2\":{...}}'.\n"
4006	"See SKILL.md 'Competitor mode' for the full protocol.\n"
4007	"\n"
4008	"HEADLESS / CRON PATH (no hosting model available): set "
4009	"BRAVE_API_KEY / EXA_API_KEY / SERPER_API_KEY / PARALLEL_API_KEY / "
4010	"PERPLEXITY_API_KEY / OPENROUTER_API_KEY and re-run.\n"
4011	"\n"
4012	"MINIMUM ESCAPE HATCH: pass --competitors-list 'A,B,C' to skip "
4013	"discovery. Without --competitors-plan, peer sub-runs fall back to "
4014	"planner defaults and produce visibly thinner data than the main.\n"
4015	)
4016	return 2
4017	discovered = competitors_mod.discover_competitors(
4018	topic, comp_count, config, lookback_days=args.lookback_days,
4019	)
4020	if not discovered:
4021	sys.stderr.write(
4022	f"[Competitors] No peers discovered for {topic!r}; aborting "
4023	"comparison run. Pass --competitors-list to override.\n"
4024	)
4025	return 2
4026
4027	# run_competitor_fanout keys its results by label, so two
4028	# submissions sharing one collapse to a single report while the
4029	# returned list still carries two entries. That yields a
4030	# comparison of an entity against itself, and it hides a failed
4031	# main topic from the survivor check below: the duplicate peer's
4032	# report answers for the label the main run was supposed to fill.
4033	distinct_peers: list[str] = []
4034	claimed_labels = {comparison_label_key(topic)}
4035	for peer in discovered:
4036	key = comparison_label_key(peer)
4037	if key in claimed_labels:
4038	sys.stderr.write(
4039	f"[Competitors] Dropping {peer!r}: duplicates the main "
4040	"topic or an earlier peer.\n"
4041	)
4042	continue
4043	claimed_labels.add(key)
4044	distinct_peers.append(peer)
4045	if not distinct_peers:
4046	sys.stderr.write(
4047	f"[Competitors] No peer distinct from {topic!r} remains; "
4048	"there is nothing to compare against. Pass "
4049	"--competitors-list with distinct entities.\n"
4050	)
4051	return 2
4052	discovered = distinct_peers
4053
4054	sys.stderr.write(
4055	f"[Competitors] Comparing: {topic} vs " + " vs ".join(discovered) + "\n"
4056	)
4057
4058	def _competitor_runner(entity: str) -> schema.Report:
4059	# Deep-copy config so per-entity auto_resolve context does not
4060	# leak across sub-runs. Each sub-run writes its own
4061	# `_auto_resolve_context` into its local config copy.
4062	entity_config = dict(config)
4063	# The Amazon keyword is entity-SPECIFIC, unlike the depth caps
4064	# this shallow copy exists to inherit. Leaving the main topic's
4065	# keyword in place would search Weber SKUs for a Traeger peer,
4066	# render a rival's products as that peer's buyer evidence, and
4067	# multiply the metered spend by the number of entities. Drop it
4068	# so each peer derives its own keyword from its own topic; a
4069	# per-entity keyword can ride in the --competitors-plan entry.
4070	entity_config.pop("_amazon_query", None)
4071	# An advertiser page is per-entity state by definition: left in
4072	# place it would render one brand's ads as every peer's.
4073	entity_config.pop("_meta_ads_page", None)
4074	plan_entry = comp_plan.get(entity.strip().lower(), {})
4075	resolved = {
4076	"entity": entity,
4077	"x_handle": "",
4078	"subreddits": [],
4079	"github_user": "",
4080	"github_repos": [],
4081	"trustpilot_domain": "",
4082	"context": "",
4083	}
4084	# Skip engine-internal auto_resolve when the hosting model
4085	# pre-resolved via --competitors-plan (saves a redundant
4086	# round-trip and makes per-entity Step 0.55 purely
4087	# hosting-model-driven).
4088	plan_covers_fully = bool(plan_entry.get("x_handle")) and bool(
4089	plan_entry.get("subreddits")
4090	)
4091	if (
4092	not args.mock
4093	and not plan_covers_fully
4094	and resolve_mod._has_backend(entity_config)
4095	):
4096	try:
4097	r = resolve_mod.auto_resolve(entity, entity_config)
4098	except Exception as exc:
4099	sys.stderr.write(
4100	f"[Competitors] auto_resolve failed for {entity!r}: "
4101	f"{type(exc).__name__}: {exc}\n"
4102	)
4103	r = {}
4104	resolved["x_handle"] = r.get("x_handle", "") or ""
4105	resolved["subreddits"] = list(r.get("subreddits") or [])
4106	resolved["github_user"] = r.get("github_user", "") or ""
4107	resolved["github_repos"] = list(r.get("github_repos") or [])
4108	resolved["trustpilot_domain"] = r.get("trustpilot_domain", "") or ""
4109	resolved["context"] = r.get("context", "") or ""
4110	kwargs = subrun_kwargs_for(entity, plan_entry, resolved=resolved)
4111	# Record effective per-entity targeting for the Resolved block.
4112	resolved_effective = {
4113	"entity": entity,
4114	"x_handle": kwargs["x_handle"] or "",
4115	"subreddits": kwargs["subreddits"] or [],
4116	"github_user": kwargs["github_user"] or "",
4117	"github_repos": kwargs["github_repos"] or [],
4118	"trustpilot_domain": kwargs["trustpilot_domain"] or "",
4119	"context": kwargs["_context"],
4120	}
4121	if kwargs["_context"]:
4122	entity_config["_auto_resolve_context"] = kwargs["_context"]
4123	sys.stderr.write(
4124	f"[Competitors] {entity}: "
4125	f"x=@{resolved_effective['x_handle'] or '-'} "
4126	f"subs={len(resolved_effective['subreddits'])} "
4127	f"gh={resolved_effective['github_user'] or '-'} "
4128	f"({'plan' if plan_entry else 'auto'})\n"
4129	)
4130	report = pipeline.run(
4131	topic=entity,
4132	config=entity_config,
4133	depth=depth,
4134	requested_sources=requested_sources,
4135	mock=args.mock,
4136	x_handle=kwargs["x_handle"],
4137	x_related=kwargs["x_related"],
4138	subreddits=kwargs["subreddits"],
4139	github_user=kwargs["github_user"],
4140	github_repos=kwargs["github_repos"],
4141	trustpilot_domain=kwargs["trustpilot_domain"],
4142	trustpilot_domain_is_hint=kwargs["_trustpilot_domain_is_hint"],
4143	web_backend=args.web_backend,
4144	lookback_days=args.lookback_days,
4145	as_of_date=args.as_of_date,
4146	hiring_signals_mode=args.hiring_signals,
4147	internal_subrun=True,
4148	save_dir=args.save_dir,
4149	corpus_dirs=args.corpus,
4150	corpus_all_time=args.corpus_all_time,
4151	x_posts=plan_entry.get("_x_envelope"),
4152	)
4153	report.artifacts["resolved"] = resolved_effective
4154	return report
4155
4156	entity_reports = fanout.run_competitor_fanout(
4157	main_topic=topic,
4158	main_runner=_main_runner,
4159	competitors=discovered,
4160	competitor_runner=_competitor_runner,
4161	)
4162	# run_competitor_fanout drops a failed sub-run from the list, and
4163	# the render takes entity_reports[0] as the comparison's subject.
4164	# Without this check, a main topic that raised while >=2 peers
4165	# succeeded silently promoted a competitor to be the subject: the
4166	# report was headed by that peer, saved under its slug, and the
4167	# topic the user actually asked about went unmentioned.
4168	survived = {label for label, _ in entity_reports}
4169	dropped = [
4170	label for label in (topic, *discovered) if label not in survived
4171	]
4172	if topic not in survived:
4173	progress.end_processing()
4174	sys.stderr.write(
4175	f"[Competitors] The main topic {topic!r} failed; "
4176	f"{len(entity_reports)} competitor sub-run(s) survived. "
4177	"Refusing to render a comparison headed by a competitor. "
4178	"Check the warnings above.\n"
4179	)
4180	return 1
4181	if len(entity_reports) < 2:
4182	progress.end_processing()
4183	sys.stderr.write(
4184	f"[Competitors] Fewer than 2 sub-runs survived ({len(entity_reports)}); "
4185	"cannot render a comparison. Re-run without --competitors or check the "
4186	"warnings above.\n"
4187	)
4188	return 1
4189	report = entity_reports[0][1]
4190	if dropped:
4191	# A narrower comparison than the user asked for is a result
4192	# they need to see, not a silent substitution.
4193	report.warnings.append(
4194	"Comparison is incomplete: "
4195	f"{len(dropped)} of {len(discovered) + 1} entities failed and "
4196	f"were dropped ({', '.join(dropped)})."
4197	)
4198	else:
4199	entity_reports = None
4200	report = _main_runner()
4201	except Exception as exc:
4202	progress.end_processing()
4203	progress.show_error(str(exc))
4204	raise
4205	if _freshness_enabled(args, config):
4206	_verify_report_set(report, entity_reports, allow_network=not args.mock)
4207
4208	_show_runtime_ui(
4209	report, progress, diag,
4210	suppress_web_promo=bool(external_plan or comp_plan),
4211	)
4212	_write_last_run(
4213	original_topic, report, entity_reports=entity_reports,
4214	x_envelope_sha256=_x_envelope_digest(x_posts_envelope, comp_plan),
4215	)
4216	# LAST30DAYS_STORE env var = persistence default-on. Read both os.environ
4217	# (for shell-exported users) and config (for users who set it in
4218	# ~/.config/last30days/.env, which env.py loads but does not propagate
4219	# to os.environ). Mirrors the LAST30DAYS_DEBUG / LAST30DAYS_SKIP_PREFLIGHT
4220	# convention; env-var or config wins, with `--store` flag still working.
4221	_store_env = (
4222	os.environ.get("LAST30DAYS_STORE")
4223	or config.get("LAST30DAYS_STORE")
4224	or ""
4225	).lower()
4226	if args.store or _store_env in ("1", "true", "yes"):
4227	counts = persist_report(report, store_db=_scoped_store_db(args))
4228	sys.stderr.write(
4229	f"[last30days] Stored {counts['new']} new, {counts['updated']} updated findings\n"
4230	)
4231	sys.stderr.flush()
4232
4233	# Show quality nudge if applicable. Explicit hiring-signal runs are
4234	# intentionally jobs-focused, so generic source setup advice is noise.
4235	if not args.hiring_signals:
4236	try:
4237	from lib import quality_nudge
4238	from lib import youtube_yt as _youtube_yt
4239	# Populate transcript-fetch ratio so quality_nudge can detect the
4240	# degraded-YouTube failure mode (videos returned but transcripts
4241	# silently failed - typically a stale yt-dlp binary).
4242	youtube_items = report.items_by_source.get("youtube") or []
4243	_yt_fetch_stats = _youtube_yt.get_transcript_fetch_stats()
4244	instagram_items = report.items_by_source.get("instagram") or []
4245	research_results = {
4246	"active_sources": diag.get("available_sources") or [],
4247	"youtube_videos_count": len(youtube_items),
4248	"youtube_transcripts_count": sum(
4249	1 for it in youtube_items
4250	if (it.metadata.get("transcript_highlights") or it.metadata.get("transcript_snippet"))
4251	),
4252	"youtube_error": report.errors_by_source.get("youtube"),
4253	"x_error": report.errors_by_source.get("x"),
4254	# Captions-disabled videos can never produce a transcript regardless
4255	# of yt-dlp version; subtract them from the degraded-ratio
4256	# denominator so a single uploader-disabled video does not trip the
4257	# "stale yt-dlp" nudge.
4258	"youtube_captions_disabled_count": sum(
4259	1 for it in youtube_items if it.metadata.get("captions_disabled")
4260	),
4261	# Actual yt-dlp fetch outcomes for this run. The counts above are
4262	# computed from post-pruning items, so they can't tell "fetches
4263	# failed (stale binary)" from "fetches succeeded but the videos
4264	# were pruned downstream"; the latter was producing false
4265	# stale-yt-dlp nudges (#531).
4266	"youtube_transcript_fetch_attempts": _yt_fetch_stats["attempts"],
4267	"youtube_transcript_fetch_failures": _yt_fetch_stats["failures"],
4268	# Track Instagram returned-zero-items so quality_nudge can detect
4269	# the silent-failure case (SC configured but the v2 reels endpoint
4270	# 500'd through both the original query and the hashtag retry).
4271	"instagram_items_count": len(instagram_items),
4272	}
4273	quality = quality_nudge.compute_quality_score(config, research_results)
4274	if quality.get("nudge_text"):
4275	sys.stderr.write(f"\n{quality['nudge_text']}\n")
4276	sys.stderr.flush()
4277	except Exception:
4278	pass
4279
4280	# Signal to render_compact whether pre-research flags were supplied.
4281	# Used to emit a Pre-Research Status warning when the model skipped
4282	# Step 0.5 / 0.55 and invoked the engine bare on an eligible topic.
4283	pre_research_flags_present = bool(
4284	args.x_handle
4285	or args.github_user
4286	or args.subreddits
4287	or args.plan
4288	or args.auto_resolve
4289	or args.tiktok_creators
4290	or args.ig_creators
4291	)
4292	report.artifacts["pre_research_flags_present"] = pre_research_flags_present
4293
4294	exit_code = _render_save_and_print(args, report, entity_reports, synthesis_md, config)
4295	if args.emit in {"compact", "md", "brief"}:
4296	x_omission = _optional_x_omission_text(diag, requested_sources)
4297	if x_omission:
4298	sys.stderr.write(f"\n{x_omission}\n")
4299	sys.stderr.flush()
4300	return exit_code
4301
4302
4303	if __name__ == "__main__":
4304	raise SystemExit(main())
4305

4305 lines PYTHON