fix(metadata): accept complete catalogs with stale summary counts

src/pullbox/services/metadata_series_adoption.py

@@ -163,8 +163,6 @@ async def fetch_source_series_bundle(
                 raise SeriesAdoptionError(
                     "The series profile belongs to a different source identity."
                 )
-            if profile.issue_count is not None and profile.issue_count > max_issues:
-                raise SeriesAdoptionError("The issue catalog exceeds the interactive add limit.")
             issues: list[ProviderIssueRead] = []
             seen: set[str] = set()
             page = 1
@@ -181,9 +179,8 @@ async def fetch_source_series_bundle(
                 current = result.data
                 if total is None:
                     total = current.total
-                if current.total != total or (
-                    profile.issue_count is not None and profile.issue_count != total
-                ):
+                # Summary counts can lag; completeness comes from the paginated catalog.
+                if current.total != total:
                     raise SeriesCatalogCountError(
                         "The provider catalog count changed. Retry the add."
                     )
@@ -208,6 +205,14 @@ async def fetch_source_series_bundle(
                 if current.next_page is None:
                     if len(issues) != total:
                         raise SeriesAdoptionError("The issue catalog is incomplete. Retry the add.")
+                    if profile.issue_count is not None and profile.issue_count != total:
+                        logger.info(
+                            "metadata_series_summary_count_disagreement",
+                            source=source.value,
+                            external_id=identifier,
+                            summary_count=profile.issue_count,
+                            catalog_count=total,
+                        )
                     return SourceSeriesBundle(
                         profile, tuple(issues), source_revision, total, started_at
                     )
@@ -285,7 +290,6 @@ def _validate_bundle(bundle: SourceSeriesBundle) -> list[tuple[float, str]]:
         or len(catalog) > 10000
         or type(bundle.catalog_total) is not int
         or bundle.catalog_total != len(catalog)
-        or (profile.issue_count is not None and profile.issue_count != len(catalog))
     ):
         raise SeriesAdoptionError("The series profile or issue catalog is incomplete.")
     if bundle.excluded_issues:

tests/api/test_source_series_add_api.py

@@ -43,13 +43,12 @@ async def source_add_setup(authenticated_client, sec_db, monkeypatch, tmp_path):
         root_id = root.id
     calls = []
     responses = []
+    profile = {**series_row(8), "name": "Fixture 8", "cv_id": None, "gcd_id": None}
 
     def handle(request):
         calls.append(request)
         if request.url.path == "/api/series/8/":
-            return httpx.Response(
-                200, json={**series_row(8), "name": "Fixture 8", "cv_id": None, "gcd_id": None}
-            )
+            return httpx.Response(200, json=profile)
         assert request.url.path == "/api/series/8/issue_list/"
         if responses:
             return responses.pop(0)
@@ -106,6 +105,7 @@ async def covers_dir(_session):
         },
         "calls": calls,
         "responses": responses,
+        "profile": profile,
         "events": events,
         "root": root_path,
     }
@@ -136,6 +136,46 @@ async def test_add_native_series_uses_full_server_catalog_and_is_repeat_safe(
         assert await session.scalar(select(func.count()).select_from(Issue)) == 3
 
 
+@pytest.mark.parametrize("invalid_after_preview", [False, True])
+async def test_stale_summary_preview_and_final_add_revalidate_the_catalog(
+    authenticated_client, source_add_setup, sec_db, invalid_after_preview
+):
+    fixture = source_add_setup
+    fixture["profile"]["issue_count"] = 1
+    complete = envelope([issue_row(100, "1"), issue_row(101, "2")])
+    fixture["responses"].append(httpx.Response(200, json=complete))
+    preview = await authenticated_client.post(
+        "/api/v1/metadata/series/preview",
+        json={key: value for key, value in fixture["body"].items() if key != "source_revision"},
+        headers=csrf(authenticated_client),
+    )
+    assert preview.status_code == 200, preview.text
+    assert preview.json()["issues"]["data"]["total"] == 2
+    assert preview.json()["series"]["data"]["issue_count"] == 1
+    assert not list(fixture["root"].iterdir())
+    latest = {**complete, "count": 3} if invalid_after_preview else complete
+    fixture["responses"].append(httpx.Response(200, json=latest))
+    added = await authenticated_client.post(
+        "/api/v1/series", json=fixture["body"], headers=csrf(authenticated_client)
+    )
+    assert added.status_code == (409 if invalid_after_preview else 201), added.text
+    assert [call.url.path for call in fixture["calls"]] == [
+        "/api/series/8/",
+        "/api/series/8/issue_list/",
+        "/api/series/8/",
+        "/api/series/8/issue_list/",
+    ]
+    async with sec_db() as session:
+        assert await session.scalar(select(func.count()).select_from(Issue)) == (
+            0 if invalid_after_preview else 2
+        )
+    if invalid_after_preview:
+        assert not fixture["events"] and not list(fixture["root"].iterdir())
+    else:
+        assert added.json()["issue_count"] == 2
+        assert len(fixture["events"]) == 1
+
+
 @pytest.mark.parametrize(
     "changes",
     [

tests/integration/metadata_identity/test_series_adoption.py

@@ -115,6 +115,22 @@ async def test_new_add_returns_canonical_snapshots_matching_persisted_metadata(i
             assert item.values.issue_number_text == issue.issue_number_text
 
 
+async def test_stale_summary_adopts_complete_catalog_and_derives_saved_count(identity_probe_db):
+    _, factory, _ = identity_probe_db
+    data = bundle(numbers=("1", "2"))
+    data.series.issue_count = 1
+    async with factory.begin() as session:
+        try:
+            result = await adopt_source_series_bundle(session, data)
+        except SeriesAdoptionError as exc:
+            pytest.fail(f"A stale summary cannot veto the complete catalog: {exc}")
+        assert result.series.issue_count == result.snapshot.values.issue_count == 2
+        origin = next(item for item in result.snapshot.origins if item.field == "issue_count")
+        assert origin.derivation == "catalog"
+        assert await session.scalar(select(func.count()).select_from(Issue)) == 2
+        assert data.series.issue_count == 1
+
+
 async def test_representative_metron_cover_survives_add_and_session_reload(identity_probe_db):
     _, factory, _ = identity_probe_db
     adapter = CatalogAdapter(1)

tests/integration/metadata_identity/test_series_refresh.py

@@ -841,7 +841,7 @@ def from_source(source):
     return factory, series_id, original, adapters, instance
 
 
-async def test_refresh_falls_back_from_stale_local_count_to_exact_api_catalog(
+async def test_refresh_accepts_complete_catalog_with_stale_summary_without_fallback(
     comicvine_count_disagreement,
 ):
     from pullbox.services.metadata_series_refresh import refresh_series_catalog_from_sources
@@ -853,24 +853,45 @@ async def test_refresh_falls_back_from_stale_local_count_to_exact_api_catalog(
         assert result.series.issue_count == 5
         assert result.created_issue_ids == ()
         assert result.series.path == "/reference/Civil War - Unmasked"
-        assert [(o.source, o.status) for o in result.outcomes] == [
-            (Source.COMICVINE_LOCAL, SourceStatus.INCOMPATIBLE_RESPONSE)
-        ]
+        assert result.outcomes == ()
         await session.commit()
     async with factory() as session:
         issues = list(await session.scalars(select(Issue).order_by(Issue.id)))
         assert [(i.id, i.comicvine_id, i.issue_number_text, i.status) for i in issues] == original
         assert set(await session.scalars(select(IssueExternalIdentity.identity_namespace))) == {
             Namespace.COMICVINE
         }
-    assert ("issues", "42", 1) in adapters[Source.COMICVINE_API].calls
+    assert not any(call[0] == "issues" for call in adapters[Source.COMICVINE_API].calls)
 
 
 @pytest.mark.parametrize("fallback", ["disabled", "rate_limited"])
-async def test_count_mismatch_without_consistent_fallback_is_actionable_and_atomic(
+async def test_changing_page_total_without_consistent_fallback_is_actionable_and_atomic(
     comicvine_count_disagreement, fallback
 ):
     factory, series_id, original, adapters, instance = comicvine_count_disagreement
+    local = adapters[Source.COMICVINE_LOCAL]
+
+    async def changing_pages(external_id, *, page=1, validator=None):
+        start, stop = (0, 100) if page == 1 else (100, 102)
+        return MetadataFetch(
+            status=SourceStatus.OK,
+            data=MetadataPage(
+                results=[
+                    local.data.issues[0].model_copy(
+                        update={
+                            "external_id": str(1000 + index),
+                            "issue_number_text": str(index + 1),
+                            "issue_number_key": str(index + 1),
+                        }
+                    )
+                    for index in range(start, stop)
+                ],
+                total=101 if page == 1 else 102,
+                next_page=2 if page == 1 else None,
+            ),
+        )
+
+    local.issues = changing_pages
     if fallback == "disabled":
         instance.runtime[Source.COMICVINE_API].policy.enabled = False
     else:

tests/unit/test_metadata_series_adoption_fetch.py

@@ -150,20 +150,46 @@ async def test_cross_page_consistency_is_required(mutation):
         await fetch(adapter)
 
 
-async def test_series_and_issue_count_disagreement_is_not_complete():
-    adapter = CatalogAdapter(0)
-    adapter.pages[1] = MetadataFetch(
-        status=SourceStatus.OK, data=MetadataPage(results=[issue_row()], total=1)
+@pytest.mark.parametrize("summary_count", [0, 1, 3, 10001, None])
+@pytest.mark.parametrize("catalog_count", [2, 101])
+async def test_stale_summary_does_not_reject_a_complete_catalog(summary_count, catalog_count):
+    adapter = CatalogAdapter(catalog_count)
+    profile = row(adapter.source).model_copy(update={"issue_count": summary_count})
+    try:
+        result = await fetch(adapter, profile=profile)
+    except SeriesAdoptionError as exc:
+        pytest.fail(f"A complete catalog must not be rejected for its stale summary: {exc}")
+    assert result.catalog_total == len(result.issues) == catalog_count
+    assert result.series.issue_count == summary_count, "Keep the original provider evidence"
+    assert adapter.calls == [
+        ("issues", "42", page) for page in range(1, (catalog_count + 99) // 100 + 1)
+    ]
+
+
+@pytest.mark.parametrize("problem", ["changed_total", "incomplete", "repeat_id", "wrong_parent"])
+async def test_stale_summary_never_excuses_an_invalid_catalog(problem):
+    adapter = CatalogAdapter(101)
+    profile = row(adapter.source).model_copy(update={"issue_count": 1})
+    item = issue_row(external_id="1100", issue_number_text="101", issue_number_key="101")
+    total = 102 if problem == "changed_total" else 101
+    if problem == "repeat_id":
+        item.external_id = "1000"
+    elif problem == "wrong_parent":
+        item.series_external_id = "99"
+    adapter.pages[2] = MetadataFetch(
+        status=SourceStatus.OK,
+        data=MetadataPage(results=[] if problem == "incomplete" else [item], total=total),
     )
-    with pytest.raises(SeriesAdoptionError, match="count"):
-        await fetch(adapter)
+    with pytest.raises(SeriesAdoptionError):
+        await fetch(adapter, profile=profile)
+    assert adapter.calls[-1] == ("issues", "42", 2)
 
 
-async def test_resource_cap_stops_before_issue_traversal():
+async def test_resource_cap_stops_after_first_page_without_traversing_catalog():
     adapter = CatalogAdapter(201)
     with pytest.raises(SeriesAdoptionError, match="limit"):
         await fetch(adapter, max_issues=200)
-    assert adapter.calls == [("series", "42")]
+    assert adapter.calls == [("series", "42"), ("issues", "42", 1)]
 
 
 async def test_total_deadline_bounds_traversal_and_closes_adapter():