github_repos: preserve cursor after partial fetch failures

nerve/sources/github_repos.py

@@ -71,9 +71,15 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
 
         # Flatten into (repo, issue) pairs, skipping repos that errored.
         items: list[tuple[str, dict]] = []
+        fetch_failed = False
         for repo, res in zip(self._repos, results):
             if isinstance(res, Exception):
                 logger.warning("github_repos: fetch failed for %s: %s", repo, res)
+                fetch_failed = True
+                continue
+            if res is None:
+                logger.warning("github_repos: fetch failed for %s", repo)
+                fetch_failed = True
                 continue
             for issue in res:
                 if isinstance(issue, dict):
@@ -84,6 +90,11 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
 
         # First run: establish baseline cursor, don't backfill history.
         if cursor is None:
+            if fetch_failed:
+                logger.info(
+                    "github_repos: first run incomplete, baseline not established",
+                )
+                return FetchResult(records=[], next_cursor=None)
             newest_ts = max(
                 (it.get("created_at", "") for _, it in items),
                 default=None,
@@ -108,21 +119,25 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
         new_items.sort(key=lambda ri: ri[1].get("created_at", ""))
         records = [self._issue_to_record(repo, it) for repo, it in new_items]
 
-        # Advance cursor to the newest created_at we just ingested.
+        # Successful repositories can be ingested immediately, but a shared
+        # cursor is safe to advance only after every repository was observed.
+        # Unchanged retries from healthy repositories are deduplicated by
+        # (source, id).
         newest_ts = max(it.get("created_at", "") for _, it in new_items)
-        return FetchResult(records=records, next_cursor=newest_ts, has_more=False)
+        next_cursor = cursor if fetch_failed else newest_ts
+        return FetchResult(records=records, next_cursor=next_cursor, has_more=False)
 
     # ------------------------------------------------------------------
     # Fetch + formatting helpers
     # ------------------------------------------------------------------
 
     async def _fetch_repo(
         self, repo: str, per_page: int, sem: asyncio.Semaphore,
-    ) -> list[dict]:
+    ) -> list[dict] | None:
         """Fetch the most-recently-created issues+PRs for a single repo.
 
         Returns a list of issue dicts (PRs included — they carry a
-        ``pull_request`` key). Returns [] on any error.
+        ``pull_request`` key), or ``None`` on error.
         """
         endpoint = (
             f"repos/{repo}/issues"
@@ -131,7 +146,7 @@ async def _fetch_repo(
         async with sem:
             data = await self._gh_api_get(endpoint)
         if not isinstance(data, list):
-            return []
+            return None
         return data
 
     def _issue_to_record(self, repo: str, issue: dict) -> SourceRecord:

tests/test_github_repos_source.py

@@ -165,6 +165,75 @@ async def flaky_gh_api_get(endpoint, timeout=30):
     repos_seen = {r.metadata["repo_name"] for r in result.records}
     assert repos_seen == {"owner/repo-a"}
     assert len(result.records) == 2
+    assert result.next_cursor == "2026-06-12T08:00:00Z"
+
+
+@pytest.mark.asyncio
+async def test_partial_failure_keeps_cursor_until_missed_item_can_be_fetched():
+    src = GitHubReposSource(config={"repos": ["owner/repo-a", "owner/repo-b"]})
+    repo_b_available = False
+    data = {
+        "owner/repo-a": [
+            _issue(1, 101, "2026-06-12T11:00:00Z", "Issue A"),
+        ],
+        "owner/repo-b": [
+            _issue(2, 102, "2026-06-12T10:30:00Z", "Issue B"),
+        ],
+    }
+
+    async def flaky_gh_api_get(endpoint, timeout=30):
+        match = re.match(r"repos/([^/]+/[^/]+)/issues", endpoint)
+        repo = match.group(1)
+        if repo == "owner/repo-b" and not repo_b_available:
+            return None
+        return data[repo]
+
+    src._gh_api_get = flaky_gh_api_get
+
+    first = await src.fetch(cursor="2026-06-12T10:00:00Z")
+
+    assert [record.id for record in first.records] == ["101"]
+    assert first.next_cursor == "2026-06-12T10:00:00Z"
+
+    repo_b_available = True
+    second = await src.fetch(cursor=first.next_cursor)
+
+    assert [record.id for record in second.records] == ["102", "101"]
+    assert second.next_cursor == "2026-06-12T11:00:00Z"
+
+
+@pytest.mark.asyncio
+async def test_first_run_waits_for_every_repository_before_setting_baseline():
+    src = GitHubReposSource(config={"repos": ["owner/repo-a", "owner/repo-b"]})
+
+    async def flaky_gh_api_get(endpoint, timeout=30):
+        if "repo-b" in endpoint:
+            return None
+        return _REPO_DATA["owner/repo-a"]
+
+    src._gh_api_get = flaky_gh_api_get
+    result = await src.fetch(cursor=None)
+
+    assert result.records == []
+    assert result.next_cursor is None
+
+
+@pytest.mark.asyncio
+async def test_empty_repository_response_does_not_prevent_cursor_advance():
+    src = _make_source(
+        ["owner/repo-a", "owner/repo-b"],
+        data={
+            "owner/repo-a": [
+                _issue(1, 101, "2026-06-12T11:00:00Z", "Issue A"),
+            ],
+            "owner/repo-b": [],
+        },
+    )
+
+    result = await src.fetch(cursor="2026-06-12T10:00:00Z")
+
+    assert [record.id for record in result.records] == ["101"]
+    assert result.next_cursor == "2026-06-12T11:00:00Z"
 
 
 @pytest.mark.asyncio