| 1 | """U3 - enrichment stage: full pipeline pass per nominated topic. |
| 2 | |
| 3 | The fault-tolerance contract is the point of these tests: one topic failing or |
| 4 | running past the batch budget must never sink the others, and the batch never |
| 5 | raises. The U4 resume section pins the leg-2 tier parameterization: the |
| 6 | one-shot path keeps quick/240/3 while a deep-tier resume upgrades to |
| 7 | default/450/4 - both ways, so neither tier can leak into the other. |
| 8 | """ |
| 9 | |
| 10 | import inspect |
| 11 | import time |
| 12 | from unittest import mock |
| 13 | |
| 14 | from lib import discovery_handoff, pipeline, schema |
| 15 | |
| 16 | |
| 17 | def _nomination(name: str, score: float = 50.0) -> pipeline.Nomination: |
| 18 | return pipeline.Nomination(name=name, seed_score=score, items=[], summary=name) |
| 19 | |
| 20 | |
| 21 | def _report(topic: str) -> schema.Report: |
| 22 | return schema.Report( |
| 23 | topic=topic, |
| 24 | range_from="2026-06-10", |
| 25 | range_to="2026-07-10", |
| 26 | generated_at="2026-07-10T00:00:00+00:00", |
| 27 | provider_runtime=schema.ProviderRuntime( |
| 28 | reasoning_provider="none", |
| 29 | planner_model="deterministic", |
| 30 | rerank_model="deterministic", |
| 31 | ), |
| 32 | query_plan=schema.QueryPlan( |
| 33 | intent="factual", |
| 34 | freshness_mode="balanced_recent", |
| 35 | cluster_mode="none", |
| 36 | raw_topic=topic, |
| 37 | subqueries=[], |
| 38 | source_weights={}, |
| 39 | ), |
| 40 | clusters=[], |
| 41 | ranked_candidates=[], |
| 42 | items_by_source={}, |
| 43 | errors_by_source={}, |
| 44 | ) |
| 45 | |
| 46 | |
| 47 | def test_enrich_all_success_preserves_order(): |
| 48 | nominations = [_nomination("Topic A"), _nomination("Topic B"), _nomination("Topic C")] |
| 49 | |
| 50 | def fake_run(*, topic, **_kwargs): |
| 51 | return _report(topic) |
| 52 | |
| 53 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 54 | enriched = pipeline.enrich_nominations(nominations, config={}) |
| 55 | |
| 56 | assert [entry.nomination.name for entry in enriched] == ["Topic A", "Topic B", "Topic C"] |
| 57 | assert all(entry.report is not None for entry in enriched) |
| 58 | assert all(entry.error is None for entry in enriched) |
| 59 | |
| 60 | |
| 61 | def test_enrich_one_failure_does_not_sink_the_batch(): |
| 62 | nominations = [_nomination("Good"), _nomination("Bad"), _nomination("Also good")] |
| 63 | |
| 64 | def fake_run(*, topic, **_kwargs): |
| 65 | if topic == "Bad": |
| 66 | raise RuntimeError("upstream exploded") |
| 67 | return _report(topic) |
| 68 | |
| 69 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 70 | enriched = pipeline.enrich_nominations(nominations, config={}) |
| 71 | |
| 72 | by_name = {entry.nomination.name: entry for entry in enriched} |
| 73 | assert by_name["Good"].report is not None |
| 74 | assert by_name["Also good"].report is not None |
| 75 | assert by_name["Bad"].report is None |
| 76 | assert "upstream exploded" in (by_name["Bad"].error or "") |
| 77 | |
| 78 | |
| 79 | def test_enrich_budget_expiry_drops_slow_topic_to_nomination_only(): |
| 80 | nominations = [_nomination("Fast"), _nomination("Slow")] |
| 81 | |
| 82 | def fake_run(*, topic, **_kwargs): |
| 83 | if topic == "Slow": |
| 84 | time.sleep(5) |
| 85 | return _report(topic) |
| 86 | |
| 87 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 88 | enriched = pipeline.enrich_nominations( |
| 89 | nominations, config={}, budget_seconds=1.0, max_workers=2, |
| 90 | ) |
| 91 | |
| 92 | by_name = {entry.nomination.name: entry for entry in enriched} |
| 93 | assert by_name["Fast"].report is not None |
| 94 | assert by_name["Slow"].report is None |
| 95 | assert "budget" in (by_name["Slow"].error or "") |
| 96 | |
| 97 | |
| 98 | def test_enrich_runs_as_internal_subrun(): |
| 99 | """Sub-runs must use the internal_subrun lane (no library context, capped |
| 100 | inner workers) exactly like comparison-mode entity passes.""" |
| 101 | seen: dict[str, object] = {} |
| 102 | |
| 103 | def fake_run(*, topic, **kwargs): |
| 104 | seen.update(kwargs) |
| 105 | return _report(topic) |
| 106 | |
| 107 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 108 | pipeline.enrich_nominations([_nomination("One")], config={}) |
| 109 | |
| 110 | assert seen.get("internal_subrun") is True |
| 111 | |
| 112 | |
| 113 | def test_enrich_empty_nominations_returns_empty(): |
| 114 | assert pipeline.enrich_nominations([], config={}) == [] |
| 115 | |
| 116 | |
| 117 | def test_enrich_workers_are_daemon_threads(): |
| 118 | """Stragglers must not block interpreter exit: every enrichment worker runs |
| 119 | as a daemon thread (the P1 from PR #816 review - a hung sub-run kept the |
| 120 | process alive past the wall-clock budget with non-daemon executor threads).""" |
| 121 | import threading |
| 122 | |
| 123 | daemon_flags: list[bool] = [] |
| 124 | |
| 125 | def fake_run(*, topic, **_kwargs): |
| 126 | daemon_flags.append(threading.current_thread().daemon) |
| 127 | return _report(topic) |
| 128 | |
| 129 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 130 | pipeline.enrich_nominations([_nomination("One"), _nomination("Two")], config={}) |
| 131 | |
| 132 | assert daemon_flags and all(daemon_flags) |
| 133 | |
| 134 | |
| 135 | def test_enrich_concurrency_capped_by_semaphore(): |
| 136 | """Never more than max_workers sub-runs in flight.""" |
| 137 | import threading |
| 138 | |
| 139 | lock = threading.Lock() |
| 140 | state = {"active": 0, "peak": 0} |
| 141 | |
| 142 | def fake_run(*, topic, **_kwargs): |
| 143 | with lock: |
| 144 | state["active"] += 1 |
| 145 | state["peak"] = max(state["peak"], state["active"]) |
| 146 | time.sleep(0.05) |
| 147 | with lock: |
| 148 | state["active"] -= 1 |
| 149 | return _report(topic) |
| 150 | |
| 151 | nominations = [_nomination(f"T{i}") for i in range(6)] |
| 152 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 153 | enriched = pipeline.enrich_nominations(nominations, config={}, max_workers=2) |
| 154 | |
| 155 | assert state["peak"] <= 2 |
| 156 | assert all(entry.report is not None for entry in enriched) |
| 157 | |
| 158 | |
| 159 | def test_enrichment_reaches_all_sources_by_default(): |
| 160 | """No user source filter -> sub-runs get requested_sources=None, which is |
| 161 | what lets Techmeme, arXiv, YouTube, and Polymarket reach discovery despite |
| 162 | having no river feed of their own.""" |
| 163 | seen: dict[str, object] = {} |
| 164 | |
| 165 | def fake_run(*, topic, **kwargs): |
| 166 | seen.update(kwargs) |
| 167 | return _report(topic) |
| 168 | |
| 169 | raw = { |
| 170 | "id": "seed1", |
| 171 | "title": "AI agents breakthrough sweeps the industry", |
| 172 | "url": "https://example.com/seed1", |
| 173 | "hn_url": "https://news.ycombinator.com/item?id=1", |
| 174 | "author": "example", |
| 175 | "date": "2026-07-09", |
| 176 | "engagement": {"points": 900, "comments": 400}, |
| 177 | "relevance": 0.9, |
| 178 | } |
| 179 | with mock.patch.object(pipeline, "available_sources", return_value=["hackernews"]), \ |
| 180 | mock.patch.object(pipeline, "_fetch_discovery_source", return_value=([raw], None)), \ |
| 181 | mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 182 | pipeline.run_discover( |
| 183 | domain="AI agents", config={}, as_of_date="2026-07-10", enrich=True, |
| 184 | ) |
| 185 | |
| 186 | assert seen.get("internal_subrun") is True |
| 187 | assert seen.get("requested_sources") is None |
| 188 | |
| 189 | |
| 190 | def test_user_source_boundary_holds_through_enrichment(): |
| 191 | """--search reddit must bound the sub-runs too, not just the sweep.""" |
| 192 | seen: dict[str, object] = {} |
| 193 | |
| 194 | def fake_run(*, topic, **kwargs): |
| 195 | seen.update(kwargs) |
| 196 | return _report(topic) |
| 197 | |
| 198 | raw = { |
| 199 | "id": "seed1", |
| 200 | "title": "AI agents breakthrough sweeps the industry", |
| 201 | "url": "https://reddit.com/r/x/seed1", |
| 202 | "subreddit": "example", |
| 203 | "date": "2026-07-09", |
| 204 | "engagement": {"score": 900, "num_comments": 400}, |
| 205 | "selftext": "AI agents breakthrough", |
| 206 | "relevance": 0.9, |
| 207 | } |
| 208 | with mock.patch.object(pipeline, "available_sources", return_value=["reddit"]), \ |
| 209 | mock.patch.object(pipeline, "_fetch_discovery_source", return_value=([raw], None)), \ |
| 210 | mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 211 | pipeline.run_discover( |
| 212 | domain="AI agents", config={}, as_of_date="2026-07-10", |
| 213 | requested_sources=["reddit"], enrich=True, |
| 214 | enrich_requested_sources=["reddit"], |
| 215 | ) |
| 216 | |
| 217 | assert seen.get("requested_sources") == ["reddit"] |
| 218 | |
| 219 | |
| 220 | # --- U4 leg 2: resume enrichment tiers ----------------------------------------- |
| 221 | # The resume leg parameterizes enrich_nominations rather than editing the |
| 222 | # one-shot constants: deep-tier bundles get default/450(config)/4, shallow-tier |
| 223 | # bundles and the one-shot --discover path keep quick/240/3. Pinned BOTH ways |
| 224 | # so neither tier can leak into the other. |
| 225 | |
| 226 | |
| 227 | def _seed_item( |
| 228 | item_id: str, |
| 229 | source: str, |
| 230 | title: str, |
| 231 | *, |
| 232 | points: int = 300, |
| 233 | published_at: str = "2026-07-09", |
| 234 | ) -> schema.SourceItem: |
| 235 | engagement = ( |
| 236 | {"score": points, "num_comments": 40} |
| 237 | if source == "reddit" |
| 238 | else {"points": points, "comments": 40} |
| 239 | ) |
| 240 | return schema.SourceItem( |
| 241 | item_id=item_id, |
| 242 | source=source, |
| 243 | title=title, |
| 244 | body=title, |
| 245 | url=f"https://{source}.example/{item_id}", |
| 246 | published_at=published_at, |
| 247 | engagement=engagement, |
| 248 | snippet=f"Evidence about {title}", |
| 249 | ) |
| 250 | |
| 251 | |
| 252 | def _bundle_row( |
| 253 | nomination_id: str, |
| 254 | name: str, |
| 255 | items: list[schema.SourceItem], |
| 256 | *, |
| 257 | heuristic_junk: bool = False, |
| 258 | ) -> discovery_handoff.BundleNomination: |
| 259 | return discovery_handoff.BundleNomination( |
| 260 | nomination_id=nomination_id, |
| 261 | nomination=pipeline.Nomination( |
| 262 | name=name, |
| 263 | seed_score=50.0, |
| 264 | items=items, |
| 265 | summary=f"Summary of {name}", |
| 266 | junk_shape=heuristic_junk, |
| 267 | worthiness=None, |
| 268 | ), |
| 269 | cluster_id=f"c-{nomination_id}", |
| 270 | heuristic_name=name, |
| 271 | heuristic_junk=heuristic_junk, |
| 272 | sources=sorted({item.source for item in items}), |
| 273 | engagement_by_source={}, |
| 274 | ) |
| 275 | |
| 276 | |
| 277 | def _resume_bundle( |
| 278 | rows: list[discovery_handoff.BundleNomination], |
| 279 | *, |
| 280 | tier: str = "deep", |
| 281 | boundary: list[str] | None = None, |
| 282 | lookback_days: int = 30, |
| 283 | ) -> discovery_handoff.NominationsBundle: |
| 284 | return discovery_handoff.NominationsBundle( |
| 285 | schema_version=schema.DISCOVERY_NOMINATIONS_SCHEMA_VERSION, |
| 286 | bundle_id="cafef00dcafef00d", |
| 287 | generated_at="2026-07-10T00:00:00Z", |
| 288 | from_date="2026-06-10", |
| 289 | to_date="2026-07-10", |
| 290 | domain="AI agents", |
| 291 | tier=tier, |
| 292 | enrichment_source_boundary=boundary, |
| 293 | requested_sources=None, |
| 294 | lookback_days=lookback_days, |
| 295 | nominations=rows, |
| 296 | ) |
| 297 | |
| 298 | |
| 299 | def _enrich_spy(seen: dict): |
| 300 | def spy(nominations, **kwargs): |
| 301 | seen["nominations"] = list(nominations) |
| 302 | seen.update(kwargs) |
| 303 | return [pipeline.EnrichedTopic(nomination=n) for n in nominations] |
| 304 | return spy |
| 305 | |
| 306 | |
| 307 | def test_one_shot_discover_enrichment_stays_quick_tier(): |
| 308 | """Tier-leak pin, direction 1: the one-shot --discover path must keep the |
| 309 | quick/240/3 enrichment constants untouched by the resume-leg tiers.""" |
| 310 | seen: dict = {} |
| 311 | raw = { |
| 312 | "id": "seed1", |
| 313 | "title": "AI agents breakthrough sweeps the industry", |
| 314 | "url": "https://example.com/seed1", |
| 315 | "hn_url": "https://news.ycombinator.com/item?id=1", |
| 316 | "author": "example", |
| 317 | "date": "2026-07-09", |
| 318 | "engagement": {"points": 900, "comments": 400}, |
| 319 | "relevance": 0.9, |
| 320 | } |
| 321 | with mock.patch.object(pipeline, "available_sources", return_value=["hackernews"]), \ |
| 322 | mock.patch.object(pipeline, "_fetch_discovery_source", return_value=([raw], None)), \ |
| 323 | mock.patch.object(pipeline, "enrich_nominations", side_effect=_enrich_spy(seen)): |
| 324 | pipeline.run_discover( |
| 325 | domain="AI agents", config={}, as_of_date="2026-07-10", enrich=True, |
| 326 | ) |
| 327 | |
| 328 | assert seen.get("depth", pipeline.ENRICH_DEPTH) == "quick" |
| 329 | assert seen.get("max_workers", pipeline.ENRICH_MAX_WORKERS) == 3 |
| 330 | assert seen.get("budget_seconds", pipeline.ENRICH_BUDGET_SECONDS) == 240.0 |
| 331 | assert (pipeline.ENRICH_DEPTH, pipeline.ENRICH_MAX_WORKERS, |
| 332 | pipeline.ENRICH_BUDGET_SECONDS) == ("quick", 3, 240.0) |
| 333 | |
| 334 | |
| 335 | def test_resume_deep_tier_uses_default_depth_budget_and_workers(): |
| 336 | """Tier-leak pin, direction 2: a deep-tier bundle upgrades sub-runs to |
| 337 | default/450(default)/4 and scores against the bundle's window/boundary.""" |
| 338 | seen: dict = {} |
| 339 | bundle = _resume_bundle( |
| 340 | [_bundle_row("n1", "Topic A", [_seed_item("a1", "hackernews", "Topic A")])], |
| 341 | tier="deep", boundary=["reddit"], lookback_days=7, |
| 342 | ) |
| 343 | with mock.patch.object(pipeline, "enrich_nominations", side_effect=_enrich_spy(seen)): |
| 344 | pipeline.run_discover_resume(bundle, {}, config={}) |
| 345 | |
| 346 | assert seen["depth"] == "default" |
| 347 | assert seen["budget_seconds"] == 450.0 |
| 348 | assert seen["max_workers"] == 4 |
| 349 | assert seen["as_of_date"] == "2026-07-10" |
| 350 | assert seen["lookback_days"] == 7 |
| 351 | assert seen["requested_sources"] == ["reddit"] |
| 352 | |
| 353 | |
| 354 | def test_resume_shallow_tier_keeps_quick_constants(): |
| 355 | """A shallow-tier bundle enriches with today's one-shot quick constants.""" |
| 356 | seen: dict = {} |
| 357 | bundle = _resume_bundle( |
| 358 | [_bundle_row("n1", "Topic A", [_seed_item("a1", "hackernews", "Topic A")])], |
| 359 | tier="shallow", |
| 360 | ) |
| 361 | with mock.patch.object(pipeline, "enrich_nominations", side_effect=_enrich_spy(seen)): |
| 362 | pipeline.run_discover_resume(bundle, {}, config={}) |
| 363 | |
| 364 | assert seen["depth"] == pipeline.ENRICH_DEPTH == "quick" |
| 365 | assert seen["budget_seconds"] == pipeline.ENRICH_BUDGET_SECONDS == 240.0 |
| 366 | assert seen["max_workers"] == pipeline.ENRICH_MAX_WORKERS == 3 |
| 367 | |
| 368 | |
| 369 | def test_resume_budget_knob_reads_config_only_never_os_environ(monkeypatch): |
| 370 | """LAST30DAYS_ENRICH_BUDGET_SECONDS comes from the RESOLVED config dict; |
| 371 | a bare os.environ value that never went through env.get_config is |
| 372 | invisible to the pipeline (no bare os.environ reads in lib/).""" |
| 373 | monkeypatch.setenv("LAST30DAYS_ENRICH_BUDGET_SECONDS", "77") |
| 374 | seen: dict = {} |
| 375 | bundle = _resume_bundle( |
| 376 | [_bundle_row("n1", "Topic A", [_seed_item("a1", "hackernews", "Topic A")])], |
| 377 | ) |
| 378 | with mock.patch.object(pipeline, "enrich_nominations", side_effect=_enrich_spy(seen)): |
| 379 | pipeline.run_discover_resume(bundle, {}, config={}) |
| 380 | assert seen["budget_seconds"] == 450.0 |
| 381 | |
| 382 | seen.clear() |
| 383 | with mock.patch.object(pipeline, "enrich_nominations", side_effect=_enrich_spy(seen)): |
| 384 | pipeline.run_discover_resume( |
| 385 | bundle, {}, config={"LAST30DAYS_ENRICH_BUDGET_SECONDS": "333"}, |
| 386 | ) |
| 387 | assert seen["budget_seconds"] == 333.0 |
| 388 | |
| 389 | |
| 390 | def test_resume_budget_env_file_seam(tmp_path, monkeypatch): |
| 391 | """The knob rides the same .env-file seam as every other config value |
| 392 | (mirrors the queue-toggle seam tests in test_discover_mode.py).""" |
| 393 | from lib import env |
| 394 | |
| 395 | monkeypatch.delenv("LAST30DAYS_ENRICH_BUDGET_SECONDS", raising=False) |
| 396 | env_file = tmp_path / "config.env" |
| 397 | env_file.write_text("LAST30DAYS_ENRICH_BUDGET_SECONDS=333\n", encoding="utf-8") |
| 398 | monkeypatch.setattr(env, "CONFIG_FILE", env_file) |
| 399 | monkeypatch.chdir(tmp_path) |
| 400 | with mock.patch.object(env, "_load_keychain", return_value={}), \ |
| 401 | mock.patch.object(env, "_load_pass", return_value={}): |
| 402 | config = env.get_config() |
| 403 | assert config["LAST30DAYS_ENRICH_BUDGET_SECONDS"] == "333" |
| 404 | |
| 405 | seen: dict = {} |
| 406 | bundle = _resume_bundle( |
| 407 | [_bundle_row("n1", "Topic A", [_seed_item("a1", "hackernews", "Topic A")])], |
| 408 | ) |
| 409 | with mock.patch.object(pipeline, "enrich_nominations", side_effect=_enrich_spy(seen)): |
| 410 | pipeline.run_discover_resume(bundle, {}, config=config) |
| 411 | assert seen["budget_seconds"] == 333.0 |
| 412 | |
| 413 | |
| 414 | def test_resume_budget_parser_rejects_garbage_and_nonpositive(): |
| 415 | default = pipeline.RESUME_DEEP_ENRICH_BUDGET_SECONDS |
| 416 | assert default == 450.0 |
| 417 | assert pipeline._resume_enrich_budget_seconds({}) == default |
| 418 | assert pipeline._resume_enrich_budget_seconds( |
| 419 | {"LAST30DAYS_ENRICH_BUDGET_SECONDS": ""}) == default |
| 420 | assert pipeline._resume_enrich_budget_seconds( |
| 421 | {"LAST30DAYS_ENRICH_BUDGET_SECONDS": "not-a-number"}) == default |
| 422 | assert pipeline._resume_enrich_budget_seconds( |
| 423 | {"LAST30DAYS_ENRICH_BUDGET_SECONDS": "0"}) == default |
| 424 | assert pipeline._resume_enrich_budget_seconds( |
| 425 | {"LAST30DAYS_ENRICH_BUDGET_SECONDS": "-5"}) == default |
| 426 | assert pipeline._resume_enrich_budget_seconds( |
| 427 | {"LAST30DAYS_ENRICH_BUDGET_SECONDS": "600"}) == 600.0 |
| 428 | assert pipeline._resume_enrich_budget_seconds( |
| 429 | {"LAST30DAYS_ENRICH_BUDGET_SECONDS": 300}) == 300.0 |
| 430 | |
| 431 | |
| 432 | def test_enrich_budget_expiry_downgrades_at_deep_tier(): |
| 433 | """The wall-clock downgrade contract holds unchanged under the deep-tier |
| 434 | parameters (default depth, 4 workers): stragglers become nomination-only.""" |
| 435 | nominations = [_nomination("Fast"), _nomination("Slow")] |
| 436 | depths: list[str] = [] |
| 437 | |
| 438 | def fake_run(*, topic, depth, **_kwargs): |
| 439 | depths.append(depth) |
| 440 | if topic == "Slow": |
| 441 | time.sleep(5) |
| 442 | return _report(topic) |
| 443 | |
| 444 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 445 | enriched = pipeline.enrich_nominations( |
| 446 | nominations, config={}, depth="default", |
| 447 | budget_seconds=1.0, max_workers=4, |
| 448 | ) |
| 449 | |
| 450 | by_name = {entry.nomination.name: entry for entry in enriched} |
| 451 | assert by_name["Fast"].report is not None |
| 452 | assert by_name["Slow"].report is None |
| 453 | assert "budget" in (by_name["Slow"].error or "") |
| 454 | assert depths and all(depth == "default" for depth in depths) |
| 455 | |
| 456 | |
| 457 | def test_enrich_workers_are_daemon_threads_at_deep_tier(): |
| 458 | """Daemon-thread containment holds at the deep tier too: a hung default- |
| 459 | depth sub-run must never block interpreter exit.""" |
| 460 | import threading |
| 461 | |
| 462 | daemon_flags: list[bool] = [] |
| 463 | |
| 464 | def fake_run(*, topic, **_kwargs): |
| 465 | daemon_flags.append(threading.current_thread().daemon) |
| 466 | return _report(topic) |
| 467 | |
| 468 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 469 | pipeline.enrich_nominations( |
| 470 | [_nomination("One"), _nomination("Two")], config={}, |
| 471 | depth="default", max_workers=4, |
| 472 | ) |
| 473 | |
| 474 | assert daemon_flags and all(daemon_flags) |
| 475 | |
| 476 | |
| 477 | def test_enrich_concurrency_capped_at_deep_tier(): |
| 478 | """Never more than the deep tier's 4 sub-runs in flight.""" |
| 479 | import threading |
| 480 | |
| 481 | lock = threading.Lock() |
| 482 | state = {"active": 0, "peak": 0} |
| 483 | |
| 484 | def fake_run(*, topic, **_kwargs): |
| 485 | with lock: |
| 486 | state["active"] += 1 |
| 487 | state["peak"] = max(state["peak"], state["active"]) |
| 488 | time.sleep(0.05) |
| 489 | with lock: |
| 490 | state["active"] -= 1 |
| 491 | return _report(topic) |
| 492 | |
| 493 | nominations = [_nomination(f"T{i}") for i in range(9)] |
| 494 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 495 | enriched = pipeline.enrich_nominations( |
| 496 | nominations, config={}, depth="default", max_workers=4, |
| 497 | ) |
| 498 | |
| 499 | assert state["peak"] <= 4 |
| 500 | assert all(entry.report is not None for entry in enriched) |
| 501 | |
| 502 | |
| 503 | def test_enrichment_path_never_uses_thread_pool_executor(): |
| 504 | """Executor threads are non-daemon and joined at shutdown, defeating the |
| 505 | wall-clock budget (docs/solutions/logic-errors/non-daemon-executor-threads- |
| 506 | defeat-wall-clock-budget.md). The enrichment batch must stay on the |
| 507 | daemon-thread + Semaphore + queue pattern.""" |
| 508 | source = inspect.getsource(pipeline.enrich_nominations) |
| 509 | # The comment naming the anti-pattern is fine; constructing one is not. |
| 510 | assert "ThreadPoolExecutor(" not in source |
| 511 | assert "daemon=True" in source |
| 512 | assert "Semaphore" in source |
| 513 | |
| 514 | |
| 515 | def test_host_judged_name_becomes_enrichment_sub_run_topic(): |
| 516 | """Relocated from the retired engine-judge suite, retargeted to the |
| 517 | judgments-file path: the host's applied name IS the enrichment sub-run |
| 518 | topic (the nomination name is what run() researches).""" |
| 519 | seen: dict = {} |
| 520 | |
| 521 | def fake_run(*, topic, **kwargs): |
| 522 | seen["topic"] = topic |
| 523 | seen.update(kwargs) |
| 524 | return _report(topic) |
| 525 | |
| 526 | bundle = _resume_bundle([ |
| 527 | _bundle_row( |
| 528 | "n1", |
| 529 | "Google is updating Gemma 4 chat templates", |
| 530 | [_seed_item("hn1", "hackernews", |
| 531 | "Google is updating Gemma 4 chat templates", |
| 532 | points=900)], |
| 533 | ), |
| 534 | ]) |
| 535 | judgments = { |
| 536 | "n1": discovery_handoff.HostJudgment( |
| 537 | name="Gemma 4 Flash Attention", junk=False, worthiness=88, |
| 538 | ), |
| 539 | } |
| 540 | with mock.patch.object(pipeline, "run", side_effect=fake_run): |
| 541 | result = pipeline.run_discover_resume(bundle, judgments, config={}) |
| 542 | |
| 543 | assert seen["topic"] == "Gemma 4 Flash Attention" |
| 544 | assert seen.get("internal_subrun") is True |
| 545 | assert [topic.name for topic in result.report.topics] == ["Gemma 4 Flash Attention"] |
| 546 |