| 1 | """File contracts for the three-command host-judged discovery protocol. |
| 2 | |
| 3 | Leg 1 (``--discover --nominate-only``) writes the nominations bundle: the |
| 4 | FULL judge pool, each nomination with its complete seed item set, serialized |
| 5 | losslessly so leg 2 can recompute floor/velocity/entity-token disambiguation |
| 6 | exactly as an in-memory run would. Leg 2 (``--discover --judgments <file>``) |
| 7 | reads host judgments (names/junk/worthiness) bound to the bundle by |
| 8 | bundle_id. Leg 3 (``--discover --finalize [--angles <file>]``) applies |
| 9 | host-written content angles. |
| 10 | |
| 11 | This module owns the handoff contracts - bundle writer/reader, judgments |
| 12 | reader, pending-report reader (the leg-2 output leg 3 finalizes from), |
| 13 | angles reader - plus the host-facing digest and the post-judgment |
| 14 | name-collision resolver. Readers are strict at the top level (typed |
| 15 | ``HandoffContractError``, mapped to exit 2 by the CLI layer) and lenient per |
| 16 | row: a malformed or omitted row falls back to the bundle's heuristics rather |
| 17 | than failing the run. |
| 18 | """ |
| 19 | |
| 20 | from __future__ import annotations |
| 21 | |
| 22 | import json |
| 23 | import secrets |
| 24 | from collections import Counter |
| 25 | from dataclasses import dataclass, field |
| 26 | from pathlib import Path |
| 27 | from typing import Any, Callable, Iterator, Sequence |
| 28 | |
| 29 | from . import env, log, pipeline, rerank, schema |
| 30 | |
| 31 | |
| 32 | # How long a nominations bundle stays valid. Deliberately a module constant |
| 33 | # and NOT the LAST30DAYS_REPORT_CACHE_TTL_SECONDS env knob: a user who |
| 34 | # lowered the report-cache TTL for drill freshness must not shrink the |
| 35 | # window a host has to author judgments. |
| 36 | DISCOVERY_HANDOFF_TTL_SECONDS = 3600.0 |
| 37 | |
| 38 | NOMINATIONS_BUNDLE_FILENAME = "discover-nominations.json" |
| 39 | PENDING_REPORT_FILENAME = "discover-pending.json" |
| 40 | |
| 41 | _VALID_TIERS = ("deep", "shallow") |
| 42 | |
| 43 | _RESWEEP_REMEDY = "Run a fresh `--discover --nominate-only` re-sweep." |
| 44 | |
| 45 | # Leg-3 remedy: the pending report is leg-2 output, so the first fix is to |
| 46 | # re-run the resume leg; only when the bundle itself has also gone stale does |
| 47 | # the whole protocol restart. |
| 48 | _RESUME_REMEDY = ( |
| 49 | "Re-run the resume leg (`--discover --judgments <file>`), or the full " |
| 50 | "protocol from `--discover --nominate-only` if the bundle is stale too." |
| 51 | ) |
| 52 | |
| 53 | # Defensive caps on host-supplied text, ported from the retired engine-judge |
| 54 | # pass: names become search queries and the /last30days handoff, angles |
| 55 | # render verbatim on trend cards, so a runaway (or adversarial) value never |
| 56 | # yields an unbounded string. |
| 57 | _NAME_MAX_CHARS = 96 |
| 58 | _ANGLE_MAX_CHARS = 200 |
| 59 | |
| 60 | # Unified trailing-punctuation charset for word-boundary truncation: names |
| 61 | # and angle sentences share it so the strip sets cannot drift. |
| 62 | _TRUNCATE_STRIP_CHARS = " \"'`.,;:!?-" |
| 63 | |
| 64 | # Digest evidence caps: the surface the engine judge used to see per |
| 65 | # nomination (leader title, leader snippet, strongest community comment). |
| 66 | _DIGEST_TITLE_MAX_CHARS = 220 |
| 67 | _DIGEST_SNIPPET_MAX_CHARS = 420 |
| 68 | _DIGEST_COMMENT_MAX_CHARS = 340 |
| 69 | |
| 70 | |
| 71 | class HandoffContractError(Exception): |
| 72 | """A handoff file failed its contract: unreadable, invalid JSON, wrong |
| 73 | shape or schema version, stale, or not bound to the current bundle. |
| 74 | The CLI layer maps this to exit code 2.""" |
| 75 | |
| 76 | def __init__(self, message: str) -> None: |
| 77 | super().__init__(message) |
| 78 | self.message = message |
| 79 | |
| 80 | |
| 81 | @dataclass(frozen=True) |
| 82 | class PoolEntry: |
| 83 | """One judge-pool nomination as handed to the bundle writer (leg 1). |
| 84 | |
| 85 | ``heuristic_name`` and ``heuristic_junk`` are the deterministic |
| 86 | topic_shape fallbacks, kept alongside the nomination so leg 2 can fill |
| 87 | any row the host omitted without re-deriving them. |
| 88 | """ |
| 89 | |
| 90 | nomination: pipeline.Nomination |
| 91 | cluster_id: str |
| 92 | heuristic_name: str |
| 93 | heuristic_junk: bool |
| 94 | |
| 95 | |
| 96 | @dataclass(frozen=True) |
| 97 | class BundleNomination: |
| 98 | """One nomination read back from a bundle, with its stable id.""" |
| 99 | |
| 100 | nomination_id: str |
| 101 | nomination: pipeline.Nomination |
| 102 | cluster_id: str |
| 103 | heuristic_name: str |
| 104 | heuristic_junk: bool |
| 105 | sources: list[str] |
| 106 | engagement_by_source: dict[str, dict[str, float | int]] = field( |
| 107 | default_factory=dict |
| 108 | ) |
| 109 | |
| 110 | |
| 111 | @dataclass(frozen=True) |
| 112 | class NominationsBundle: |
| 113 | """A parsed leg-1 nominations bundle (also returned by the writer). |
| 114 | |
| 115 | ``source_status`` is the leg-1 sweep's finalized per-source outcome map: |
| 116 | legs 2 and 3 restore it so degraded sweep coverage survives the protocol |
| 117 | instead of silently reading as clean. ``mock`` is the writing run's |
| 118 | provenance - mock-born state must never be finalized by a real run (and |
| 119 | vice versa); files written before either field existed read as an empty |
| 120 | map and a real run.""" |
| 121 | |
| 122 | schema_version: str |
| 123 | bundle_id: str |
| 124 | generated_at: str |
| 125 | from_date: str |
| 126 | to_date: str |
| 127 | domain: str |
| 128 | tier: str |
| 129 | enrichment_source_boundary: list[str] | None |
| 130 | requested_sources: list[str] | None |
| 131 | lookback_days: int |
| 132 | nominations: list[BundleNomination] |
| 133 | source_status: dict[str, schema.SourceOutcome] = field(default_factory=dict) |
| 134 | mock: bool = False |
| 135 | path: Path | None = None |
| 136 | |
| 137 | |
| 138 | @dataclass(frozen=True) |
| 139 | class HostJudgment: |
| 140 | """One host verdict row. ``None`` on any field means the host left it |
| 141 | absent for that row and the caller falls back to the bundle's heuristic |
| 142 | value (name/junk) or to no worthiness signal.""" |
| 143 | |
| 144 | name: str | None |
| 145 | junk: bool | None |
| 146 | worthiness: int | None |
| 147 | |
| 148 | |
| 149 | # The per-row-absent marker: what ``judgment_for`` returns for a nomination |
| 150 | # the host omitted entirely. Every field falls back to the bundle heuristics. |
| 151 | ROW_ABSENT = HostJudgment(name=None, junk=None, worthiness=None) |
| 152 | |
| 153 | |
| 154 | @dataclass(frozen=True) |
| 155 | class HostAngles: |
| 156 | """One host-written angle row; either field may be absent.""" |
| 157 | |
| 158 | podcast: str | None |
| 159 | x_article: str | None |
| 160 | |
| 161 | |
| 162 | @dataclass(frozen=True) |
| 163 | class PendingReport: |
| 164 | """A parsed leg-2 pending report: the floored/folded/ranked discovery |
| 165 | report (as its raw ``schema.to_dict`` payload - leg 3 rebuilds it via |
| 166 | ``schema.discovery_report_from_dict``) plus the angle inputs keyed by |
| 167 | surviving nomination id. ``run_ref`` is the leg-2 run identity the |
| 168 | finalize leg replays into the topic queue so retries stay idempotent.""" |
| 169 | |
| 170 | schema_version: str |
| 171 | bundle_id: str |
| 172 | generated_at: str |
| 173 | run_ref: str |
| 174 | report: dict[str, Any] |
| 175 | angle_inputs: dict[str, dict[str, str]] |
| 176 | # Leg-2 provenance: True when a --mock resume wrote this file. Files |
| 177 | # written before the flag existed read as real (False). |
| 178 | mock: bool = False |
| 179 | path: Path | None = None |
| 180 | |
| 181 | |
| 182 | def _warn(message: str) -> None: |
| 183 | log.source_log("Discover", message, tty_only=False) |
| 184 | |
| 185 | |
| 186 | def handoff_state_dir( |
| 187 | save_dir: str | Path | None, |
| 188 | config_dir: Path | None, |
| 189 | ) -> Path | None: |
| 190 | """Resolve the handoff state directory: ``save_dir`` when provided, else |
| 191 | the config dir (mirrors the report-cache convention in last30days.py). |
| 192 | Both are accepted as arguments so this module never imports the CLI |
| 193 | layer above it. Returns None when neither location is available.""" |
| 194 | if save_dir: |
| 195 | return Path(save_dir).expanduser().resolve() |
| 196 | if config_dir is not None: |
| 197 | return Path(config_dir) |
| 198 | return None |
| 199 | |
| 200 | |
| 201 | def nominations_bundle_path(state_dir: str | Path) -> Path: |
| 202 | """The nominations bundle file inside a handoff state directory.""" |
| 203 | return Path(state_dir) / NOMINATIONS_BUNDLE_FILENAME |
| 204 | |
| 205 | |
| 206 | def pending_report_path(state_dir: str | Path) -> Path: |
| 207 | """The leg-2 pending-report file inside a handoff state directory.""" |
| 208 | return Path(state_dir) / PENDING_REPORT_FILENAME |
| 209 | |
| 210 | |
| 211 | def _search_paths( |
| 212 | save_dir: str | Path | None, |
| 213 | config_dir: Path | None, |
| 214 | path_fn: Callable[[Path], Path], |
| 215 | ) -> list[Path]: |
| 216 | """Candidate handoff-file locations: ONLY the save dir when one was |
| 217 | supplied, else the config dir. An explicit save dir is the protocol's |
| 218 | single handoff store (mirroring ``_scoped_store_db`` and SKILL.md's "a |
| 219 | different or missing save dir on a later leg means the leg cannot find |
| 220 | them" contract), so a handoff file in the config dir must never silently |
| 221 | satisfy a save-dir run. ``path_fn`` picks which handoff file (bundle vs |
| 222 | pending).""" |
| 223 | if save_dir: |
| 224 | return [path_fn(Path(save_dir).expanduser().resolve())] |
| 225 | if config_dir is not None: |
| 226 | return [path_fn(Path(config_dir))] |
| 227 | return [] |
| 228 | |
| 229 | |
| 230 | def _searched_lines(searched: list[Path]) -> str: |
| 231 | if not searched: |
| 232 | return " (no --save-dir and no config directory available)" |
| 233 | return "\n".join(f" - {path}" for path in searched) |
| 234 | |
| 235 | |
| 236 | def write_nominations_bundle( |
| 237 | entries: Sequence[PoolEntry], |
| 238 | *, |
| 239 | domain: str, |
| 240 | tier: str, |
| 241 | from_date: str, |
| 242 | to_date: str, |
| 243 | lookback_days: int, |
| 244 | enrichment_source_boundary: list[str] | None, |
| 245 | requested_sources: list[str] | None, |
| 246 | source_status: dict[str, schema.SourceOutcome] | None = None, |
| 247 | mock: bool = False, |
| 248 | save_dir: str | Path | None = None, |
| 249 | config_dir: Path | None = None, |
| 250 | ) -> NominationsBundle: |
| 251 | """Write the leg-1 nominations bundle and return its parsed form. |
| 252 | |
| 253 | Nomination ids are assigned ``n1, n2, ...`` in pool order. The leg-1 |
| 254 | invocation context (enrichment source boundary, requested discovery |
| 255 | sources, lookback days) rides along so leg 2 resumes with identical |
| 256 | settings. ``None`` boundaries are preserved as null - "no boundary" and |
| 257 | "empty boundary" are different contracts. ``source_status`` is the |
| 258 | sweep's finalized per-source outcome map (serialized via the same |
| 259 | ``schema.to_dict`` round trip every report uses) so degraded coverage |
| 260 | survives into legs 2-3; ``mock`` stamps the writing run's provenance. |
| 261 | """ |
| 262 | if tier not in _VALID_TIERS: |
| 263 | raise ValueError(f"tier must be one of {_VALID_TIERS}, got {tier!r}") |
| 264 | state_dir = handoff_state_dir(save_dir, config_dir) |
| 265 | if state_dir is None: |
| 266 | raise HandoffContractError( |
| 267 | "No handoff location available to write the nominations bundle: " |
| 268 | "pass --save-dir or configure ~/.config/last30days/." |
| 269 | ) |
| 270 | bundle_id = secrets.token_hex(8) |
| 271 | generated_at = schema._utc_now() |
| 272 | |
| 273 | rows: list[dict[str, Any]] = [] |
| 274 | nominations: list[BundleNomination] = [] |
| 275 | for index, entry in enumerate(entries, start=1): |
| 276 | nomination_id = f"n{index}" |
| 277 | sources = sorted({item.source for item in entry.nomination.items}) |
| 278 | engagement = pipeline._discovery_engagement(entry.nomination.items) |
| 279 | rows.append({ |
| 280 | "id": nomination_id, |
| 281 | "cluster_id": entry.cluster_id, |
| 282 | "heuristic_name": entry.heuristic_name, |
| 283 | "heuristic_junk": bool(entry.heuristic_junk), |
| 284 | "sources": sources, |
| 285 | "engagement_by_source": engagement, |
| 286 | "nomination": schema.nomination_to_dict(entry.nomination), |
| 287 | }) |
| 288 | nominations.append(BundleNomination( |
| 289 | nomination_id=nomination_id, |
| 290 | nomination=entry.nomination, |
| 291 | cluster_id=entry.cluster_id, |
| 292 | heuristic_name=entry.heuristic_name, |
| 293 | heuristic_junk=bool(entry.heuristic_junk), |
| 294 | sources=sources, |
| 295 | engagement_by_source=engagement, |
| 296 | )) |
| 297 | |
| 298 | payload = { |
| 299 | "schema_version": schema.DISCOVERY_NOMINATIONS_SCHEMA_VERSION, |
| 300 | "kind": schema.DISCOVERY_NOMINATIONS_KIND, |
| 301 | "bundle_id": bundle_id, |
| 302 | "generated_at": generated_at, |
| 303 | "from_date": from_date, |
| 304 | "to_date": to_date, |
| 305 | "domain": domain, |
| 306 | "tier": tier, |
| 307 | "mock": bool(mock), |
| 308 | "source_status": { |
| 309 | source: schema.to_dict(outcome) |
| 310 | for source, outcome in (source_status or {}).items() |
| 311 | }, |
| 312 | "context": { |
| 313 | "enrichment_source_boundary": ( |
| 314 | list(enrichment_source_boundary) |
| 315 | if enrichment_source_boundary is not None |
| 316 | else None |
| 317 | ), |
| 318 | "requested_sources": ( |
| 319 | list(requested_sources) if requested_sources is not None else None |
| 320 | ), |
| 321 | "lookback_days": int(lookback_days), |
| 322 | }, |
| 323 | "nominations": rows, |
| 324 | } |
| 325 | path = nominations_bundle_path(state_dir) |
| 326 | try: |
| 327 | state_dir.mkdir(parents=True, exist_ok=True) |
| 328 | path.write_text(json.dumps(payload, indent=2), encoding="utf-8") |
| 329 | except OSError as exc: |
| 330 | # A locked/read-only/full disk is the protocol's clean exit-2 path, |
| 331 | # never a traceback. |
| 332 | raise HandoffContractError( |
| 333 | f"Could not write nominations bundle {path}: {exc}" |
| 334 | ) from exc |
| 335 | return NominationsBundle( |
| 336 | schema_version=schema.DISCOVERY_NOMINATIONS_SCHEMA_VERSION, |
| 337 | bundle_id=bundle_id, |
| 338 | generated_at=generated_at, |
| 339 | from_date=from_date, |
| 340 | to_date=to_date, |
| 341 | domain=domain, |
| 342 | tier=tier, |
| 343 | enrichment_source_boundary=( |
| 344 | list(enrichment_source_boundary) |
| 345 | if enrichment_source_boundary is not None |
| 346 | else None |
| 347 | ), |
| 348 | requested_sources=( |
| 349 | list(requested_sources) if requested_sources is not None else None |
| 350 | ), |
| 351 | lookback_days=int(lookback_days), |
| 352 | nominations=nominations, |
| 353 | source_status=dict(source_status or {}), |
| 354 | mock=bool(mock), |
| 355 | path=path, |
| 356 | ) |
| 357 | |
| 358 | |
| 359 | def read_nominations_bundle( |
| 360 | *, |
| 361 | save_dir: str | Path | None = None, |
| 362 | config_dir: Path | None = None, |
| 363 | ) -> NominationsBundle: |
| 364 | """Locate and parse the nominations bundle for legs 2 and 3. |
| 365 | |
| 366 | The bundle lives in the save dir when one was supplied, else the config |
| 367 | dir - never both (no cross-store fallback). Raises HandoffContractError |
| 368 | (naming the searched location and the re-sweep remedy) when no bundle |
| 369 | exists, and for any top-level contract violation in the file found. |
| 370 | """ |
| 371 | searched = _search_paths(save_dir, config_dir, nominations_bundle_path) |
| 372 | path = next((candidate for candidate in searched if candidate.exists()), None) |
| 373 | if path is None: |
| 374 | raise HandoffContractError( |
| 375 | "No discovery nominations bundle found. Searched:\n" |
| 376 | f"{_searched_lines(searched)}\n{_RESWEEP_REMEDY}" |
| 377 | ) |
| 378 | return _parse_bundle_file(path) |
| 379 | |
| 380 | |
| 381 | def _parse_handoff_envelope( |
| 382 | path: Path, |
| 383 | *, |
| 384 | label: str, |
| 385 | kind: str, |
| 386 | schema_version: str, |
| 387 | remedy: str, |
| 388 | missing_id_context: str, |
| 389 | stale_context: str, |
| 390 | ) -> tuple[dict[str, Any], str, Any]: |
| 391 | """Shared strict top-level validation for the two engine-written handoff |
| 392 | files (nominations bundle, pending report): readable, valid JSON object, |
| 393 | right kind and schema version, bundle_id present, within TTL. Returns |
| 394 | (payload, bundle_id, generated_at).""" |
| 395 | try: |
| 396 | raw = path.read_text(encoding="utf-8") |
| 397 | except OSError as exc: |
| 398 | raise HandoffContractError( |
| 399 | f"Could not read {label.lower()} {path}: {exc}" |
| 400 | ) from exc |
| 401 | try: |
| 402 | payload = json.loads(raw) |
| 403 | except json.JSONDecodeError as exc: |
| 404 | raise HandoffContractError( |
| 405 | f"{label} {path} is not valid JSON: {exc}" |
| 406 | ) from exc |
| 407 | if not isinstance(payload, dict): |
| 408 | raise HandoffContractError( |
| 409 | f"{label} {path} must be a top-level JSON object, " |
| 410 | f"got {type(payload).__name__}." |
| 411 | ) |
| 412 | version = payload.get("schema_version") |
| 413 | if version != schema_version: |
| 414 | raise HandoffContractError( |
| 415 | f"{label} {path} has schema version {version!r}; this " |
| 416 | f"build reads {schema_version!r}. {remedy}" |
| 417 | ) |
| 418 | file_kind = payload.get("kind") |
| 419 | if file_kind != kind: |
| 420 | raise HandoffContractError( |
| 421 | f"{label} {path} has kind {file_kind!r}; expected " |
| 422 | f"{kind!r}. {remedy}" |
| 423 | ) |
| 424 | bundle_id = str(payload.get("bundle_id") or "") |
| 425 | if not bundle_id: |
| 426 | raise HandoffContractError( |
| 427 | f"{label} {path} is missing its bundle_id; " |
| 428 | f"{missing_id_context}. {remedy}" |
| 429 | ) |
| 430 | generated_at = payload.get("generated_at") |
| 431 | if not env.is_timestamp_fresh(generated_at, DISCOVERY_HANDOFF_TTL_SECONDS): |
| 432 | raise HandoffContractError( |
| 433 | f"{label} {path} is stale (generated_at=" |
| 434 | f"{generated_at!r}, TTL {int(DISCOVERY_HANDOFF_TTL_SECONDS)}s): " |
| 435 | f"{stale_context}. {remedy}" |
| 436 | ) |
| 437 | return payload, bundle_id, generated_at |
| 438 | |
| 439 | |
| 440 | def _parse_bundle_file(path: Path) -> NominationsBundle: |
| 441 | payload, bundle_id, generated_at = _parse_handoff_envelope( |
| 442 | path, |
| 443 | label="Nominations bundle", |
| 444 | kind=schema.DISCOVERY_NOMINATIONS_KIND, |
| 445 | schema_version=schema.DISCOVERY_NOMINATIONS_SCHEMA_VERSION, |
| 446 | remedy=_RESWEEP_REMEDY, |
| 447 | missing_id_context="judgments cannot bind to it", |
| 448 | stale_context="the momentum window it captured has moved on", |
| 449 | ) |
| 450 | version = payload.get("schema_version") |
| 451 | |
| 452 | context = payload.get("context") or {} |
| 453 | boundary = context.get("enrichment_source_boundary") |
| 454 | requested = context.get("requested_sources") |
| 455 | try: |
| 456 | lookback_days = int(context.get("lookback_days") or 30) |
| 457 | except (TypeError, ValueError): |
| 458 | lookback_days = 30 |
| 459 | |
| 460 | rows_raw = payload.get("nominations") |
| 461 | if not isinstance(rows_raw, list): |
| 462 | raise HandoffContractError( |
| 463 | f"Nominations bundle {path} must carry a top-level " |
| 464 | f"\"nominations\" list, got {type(rows_raw).__name__}. " |
| 465 | f"{_RESWEEP_REMEDY}" |
| 466 | ) |
| 467 | |
| 468 | nominations: list[BundleNomination] = [] |
| 469 | for position, row in enumerate(rows_raw, start=1): |
| 470 | # Lenient per row: the bundle is engine-written, but one corrupted |
| 471 | # row must not discard the rest of the pool. |
| 472 | if not isinstance(row, dict): |
| 473 | _warn( |
| 474 | f"skipping malformed nomination row {position} in " |
| 475 | f"{path.name} (not an object)" |
| 476 | ) |
| 477 | continue |
| 478 | try: |
| 479 | nomination = pipeline.Nomination( |
| 480 | **schema.nomination_kwargs_from_dict(row.get("nomination") or {}) |
| 481 | ) |
| 482 | except (KeyError, TypeError, ValueError) as exc: |
| 483 | _warn( |
| 484 | f"skipping unparseable nomination row {position} in " |
| 485 | f"{path.name}: {type(exc).__name__}: {exc}" |
| 486 | ) |
| 487 | continue |
| 488 | engagement_raw = row.get("engagement_by_source") |
| 489 | engagement = { |
| 490 | str(source): dict(metrics) |
| 491 | for source, metrics in ( |
| 492 | engagement_raw.items() if isinstance(engagement_raw, dict) else () |
| 493 | ) |
| 494 | if isinstance(metrics, dict) |
| 495 | } |
| 496 | nominations.append(BundleNomination( |
| 497 | nomination_id=str(row.get("id") or f"n{position}"), |
| 498 | nomination=nomination, |
| 499 | cluster_id=str(row.get("cluster_id") or ""), |
| 500 | heuristic_name=str(row.get("heuristic_name") or ""), |
| 501 | heuristic_junk=bool(row.get("heuristic_junk")), |
| 502 | sources=[str(source) for source in row.get("sources") or []], |
| 503 | engagement_by_source=engagement, |
| 504 | )) |
| 505 | |
| 506 | if not nominations: |
| 507 | # Leg 1 never writes an empty bundle (a zero-nomination sweep |
| 508 | # short-circuits with no bundle file), so an empty or all-invalid |
| 509 | # nominations array is corrupt state: fail closed, never hand the |
| 510 | # resume leg a silently empty pool. |
| 511 | raise HandoffContractError( |
| 512 | f"Nominations bundle {path} contains no readable nominations " |
| 513 | f"(leg 1 never writes an empty pool). {_RESWEEP_REMEDY}" |
| 514 | ) |
| 515 | |
| 516 | # Sweep status is advisory coverage context: restore it through the same |
| 517 | # deserializer every report uses, but degrade a malformed map to empty |
| 518 | # rather than discarding an otherwise-valid pool. |
| 519 | try: |
| 520 | source_status = schema._source_status_from_dict(payload) |
| 521 | except (AttributeError, KeyError, TypeError, ValueError): |
| 522 | _warn(f"ignoring malformed source_status map in {path.name}") |
| 523 | source_status = {} |
| 524 | |
| 525 | return NominationsBundle( |
| 526 | schema_version=str(version), |
| 527 | bundle_id=bundle_id, |
| 528 | generated_at=str(generated_at or ""), |
| 529 | from_date=str(payload.get("from_date") or ""), |
| 530 | to_date=str(payload.get("to_date") or ""), |
| 531 | domain=str(payload.get("domain") or ""), |
| 532 | tier=str(payload.get("tier") or "deep"), |
| 533 | enrichment_source_boundary=( |
| 534 | [str(source) for source in boundary] |
| 535 | if isinstance(boundary, list) else None |
| 536 | ), |
| 537 | requested_sources=( |
| 538 | [str(source) for source in requested] |
| 539 | if isinstance(requested, list) else None |
| 540 | ), |
| 541 | lookback_days=lookback_days, |
| 542 | nominations=nominations, |
| 543 | source_status=source_status, |
| 544 | mock=bool(payload.get("mock")), |
| 545 | path=path, |
| 546 | ) |
| 547 | |
| 548 | |
| 549 | def read_pending_report( |
| 550 | *, |
| 551 | save_dir: str | Path | None = None, |
| 552 | config_dir: Path | None = None, |
| 553 | ) -> PendingReport: |
| 554 | """Locate and parse the leg-2 pending report for the finalize leg. |
| 555 | |
| 556 | Same strictness family as the bundle reader: missing file (the searched |
| 557 | location named - save dir when supplied, else config dir, never a |
| 558 | cross-store fallback), unreadable, invalid JSON, wrong kind or schema version, |
| 559 | missing bundle_id, or stale TTL all raise HandoffContractError (mapped to |
| 560 | exit 2 by the CLI layer). Staleness is measured from the PENDING report's |
| 561 | own generated_at - the leg-2 write started a fresh authoring window - and |
| 562 | the remedy is the resume leg, not a full re-sweep. |
| 563 | """ |
| 564 | searched = _search_paths(save_dir, config_dir, pending_report_path) |
| 565 | path = next((candidate for candidate in searched if candidate.exists()), None) |
| 566 | if path is None: |
| 567 | raise HandoffContractError( |
| 568 | "No pending discovery report found. Searched:\n" |
| 569 | f"{_searched_lines(searched)}\n{_RESUME_REMEDY}" |
| 570 | ) |
| 571 | return _parse_pending_file(path) |
| 572 | |
| 573 | |
| 574 | def _parse_pending_file(path: Path) -> PendingReport: |
| 575 | payload, bundle_id, generated_at = _parse_handoff_envelope( |
| 576 | path, |
| 577 | label="Pending discovery report", |
| 578 | kind=schema.DISCOVERY_PENDING_KIND, |
| 579 | schema_version=schema.DISCOVERY_PENDING_SCHEMA_VERSION, |
| 580 | remedy=_RESUME_REMEDY, |
| 581 | missing_id_context="angles cannot bind to it", |
| 582 | stale_context="the judged window it captured has moved on", |
| 583 | ) |
| 584 | version = payload.get("schema_version") |
| 585 | report = payload.get("report") |
| 586 | if not isinstance(report, dict): |
| 587 | raise HandoffContractError( |
| 588 | f"Pending discovery report {path} must carry a top-level " |
| 589 | f"\"report\" object. {_RESUME_REMEDY}" |
| 590 | ) |
| 591 | # Lenient per row (engine-written, but one corrupt row must not discard |
| 592 | # the rest): keep only well-shaped angle-input entries. |
| 593 | angle_inputs_raw = payload.get("angle_inputs") |
| 594 | angle_inputs = { |
| 595 | str(nomination_id): { |
| 596 | str(key): str(value) for key, value in info.items() |
| 597 | } |
| 598 | for nomination_id, info in ( |
| 599 | angle_inputs_raw.items() if isinstance(angle_inputs_raw, dict) else () |
| 600 | ) |
| 601 | if isinstance(info, dict) |
| 602 | } |
| 603 | return PendingReport( |
| 604 | schema_version=str(version), |
| 605 | bundle_id=bundle_id, |
| 606 | generated_at=str(generated_at or ""), |
| 607 | run_ref=str(payload.get("run_ref") or ""), |
| 608 | report=report, |
| 609 | angle_inputs=angle_inputs, |
| 610 | mock=bool(payload.get("mock")), |
| 611 | path=path, |
| 612 | ) |
| 613 | |
| 614 | |
| 615 | def _load_host_file(path: str | Path, label: str) -> dict[str, Any]: |
| 616 | """Load a host-authored handoff file with strict top-level checks.""" |
| 617 | file_path = Path(path).expanduser() |
| 618 | try: |
| 619 | raw = file_path.read_text(encoding="utf-8") |
| 620 | except OSError as exc: |
| 621 | raise HandoffContractError( |
| 622 | f"Could not read {label} file {file_path}: {exc}" |
| 623 | ) from exc |
| 624 | try: |
| 625 | payload = json.loads(raw) |
| 626 | except json.JSONDecodeError as exc: |
| 627 | raise HandoffContractError( |
| 628 | f"{label.capitalize()} file {file_path} is not valid JSON: {exc}" |
| 629 | ) from exc |
| 630 | if not isinstance(payload, dict): |
| 631 | raise HandoffContractError( |
| 632 | f"{label.capitalize()} file {file_path} must be a top-level JSON " |
| 633 | f"object, got {type(payload).__name__}." |
| 634 | ) |
| 635 | return payload |
| 636 | |
| 637 | |
| 638 | def _require_bundle_binding( |
| 639 | payload: dict[str, Any], |
| 640 | bundle: NominationsBundle | PendingReport, |
| 641 | *, |
| 642 | label: str, |
| 643 | save_dir: str | Path | None, |
| 644 | config_dir: Path | None, |
| 645 | ) -> None: |
| 646 | """Enforce bundle-id binding between a host file and the current bundle |
| 647 | (or, on the finalize leg, the pending report that inherited its id). |
| 648 | The mismatch message names the file actually validated against - the |
| 649 | pending report on the finalize leg - so a host's retry is not misdirected |
| 650 | at the nominations bundle. A mismatch means the host echoed the wrong id |
| 651 | into an otherwise-current file, so the remedy is the cheap one - correct |
| 652 | the bundle_id field and re-run this same leg - never the expensive |
| 653 | re-sweep/resume remedies (those belong to missing/stale state).""" |
| 654 | file_bundle_id = str(payload.get("bundle_id") or "") |
| 655 | if file_bundle_id == bundle.bundle_id: |
| 656 | return |
| 657 | if isinstance(bundle, PendingReport): |
| 658 | searched = _search_paths(save_dir, config_dir, pending_report_path) |
| 659 | noun = "current pending discovery report" |
| 660 | location_label = "Pending-report locations searched" |
| 661 | else: |
| 662 | searched = _search_paths(save_dir, config_dir, nominations_bundle_path) |
| 663 | noun = "current nominations bundle" |
| 664 | location_label = "Bundle locations searched" |
| 665 | if not searched and bundle.path is not None: |
| 666 | searched = [bundle.path] |
| 667 | raise HandoffContractError( |
| 668 | f"The {label} file is bound to bundle_id {file_bundle_id!r} but the " |
| 669 | f"{noun} is {bundle.bundle_id!r}. {location_label}:\n" |
| 670 | f"{_searched_lines(searched)}\n" |
| 671 | f"Correct the bundle_id field in your {label} file to " |
| 672 | f"{bundle.bundle_id!r} and re-run this same leg." |
| 673 | ) |
| 674 | |
| 675 | |
| 676 | def _truncate_at_word(text: str, max_chars: int) -> str: |
| 677 | """Cap ``text`` at ``max_chars``, cutting back to a word boundary and |
| 678 | stripping trailing punctuation. Text within the cap passes through |
| 679 | untouched.""" |
| 680 | if len(text) <= max_chars: |
| 681 | return text |
| 682 | return text[:max_chars].rsplit(" ", 1)[0].rstrip(_TRUNCATE_STRIP_CHARS) |
| 683 | |
| 684 | |
| 685 | def _sanitized_name(raw: object) -> str | None: |
| 686 | """One whitespace-collapsed, punctuation-stripped, length-capped topic |
| 687 | name, or None for anything unusable (non-strings, and names that |
| 688 | sanitize to empty - e.g. emoji-only - count as per-row-absent).""" |
| 689 | if not isinstance(raw, str): |
| 690 | return None |
| 691 | name = " ".join(raw.split()).strip(_TRUNCATE_STRIP_CHARS) |
| 692 | name = _truncate_at_word(name, _NAME_MAX_CHARS) |
| 693 | if not any(char.isalnum() for char in name): |
| 694 | return None |
| 695 | return name |
| 696 | |
| 697 | |
| 698 | def _sanitized_angle(raw: object) -> str | None: |
| 699 | """One whitespace-collapsed, length-capped angle sentence, or None for |
| 700 | anything unusable. Non-strings are rejected outright, never coerced.""" |
| 701 | if not isinstance(raw, str): |
| 702 | return None |
| 703 | text = _truncate_at_word(" ".join(raw.split()), _ANGLE_MAX_CHARS) |
| 704 | return text or None |
| 705 | |
| 706 | |
| 707 | def _known_rows( |
| 708 | rows: list[Any], |
| 709 | known: set[str], |
| 710 | *, |
| 711 | row_label: str, |
| 712 | unknown_label: str, |
| 713 | ) -> Iterator[tuple[str, dict[str, Any]]]: |
| 714 | """Shared lenient per-row gate for host-authored files: skip non-object |
| 715 | rows, rows with no nomination id, and rows for unknown ids - warning on |
| 716 | each - and yield (row_id, row) for the rest.""" |
| 717 | for row in rows: |
| 718 | if not isinstance(row, dict): |
| 719 | _warn(f"skipping malformed {row_label} row (not an object)") |
| 720 | continue |
| 721 | row_id = str(row.get("id") or "").strip() |
| 722 | if not row_id: |
| 723 | _warn(f"skipping {row_label} row with no nomination id") |
| 724 | continue |
| 725 | if row_id not in known: |
| 726 | _warn(f"ignoring {unknown_label} for unknown nomination id {row_id!r}") |
| 727 | continue |
| 728 | yield row_id, row |
| 729 | |
| 730 | |
| 731 | def _clamped_worthiness(raw: object) -> int | None: |
| 732 | """Worthiness clamped to 0-100 integers; anything non-numeric is absent.""" |
| 733 | if isinstance(raw, bool): |
| 734 | return None |
| 735 | try: |
| 736 | value = float(raw) # type: ignore[arg-type] |
| 737 | except (TypeError, ValueError): |
| 738 | return None |
| 739 | return max(0, min(100, round(value))) |
| 740 | |
| 741 | |
| 742 | def read_judgments( |
| 743 | path: str | Path, |
| 744 | bundle: NominationsBundle, |
| 745 | *, |
| 746 | save_dir: str | Path | None = None, |
| 747 | config_dir: Path | None = None, |
| 748 | ) -> dict[str, HostJudgment]: |
| 749 | """Read the host judgments file for leg 2, keyed by nomination id. |
| 750 | |
| 751 | Strict at the top level (readable, valid JSON object, ``judgments`` list, |
| 752 | bundle_id bound to ``bundle``), lenient per row: an unknown id is warned |
| 753 | and ignored, a missing/unusable name or junk field is per-row-absent, and |
| 754 | worthiness is clamped to 0-100 integers. Nominations with no row at all |
| 755 | are simply missing from the mapping - use ``judgment_for`` to get the |
| 756 | ROW_ABSENT marker for them. |
| 757 | """ |
| 758 | payload = _load_host_file(path, "judgments") |
| 759 | _require_bundle_binding( |
| 760 | payload, bundle, label="judgments", save_dir=save_dir, config_dir=config_dir, |
| 761 | ) |
| 762 | rows = payload.get("judgments") |
| 763 | if not isinstance(rows, list): |
| 764 | raise HandoffContractError( |
| 765 | f"Judgments file {path} must carry a top-level \"judgments\" list." |
| 766 | ) |
| 767 | known = {entry.nomination_id for entry in bundle.nominations} |
| 768 | judgments: dict[str, HostJudgment] = {} |
| 769 | for row_id, row in _known_rows( |
| 770 | rows, known, row_label="judgments", unknown_label="judgment" |
| 771 | ): |
| 772 | # Only a real JSON boolean is a junk verdict: null, "false", 0, or |
| 773 | # any other non-bool value is per-row-absent (bundle heuristic), |
| 774 | # never coerced - bool("false") is True. |
| 775 | raw_junk = row.get("junk") |
| 776 | judgments[row_id] = HostJudgment( |
| 777 | name=_sanitized_name(row.get("name")), |
| 778 | junk=raw_junk if isinstance(raw_junk, bool) else None, |
| 779 | worthiness=_clamped_worthiness(row.get("worthiness")), |
| 780 | ) |
| 781 | return judgments |
| 782 | |
| 783 | |
| 784 | def judgment_for( |
| 785 | judgments: dict[str, HostJudgment], |
| 786 | nomination_id: str, |
| 787 | ) -> HostJudgment: |
| 788 | """The host's verdict for one nomination, or ROW_ABSENT when the host |
| 789 | omitted the row (caller falls back to the bundle's heuristic name/junk).""" |
| 790 | return judgments.get(nomination_id, ROW_ABSENT) |
| 791 | |
| 792 | |
| 793 | def read_angles( |
| 794 | path: str | Path | None, |
| 795 | bundle: NominationsBundle | PendingReport, |
| 796 | *, |
| 797 | save_dir: str | Path | None = None, |
| 798 | config_dir: Path | None = None, |
| 799 | ) -> dict[str, HostAngles]: |
| 800 | """Read the host angles file for leg 3, keyed by nomination id. |
| 801 | |
| 802 | ``bundle`` is the binding target: the finalize leg passes the pending |
| 803 | report (the bundle_id echo validates against it, and the known ids are |
| 804 | its surviving ``angle_inputs`` ids), while a NominationsBundle binds |
| 805 | against the full pool. A missing angles file is legal: ``path=None`` |
| 806 | returns an empty mapping and every topic ships without angles. When a |
| 807 | path is given the same strict-top-level / lenient-per-row rules as |
| 808 | judgments apply; angle sentences are word-boundary capped at 200 chars. |
| 809 | """ |
| 810 | if path is None: |
| 811 | return {} |
| 812 | payload = _load_host_file(path, "angles") |
| 813 | _require_bundle_binding( |
| 814 | payload, bundle, label="angles", save_dir=save_dir, config_dir=config_dir, |
| 815 | ) |
| 816 | rows = payload.get("angles") |
| 817 | if not isinstance(rows, list): |
| 818 | raise HandoffContractError( |
| 819 | f"Angles file {path} must carry a top-level \"angles\" list." |
| 820 | ) |
| 821 | known = ( |
| 822 | set(bundle.angle_inputs) |
| 823 | if isinstance(bundle, PendingReport) |
| 824 | else {entry.nomination_id for entry in bundle.nominations} |
| 825 | ) |
| 826 | angles: dict[str, HostAngles] = {} |
| 827 | for row_id, row in _known_rows( |
| 828 | rows, known, row_label="angles", unknown_label="angles" |
| 829 | ): |
| 830 | podcast = _sanitized_angle(row.get("podcast")) |
| 831 | x_article = _sanitized_angle(row.get("x_article")) |
| 832 | if podcast is None and x_article is None: |
| 833 | # No usable hook at all: treat the row as absent. |
| 834 | continue |
| 835 | angles[row_id] = HostAngles(podcast=podcast, x_article=x_article) |
| 836 | return angles |
| 837 | |
| 838 | |
| 839 | def resolve_name_collisions( |
| 840 | pairs: Sequence[tuple[pipeline.Nomination, str]], |
| 841 | ) -> list[str]: |
| 842 | """Re-run the nominate-stage casefold/entity-token collision rules over |
| 843 | host-applied names, returning one collision-free name per input pair in |
| 844 | order. |
| 845 | |
| 846 | Short host-judged names collide far more often than raw titles; a |
| 847 | colliding name gets the later nomination's strongest non-shared entity |
| 848 | token appended (``pipeline._disambiguated_topic_name``, fed synthetic |
| 849 | per-nomination clusters built from the seed items). Unlike the nominate |
| 850 | stage, a collision can never DROP a nomination here - the pool already |
| 851 | de-duplicated same-story clusters at leg 1 - so when no distinguishing |
| 852 | entity token exists the name falls back to an ordinal suffix. |
| 853 | """ |
| 854 | candidate_map: dict[str, schema.Candidate] = {} |
| 855 | clusters: list[schema.Cluster] = [] |
| 856 | for index, (nomination, _applied) in enumerate(pairs): |
| 857 | candidate_ids: list[str] = [] |
| 858 | for item_index, item in enumerate(nomination.items): |
| 859 | candidate_id = f"handoff-{index}-{item_index}" |
| 860 | candidate_map[candidate_id] = schema.Candidate( |
| 861 | candidate_id=candidate_id, |
| 862 | item_id=item.item_id, |
| 863 | source=item.source, |
| 864 | title=item.title, |
| 865 | url=item.url, |
| 866 | snippet=item.snippet, |
| 867 | subquery_labels=[], |
| 868 | native_ranks={}, |
| 869 | local_relevance=0.0, |
| 870 | freshness=0, |
| 871 | engagement=None, |
| 872 | source_quality=0.0, |
| 873 | rrf_score=0.0, |
| 874 | ) |
| 875 | candidate_ids.append(candidate_id) |
| 876 | clusters.append(schema.Cluster( |
| 877 | cluster_id=f"handoff-n{index}", |
| 878 | title=nomination.name, |
| 879 | candidate_ids=candidate_ids, |
| 880 | representative_ids=candidate_ids[:1], |
| 881 | sources=sorted({item.source for item in nomination.items}), |
| 882 | score=nomination.seed_score, |
| 883 | )) |
| 884 | |
| 885 | resolved_names: list[str] = [] |
| 886 | taken: dict[str, schema.Cluster] = {} |
| 887 | entity_counts_cache: dict[str, Counter] = {} |
| 888 | for index, (_nomination, applied) in enumerate(pairs): |
| 889 | cluster = clusters[index] |
| 890 | name = applied |
| 891 | key = name.casefold() |
| 892 | if key in taken: |
| 893 | resolved = pipeline._disambiguated_topic_name( |
| 894 | name, cluster, taken[key], candidate_map, entity_counts_cache, |
| 895 | taken, |
| 896 | ) |
| 897 | if resolved is None: |
| 898 | # Indistinguishable by content: keep the nomination anyway |
| 899 | # (distinct stories at leg 1) under an ordinal suffix. |
| 900 | suffix = 2 |
| 901 | while f"{name} {suffix}".casefold() in taken: |
| 902 | suffix += 1 |
| 903 | resolved = f"{name} {suffix}" |
| 904 | name = resolved |
| 905 | key = name.casefold() |
| 906 | taken[key] = cluster |
| 907 | resolved_names.append(name) |
| 908 | return resolved_names |
| 909 | |
| 910 | |
| 911 | def _one_line(text: str) -> str: |
| 912 | return " ".join(text.split()) |
| 913 | |
| 914 | |
| 915 | def build_host_digest(bundle: NominationsBundle) -> str: |
| 916 | """The host-facing judging digest for a nominations bundle: plain, |
| 917 | promptable text with one structural line per nomination (id, seed source |
| 918 | names, velocity/engagement signal) plus capped evidence lines (leader |
| 919 | title, leader snippet, strongest community comment - the surface the |
| 920 | engine judge used to see). Names the bundle file and instructs the host |
| 921 | to read its full evidence before judging. |
| 922 | |
| 923 | The evidence lines are scraped third-party text, so they are fenced the |
| 924 | way the deleted engine judge fenced its candidate block (the exact |
| 925 | ``rerank._fenced_untrusted_content`` fence: a security-notice header |
| 926 | stating the fenced content is data, never instructions, around |
| 927 | ``<untrusted_content>`` tags). The structural lines - nomination ids, |
| 928 | sources, signal, bundle path, judging instructions - stay outside the |
| 929 | fence.""" |
| 930 | location = str(bundle.path) if bundle.path is not None else ( |
| 931 | NOMINATIONS_BUNDLE_FILENAME |
| 932 | ) |
| 933 | domain_label = bundle.domain or "global trending (no domain filter)" |
| 934 | lines = [ |
| 935 | f"Discovery nominations awaiting host judgment " |
| 936 | f"({len(bundle.nominations)} topics).", |
| 937 | f"Domain: {domain_label} | window {bundle.from_date} -> " |
| 938 | f"{bundle.to_date} | tier {bundle.tier}", |
| 939 | f"Bundle file: {location} (bundle_id {bundle.bundle_id})", |
| 940 | "Read the bundle file's per-nomination evidence before judging; the " |
| 941 | "lines below are only a digest.", |
| 942 | "", |
| 943 | ] |
| 944 | evidence_lines: list[str] = [] |
| 945 | for entry in bundle.nominations: |
| 946 | items = entry.nomination.items |
| 947 | leader = items[0] if items else None |
| 948 | title = _one_line((leader.title if leader else "") or entry.nomination.name) |
| 949 | sources = ", ".join(entry.sources) if entry.sources else "unknown" |
| 950 | native_total = sum( |
| 951 | rerank.discovery_engagement_total(item) for item in items |
| 952 | ) |
| 953 | lines.append( |
| 954 | f"{entry.nomination_id} | sources: {sources} | " |
| 955 | f"signal: seed velocity {entry.nomination.seed_score:.1f}, " |
| 956 | f"{native_total:,.0f} native interactions" |
| 957 | ) |
| 958 | evidence_lines.append(f"- id: {entry.nomination_id}") |
| 959 | evidence_lines.append(f" title: {title[:_DIGEST_TITLE_MAX_CHARS]}") |
| 960 | snippet_text = _one_line( |
| 961 | (leader.snippet if leader else "") or entry.nomination.summary |
| 962 | ) |
| 963 | if snippet_text: |
| 964 | evidence_lines.append( |
| 965 | f" snippet: {snippet_text[:_DIGEST_SNIPPET_MAX_CHARS]}" |
| 966 | ) |
| 967 | top_comment = pipeline._best_community_comment(items) |
| 968 | if top_comment: |
| 969 | evidence_lines.append( |
| 970 | f" top comment: " |
| 971 | f"{_one_line(top_comment)[:_DIGEST_COMMENT_MAX_CHARS]}" |
| 972 | ) |
| 973 | if evidence_lines: |
| 974 | lines.append("") |
| 975 | lines.append(rerank._fenced_untrusted_content("\n".join(evidence_lines))) |
| 976 | return "\n".join(lines) |
| 977 |