GitHub source: read every notifications page instead of only the first

docs/sources.md

@@ -7,7 +7,7 @@ Sources are cursor-based data streams that persist records to a local inbox
 drain records that Slack or Telegram already delivered to a local buffer. The
 architecture follows a **producer/consumer** pattern inspired by Kafka:
 
-- **Producers** (source runners) — Fetch or drain records, preprocess and condense them, persist to the inbox, and advance the source cursor. No agent processing happens here.
+- **Producers** (source runners) — Fetch or drain records, preprocess them, persist them to the inbox, condense long ones, and advance the source cursor. No agent processing happens here.
 - **Consumers** (agent tools) — Read from the inbox using independent persistent cursors. Multiple consumers can read the same messages without interfering with each other.
 
 ```
@@ -35,12 +35,11 @@ CONSUMERS (agent tools):
 ### Ingestion (Producer Side)
 
 1. **Fetch** — The source adapter reads an external service (gh CLI, gog CLI, Telethon) or a local channel buffer and returns normalized `SourceRecord` objects with an opaque cursor
-2. **Preprocess** — Two-stage content cleanup:
-   - **Source-specific** (`source.preprocess()`) — Each source can override this for programmatic cleanup. Gmail strips boilerplate paragraphs (legal disclaimers, unsubscribe blocks, tracking URLs). Default: no-op
-   - **LLM condensation** (`condense: true`) — Records still over 800 chars are sent to a fast model (Haiku) that extracts only essential information. Configurable per source, runs concurrently with a 30s timeout per record, falls back to original content on failure
+2. **Preprocess** (`source.preprocess()`) — Each source can override this for programmatic cleanup. Gmail strips boilerplate paragraphs (legal disclaimers, unsubscribe blocks, tracking URLs). Default: no-op
 3. **Guardrail** — An optional `InboxFilter` drops records that fail the source's allow/deny rules (see [Guardrails](#guardrails-inbox-filtering)). Dropped records are never persisted
-4. **Persist** — Records are saved to the `source_messages` table with a configurable TTL
-5. **Advance** — Source cursor is saved to SQLite after successful persistence
+4. **Persist** — Records are saved to the `source_messages` table with a configurable TTL. A record identical to the stored one is skipped: it is not condensed and not counted as ingested.
+5. **LLM condensation** (`condense: true`) — Stored records over 800 chars are sent to a fast model (Haiku) that extracts only essential information, saved as `processed_content`. Configurable per source, runs concurrently with a 30s timeout per record, falls back to original content on failure
+6. **Advance** — Source cursor is saved to SQLite after successful persistence
 
 ### Consumption (Consumer Side)
 
@@ -142,9 +141,10 @@ visible rather than failing the fetch.
 
 ### GitHub
 - **Adapter:** `nerve/sources/github.py` — uses `gh api notifications` CLI
-- **Cursor:** ISO 8601 timestamp of the newest notification's `updated_at`
+- **Cursor:** ISO 8601 `updated_at` of the first notification in the last listing (GitHub lists the most recently notified thread first)
+- **Paging:** every page since the cursor is read in one run (`gh api --paginate`), so a backlog after downtime is ingested in full; `batch_size` does not cap this source
 - **First run:** Fetches from the last 24 hours
-- **Subsequent runs:** Uses `since=<cursor + 1s>` with `Z` suffix (not `+00:00` — the `+` in a URL query string is decoded as a space, breaking the filter)
+- **Subsequent runs:** Uses `since=<cursor>` (inclusive: the cursor's own second is listed again, and an unchanged re-listed thread is skipped by storage) with `Z` suffix (not `+00:00` — the `+` in a URL query string is decoded as a space, breaking the filter)
 - **Filter:** `participating=true` (assigned, review requested, mentioned). Note it never returns `ci_activity` — workflow-run notifications are not "participating", so CI never reaches the inbox through this source at all (verified against a live account: 25 `ci_activity` with `all=true`, 0 with `participating=true`)
 - **Enrichment:** Each notification is enriched with actual content from the subject (PR/issue body, state, assignees, labels) and the latest comment, fetched in parallel (up to 5 concurrent `gh api` calls)
 - **Metadata:** `reason`, `repo_name`, `actors` (every login involved) and `ci_branch` (branch of a CheckSuite run, `""` otherwise) — all four are guardrail-filterable, see [Guardrails](#guardrails-inbox-filtering)
@@ -371,7 +371,7 @@ sync:
 | `enabled` | bool | `true` | Enable/disable this source |
 | `schedule` | string | varies | Crontab expression or interval (`2h`, `30m`) |
 | `batch_size` | int | `50` | Max records per fetch cycle |
-| `condense` | bool | `false` | LLM-condense long records via `memory.fast_model` before storing |
+| `condense` | bool | `false` | LLM-condense long stored records via `memory.fast_model` |
 | `message_ttl_days` | int | `7` | How long to keep inbox messages |
 | `consumer_cursor_ttl_days` | int | `2` | Consumer cursors expire after N days of inactivity |
 
@@ -786,7 +786,7 @@ export function MyRenderer({ content, metadata, summary }: Props) {
 
 **Consumer cursors**: Stored in `consumer_cursors` table with composite key `(consumer, source)`. Each consumer has an independent cursor per source, using the implicit `rowid` of `source_messages` as the offset. New cursors initialize to `MAX(rowid)` (no backlog). Cursors expire after `consumer_cursor_ttl_days` of inactivity.
 
-**Inclusive APIs:** Both GitHub and Gmail APIs use inclusive cursor semantics (returning records `>= cursor`). Sources handle this by advancing the cursor at query time (+1s) and applying client-side dedup as a safety net.
+**Inclusive APIs:** Both GitHub and Gmail APIs use inclusive cursor semantics (returning records `>= cursor`). Gmail advances the cursor at query time (+1s) and applies client-side dedup as a safety net. GitHub queries the cursor itself, because a notification can land in the cursor's own second; storage skips the unchanged thread listed again.
 
 **Concurrency safety:** Each `SourceRunner` holds an `asyncio.Lock` to prevent concurrent execution of the same source. If a cron schedule fires while a manual sync is in progress, the second call returns immediately with 0 records.
 

nerve/db/sources.py

@@ -108,12 +108,15 @@ async def insert_source_messages(
         records: list,
         source: str,
         ttl_days: int = 7,
-    ) -> int:
-        """Bulk insert source records into the inbox. Returns count inserted.
+    ) -> list[str]:
+        """Bulk insert source records into the inbox. Returns the ids it stored.
 
         The batch is atomic; if any record cannot be persisted, no records are
         committed and the exception is propagated.
 
+        A record identical to the stored one (same metadata and content) is
+        skipped and not returned.
+
         If a record with the same (source, id) already exists but has different
         metadata or content, the old record is deleted and re-inserted at a
         strictly-higher rowid. This ensures mutable sources (e.g. GitHub
@@ -124,7 +127,7 @@ async def insert_source_messages(
         now = datetime.now(timezone.utc)
         expires = (now + timedelta(days=ttl_days)).isoformat()
         now_iso = now.isoformat()
-        inserted = 0
+        stored: list[str] = []
         async with self._atomic():
             for r in records:
                 try:
@@ -192,14 +195,14 @@ async def insert_source_messages(
                              r.timestamp, new_metadata,
                              now_iso, expires),
                         )
-                    inserted += 1
+                    stored.append(r.id)
                 except Exception as e:
                     logger.warning(
                         "Failed to persist source message %s/%s: %s",
                         source, r.id, e,
                     )
                     raise
-        return inserted
+        return stored
 
     async def update_source_messages_processed(
         self,

nerve/sources/github.py

@@ -1,6 +1,7 @@
 """GitHub source — fetches notifications via the gh CLI.
 
-Cursor semantics: ISO 8601 timestamp of the newest notification's `updated_at`.
+Cursor semantics: ISO 8601 `updated_at` of the first notification in the last
+listing (GitHub lists the most recently notified thread first).
 On first run (no cursor), fetches from the last 24 hours.
 
 Each notification is enriched with actual content from the subject (PR/issue),
@@ -113,14 +114,14 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
 
         On first run (cursor=None): fetches from the last 24 hours.
         """
-        # GitHub's `since` is inclusive (>=), so advance by 1s to skip
-        # already-seen notifications.
+        # `since` is not advanced past the cursor: a notification can land in the
+        # cursor's own second. A re-listed unchanged thread is skipped by storage.
         # IMPORTANT: use Z suffix, not +00:00 — the `+` in a URL query
         # string is interpreted as a space, silently breaking the filter.
         if cursor:
             try:
                 cursor_dt = datetime.fromisoformat(cursor.replace("Z", "+00:00"))
-                since = (cursor_dt + timedelta(seconds=1)).strftime("%Y-%m-%dT%H:%M:%SZ")
+                since = cursor_dt.strftime("%Y-%m-%dT%H:%M:%SZ")
             except ValueError:
                 since = cursor
         else:
@@ -129,9 +130,11 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
         try:
             # Query params go in the URL — -f flags would force POST
             endpoint = f"notifications?since={since}&participating=true"
+            # `since` filters on when a thread last notified, not on the
+            # returned `updated_at`, so read every page of the window.
             proc = await asyncio.create_subprocess_exec(
                 "gh", "api", endpoint,
-                "--jq", ".",
+                "--paginate", "--jq", ".[]",
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
             )
@@ -141,8 +144,13 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
                 logger.error("gh api notifications failed: %s", stderr.decode())
                 return FetchResult(records=[], next_cursor=cursor)
 
-            stdout_text = stdout.decode()
-            notifications = json.loads(stdout_text) if stdout_text.strip() else []
+            # gh prints one JSON value per line; `splitlines()` also splits on U+0085 in a string.
+            listing = [
+                json.loads(line)
+                for line in stdout.decode().split("\n")
+                if line.strip()
+            ]
+            notifications = list({n.get("id", ""): n for n in listing}.values())
 
             # Enrich notifications with actual content in parallel
             sem = asyncio.Semaphore(_MAX_CONCURRENT_FETCHES)
@@ -153,7 +161,6 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
             enriched = await asyncio.gather(*enrich_tasks, return_exceptions=True)
 
             records: list[SourceRecord] = []
-            newest_ts: str | None = None
 
             for notif, extra in zip(notifications, enriched):
                 subject = notif.get("subject", {})
@@ -281,10 +288,9 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
                     },
                 ))
 
-                if newest_ts is None or updated_at > (newest_ts or ""):
-                    newest_ts = updated_at
-
-            next_cursor = newest_ts if newest_ts else cursor
+            # The first item came from the first page, so no notification that arrives
+            # while later pages are read is older than its `updated_at`.
+            next_cursor = (listing[0].get("updated_at") or cursor) if listing else cursor
             return FetchResult(
                 records=records,
                 next_cursor=next_cursor,

nerve/sources/runner.py

@@ -367,7 +367,7 @@ async def _ingest_batch(self, cursor: str | None) -> _BatchResult:
 
         # 2. Persist to inbox (post-preprocess, pre-condense — human-readable)
         try:
-            await self._persist_to_inbox(records)
+            stored_ids = set(await self._persist_to_inbox(records))
         except Exception as e:
             logger.error(
                 "Source %s persistence failed: %s",
@@ -378,6 +378,9 @@ async def _ingest_batch(self, cursor: str | None) -> _BatchResult:
                 records_dropped=dropped_count,
                 error=str(e),
             )
+        # Storage skips an unchanged record and keeps the last copy of a repeated id;
+        # condense and count only what it stored, once per id.
+        records = list({r.id: r for r in records if r.id in stored_ids}.values())
 
         # 3. LLM-based condensation for still-long records (configurable per source)
         if self.condense:
@@ -410,9 +413,9 @@ async def _ingest_batch(self, cursor: str | None) -> _BatchResult:
     # Inbox persistence
     # ------------------------------------------------------------------
 
-    async def _persist_to_inbox(self, records: list[SourceRecord]) -> None:
+    async def _persist_to_inbox(self, records: list[SourceRecord]) -> list[str]:
         """Save records to the source_messages table for inbox display."""
-        await self.db.insert_source_messages(
+        return await self.db.insert_source_messages(
             records, source=self.source.source_name, ttl_days=self.ttl_days,
         )
 

tests/test_channel_observation.py

@@ -1117,7 +1117,7 @@ async def test_the_same_message_observed_twice_lands_once(self, db):
         inserted = await db.insert_source_messages(result.records, source="slack")
 
         assert len(result.records) == 2
-        assert inserted == 1
+        assert len(inserted) == 1
 
 
 # ---------------------------------------------------------------------- #

tests/test_github_ci_branch.py

@@ -121,7 +121,7 @@ async def test_fetch_populates_ci_branch_metadata(monkeypatch):
     ]
 
     async def fake_exec(*args, **kwargs):
-        return _FakeProc(json.dumps(notifications).encode())
+        return _FakeProc("\n".join(json.dumps(n) for n in notifications).encode())
 
     monkeypatch.setattr(
         "nerve.sources.github.asyncio.create_subprocess_exec", fake_exec,

tests/test_github_source_actors.py

@@ -148,7 +148,7 @@ async def test_fetch_populates_actors_metadata(monkeypatch):
     }]
 
     async def fake_exec(*args, **kwargs):
-        return _FakeProc(json.dumps(notifications).encode())
+        return _FakeProc("\n".join(json.dumps(n) for n in notifications).encode())
 
     monkeypatch.setattr(
         "nerve.sources.github.asyncio.create_subprocess_exec", fake_exec,
@@ -438,7 +438,7 @@ async def test_maintainer_assignment_on_self_filed_issue_passes_guardrail(
     }]
 
     async def fake_exec(*args, **kwargs):
-        return _FakeProc(json.dumps(notifications).encode())
+        return _FakeProc("\n".join(json.dumps(n) for n in notifications).encode())
 
     monkeypatch.setattr(
         "nerve.sources.github.asyncio.create_subprocess_exec", fake_exec,
@@ -502,7 +502,7 @@ async def test_fetch_actors_empty_when_enrichment_fails(monkeypatch):
     }]
 
     async def fake_exec(*args, **kwargs):
-        return _FakeProc(json.dumps(notifications).encode())
+        return _FakeProc("\n".join(json.dumps(n) for n in notifications).encode())
 
     monkeypatch.setattr(
         "nerve.sources.github.asyncio.create_subprocess_exec", fake_exec,