A fired page is the work, not the new leads on it: a page of familiar profiles no longer ends the job with the frontier wide open

Author: eracle

openoutreach/core/pipeline/discover.py

@@ -151,12 +151,19 @@ def _handle_empty(node, offset: int, page) -> str | None:
     return verdict
 
 
-def discover(campaign, qualifier=None) -> int:
-    """Fire frontier nodes until one returns leads. Returns the count of new Leads.
-
-    ``0`` means the frontier is spanned (nothing unfired and nothing left to deepen) or a
-    fetch was unavailable — both best-effort, since a provider outage must not fail the
-    enclosing task. A provider *refusal* is the exception and propagates
+def discover(campaign, qualifier=None) -> bool:
+    """Fire frontier nodes until one returns a page. Returns whether the walk moved.
+
+    **The answer is "did a page come back", not "how many leads were new"** — bug 8 again,
+    one level up. ``_harvest`` counts *newly created* leads, and a page of profiles this
+    campaign already holds counts 0 while still being a perfectly good page: the node's
+    offset advanced, its children joined the frontier, and the next pass draws the next
+    node. Reporting that as nothing left to do stopped a live run dead with 100 rows in
+    hand and the frontier wide open (``top_up`` reads this return as *was there work*).
+
+    ``False`` means the frontier is spanned (nothing unfired and nothing left to deepen)
+    or a fetch was unavailable — both best-effort, since a provider outage must not fail
+    the enclosing task. A provider *refusal* is the exception and propagates
     (``_fetch``): a rejected key must never be read as a query that matches nobody.
     ``qualifier`` is accepted and ignored: the GP no longer selects
     queries (§13), and the parameter stays only so the call sites in ``pools`` read the
@@ -172,9 +179,9 @@ def discover(campaign, qualifier=None) -> int:
     from openoutreach.enrichment import bettercontact
 
     if not bettercontact.is_configured():
-        return 0
+        return False
     if not (campaign.product_docs or campaign.campaign_target):
-        return 0
+        return False
 
     logger.info(colored(f"▶ discover · {campaign}", "blue", attrs=["bold"]))
 
@@ -189,16 +196,16 @@ def discover(campaign, qualifier=None) -> int:
                 f"■ no more people left to find for {campaign} — every search this "
                 f"campaign knows how to make is exhausted", "blue"))
             logger.debug("frontier spanned · %d node(s) retired this pass", retired)
-            return 0
+            return False
 
         offset = node.next_offset
         page = _fetch(node, offset)
         if page is None:
-            return 0  # outage: the node keeps its place, the caller carries on
+            return False  # outage: the node keeps its place, the caller carries on
 
         if not page.leads:
             if _handle_empty(node, offset, page) is None:
-                return 0  # transport artifact — re-firing now would just repeat it
+                return False  # transport artifact — re-firing now would just repeat it
             retired += 1
             continue
 
@@ -210,4 +217,4 @@ def discover(campaign, qualifier=None) -> int:
             glyph="✓", color="green"))
         logger.debug("%s", step_line(
             "frontier", f"+{grown} node(s) from this page", glyph="+", color="cyan"))
-        return created
+        return True

openoutreach/core/pipeline/top_up.py

@@ -80,15 +80,21 @@ def top_up(campaign, qualifier: BayesianQualifier) -> bool:
         consumable = _consumable_candidates(qualifier, candidates)
         if consumable:
             return run_qualification(campaign, qualifier, candidates=consumable) is not None
-        return discover(campaign, qualifier) > 0
+        # A fired page is the work, however much of it we had already seen. Reading
+        # *new leads* here is what stopped a live run with 100 rows in hand: the page
+        # was all familiar profiles, so discovery reported nothing and the whole job
+        # ended `goal_unreached` with the frontier untouched below it.
+        return discover(campaign, qualifier)
 
     # Explore — label the most informative lead we have. The GP is fitted here, so it
     # ranks the pool and there is a best lead to pick; an empty pool is the one case
     # with no lead to label, so page one in first.
     if not candidates:
-        if discover(campaign, qualifier) <= 0:
+        if not discover(campaign, qualifier):
             return False
         candidates = fetch_qualification_candidates(campaign)
+        if not candidates:
+            return True  # every profile on that page was already ours — still a move
     return run_qualification(campaign, qualifier, candidates=candidates) is not None
 
 

tests/test_discovery_wiring.py

@@ -69,13 +69,13 @@ class TestGates:
     def test_no_finder_key_is_a_no_op(self, db):
         SiteConfig.objects.update(bettercontact_api_key="")
         with patch.object(discover_mod, "_fetch") as fetch:
-            assert discover(_campaign()) == 0
+            assert discover(_campaign()) is False
         fetch.assert_not_called()
 
     def test_no_icp_text_is_a_no_op(self, db):
         c = _campaign(product_docs="", campaign_target="")
         with patch.object(discover_mod, "_fetch") as fetch:
-            assert discover(c) == 0
+            assert discover(c) is False
         fetch.assert_not_called()
 
 
@@ -86,9 +86,9 @@ def test_a_productive_page_creates_leads_and_advances_the_node(self, db):
         page = Page([_row()], 9027)
 
         with patch.object(discover_mod, "_fetch", return_value=page):
-            created = discover(c)
+            assert discover(c) is True
 
-        assert created == 1
+        assert Lead.objects.count() == 1
         node.refresh_from_db()
         assert node.state == QueryNode.State.FIRED
         assert node.next_offset == select.DISCOVERY_PAGE_SIZE
@@ -98,14 +98,16 @@ def test_a_productive_page_creates_leads_and_advances_the_node(self, db):
     def test_a_page_of_duplicates_is_not_a_stall(self, db):
         # Bug 8: `_harvest` counts *newly created* leads, and a page of already-seen
         # profiles used to read as "nothing left to do" and halt the engine with the
-        # frontier wide open. The node must still advance.
+        # frontier wide open. The node must still advance, and the walk must report
+        # that it moved — `top_up` stops the whole job on a False here.
         c = _campaign()
         node = _node(c, [("lead_job_title", "founder")])
         Lead.objects.create(profile_url="https://linkedin.com/in/a", profile_text="x")
 
         with patch.object(discover_mod, "_fetch", return_value=Page([_row()], 10)):
-            assert discover(c) == 0
+            assert discover(c) is True
 
+        assert Lead.objects.count() == 1  # nothing new created — and that is fine
         node.refresh_from_db()
         assert node.state == QueryNode.State.FIRED
         assert node.next_offset == select.DISCOVERY_PAGE_SIZE
@@ -147,7 +149,7 @@ def test_a_positive_count_with_no_rows_is_a_transport_artifact(self, db):
 
         with patch.object(discover_mod, "_fetch",
                           return_value=Page([], 71403396)) as fetch:
-            assert discover(c) == 0
+            assert discover(c) is False
 
         assert fetch.call_count == 1  # not even retried — the count already answered
         node.refresh_from_db()
@@ -172,15 +174,15 @@ def test_the_loop_tries_the_next_node_after_a_dead_one(self, db):
 
         pages = [Page([], 0), Page([_row()], 10)]
         with patch.object(discover_mod, "_fetch", side_effect=pages):
-            assert discover(c) == 1
+            assert discover(c) is True
 
         assert QueryNode.objects.filter(campaign=c, state=QueryNode.State.DEAD).count() == 1
 
-    def test_saturation_returns_zero(self, db):
+    def test_saturation_is_the_one_false(self, db):
         c = _campaign()
         _node(c, [("lead_job_title", "a")], state=QueryNode.State.DRAINED)
         with patch.object(discover_mod, "_fetch") as fetch:
-            assert discover(c) == 0
+            assert discover(c) is False
         fetch.assert_not_called()
 
 
@@ -192,7 +194,7 @@ def test_an_outage_leaves_the_node_on_the_frontier(self, db):
         node = _node(c, [("lead_job_title", "founder")])
 
         with patch.object(discover_mod, "_fetch", return_value=None):
-            assert discover(c) == 0
+            assert discover(c) is False
 
         node.refresh_from_db()
         assert node.state == QueryNode.State.FRONTIER

tests/test_top_up.py

@@ -16,7 +16,7 @@ def test_the_cold_phase_names_itself_and_the_anchor_progress(campaign, caplog):
     qualifier.set_anchors(np.random.RandomState(0).rand(3, 8))
 
     with (
-        patch("openoutreach.core.pipeline.top_up.discover", return_value=0),
+        patch("openoutreach.core.pipeline.top_up.discover", return_value=False),
         patch("openoutreach.core.pipeline.top_up.fetch_qualification_candidates",
               return_value=[]),
         caplog.at_level("INFO"),
@@ -69,9 +69,32 @@ def test_exploit_discovers_rather_than_qualifying_below_the_gate(campaign):
         patch("openoutreach.core.pipeline.top_up.fetch_qualification_candidates",
               return_value=[_Candidate()]),
         patch("openoutreach.core.pipeline.top_up.run_qualification") as qualify,
-        patch("openoutreach.core.pipeline.top_up.discover", return_value=10) as discover,
+        patch("openoutreach.core.pipeline.top_up.discover", return_value=True) as discover,
     ):
         assert top_up(campaign, qualifier) is True
 
     assert not qualify.called
     assert discover.called
+
+
+@pytest.mark.django_db
+def test_explore_counts_a_page_of_familiar_profiles_as_work(campaign):
+    """A fired page is the unit of work, whatever fraction of it was new.
+
+    A live run ended `goal_unreached` on this shape: the page came back 100 rows all
+    already ours, so it left no candidate to label — which is not the same fact as a
+    spanned frontier, and must not stop the job with the walk one node in."""
+    qualifier = BayesianQualifier(embedding_dim=8)
+    rng = np.random.RandomState(0)
+    qualifier.warm_start(rng.rand(7, 8), np.array([1, 1, 1, 1, 0, 0, 0]))
+    assert qualifier.acquisition_mode() != "exploit (p)"
+
+    with (
+        patch("openoutreach.core.pipeline.top_up.fetch_qualification_candidates",
+              return_value=[]),
+        patch("openoutreach.core.pipeline.top_up.run_qualification") as qualify,
+        patch("openoutreach.core.pipeline.top_up.discover", return_value=True),
+    ):
+        assert top_up(campaign, qualifier) is True
+
+    assert not qualify.called