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

3615 lines PYTHON