From c3cafd52a643218f39c5788f3e4292bf9d38b0bc Mon Sep 17 00:00:00 2001
From: Groene AI <270696204+groeneai@users.noreply.github.com>
Date: Tue, 6 Oct 2026 06:14:25 +0000
Subject: [PATCH 1/5] GitHub source: read every notifications page instead of
 only the first

fetch() ran `gh api notifications?since=...` without --paginate, so it
read only the first page (50 threads, the per_page cap) and then set the
cursor to the newest updated_at on that page. The next `since` excluded
every thread on the unread pages, so they never reached source_messages.
Any sync gap with more than 50 participating notifications lost the rest.

GitHub orders /notifications newest-first by, and applies `since` to, the
time a thread last notified the user, which the API does not return.
updated_at also moves on events that do not notify (own pushes, CI,
labels, edits), so it can sit hours above that key. That rules out any
updated_at watermark inside a window: oldest-first batches with such a
cursor lost 226 of 497 threads on a live 48 h window. So:

- Read the whole window in one call: `--paginate --jq '.[]'`, parsed as
  JSON Lines. The output is identical on gh 2.23.0 and 2.93.0, so there
  is no version floor. Any failure still returns nothing and holds the
  cursor; partial output with rc != 0 is never parsed.
- De-duplicate by id, keeping the later copy. A thread notified while
  the listing runs moves the boundary item onto the next page, so it is
  listed twice, and the later read is the newer state.
- Take the cursor from the first listed item, not the max. The first
  item comes from the first response, so its updated_at is at most the
  first page's read time, below the key of any notification that lands
  while later pages are read. The max can come from a later page whose
  updated_at was bumped by a non-notifying event after such a
  notification, and pass it. The first item also carries the largest key
  in the window, so nothing already returned is listed again.

limit stays unused, as it has been since the source was added, and the
docs now say batch_size does not cap this source. A run after a long gap
enriches the whole backlog before persisting.

Tests model the hidden key with a per-thread `_k <= updated_at`. Both
fail on main; the backlog test also fails on the oldest-first batching
design. Live, read-only, 48 h window: main misses 445 of 495 threads,
this change 0.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/sources.md                        |   3 +-
 nerve/sources/github.py                |  23 ++--
 tests/test_github_ci_branch.py         |   2 +-
 tests/test_github_source_actors.py     |   6 +-
 tests/test_github_source_pagination.py | 157 +++++++++++++++++++++++++
 5 files changed, 177 insertions(+), 14 deletions(-)
 create mode 100644 tests/test_github_source_pagination.py

diff --git a/docs/sources.md b/docs/sources.md
index 8120cbb34..cec71829c 100644
--- a/docs/sources.md
+++ b/docs/sources.md
@@ -142,7 +142,8 @@ 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)
 - **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`)
diff --git a/nerve/sources/github.py b/nerve/sources/github.py
index c54806a20..c51f674a0 100644
--- a/nerve/sources/github.py
+++ b/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),
@@ -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,12 @@ 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 []
+            listing = [
+                json.loads(line)
+                for line in stdout.decode().splitlines()
+                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 +160,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 +287,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 its `updated_at` stays
+            # below any notification that arrives while later pages are read.
+            next_cursor = (listing[0].get("updated_at") or cursor) if listing else cursor
             return FetchResult(
                 records=records,
                 next_cursor=next_cursor,
diff --git a/tests/test_github_ci_branch.py b/tests/test_github_ci_branch.py
index 9f29a20d9..438a14db1 100644
--- a/tests/test_github_ci_branch.py
+++ b/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,
diff --git a/tests/test_github_source_actors.py b/tests/test_github_source_actors.py
index 6ea62df57..f8104a7a7 100644
--- a/tests/test_github_source_actors.py
+++ b/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,
diff --git a/tests/test_github_source_pagination.py b/tests/test_github_source_pagination.py
new file mode 100644
index 000000000..34a00fa10
--- /dev/null
+++ b/tests/test_github_source_pagination.py
@@ -0,0 +1,157 @@
+"""The GitHub notifications source must not skip notifications beyond one page.
+
+GitHub lists ``/notifications`` 50 per page, newest first by the time a thread
+last notified the user, and applies ``since`` to that same time. The API never
+returns it: ``updated_at`` also moves on events that do not notify (pushes,
+labels, edits), so it can be far above the listing key. The fake below models
+that key as a hidden ``_k <= updated_at`` per thread.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
+
+from nerve.sources.github import GitHubSource
+
+_PAGE_SIZE = 50
+_T0 = datetime(2026, 1, 2, 10, 0, 0, tzinfo=timezone.utc)
+
+
+def _ts(seconds: int) -> str:
+    return (_T0 + timedelta(seconds=seconds)).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _thread(tid: str, k: int, updated_at: int | None = None) -> dict:
+    return {
+        "id": tid,
+        "_k": _ts(k),
+        "reason": "mention",
+        "unread": True,
+        "updated_at": _ts(k if updated_at is None else updated_at),
+        "subject": {"title": f"Thread {tid}", "type": "Issue", "url": None},
+        "repository": {
+            "full_name": "owner/repo",
+            "html_url": "https://github.com/owner/repo",
+        },
+    }
+
+
+class _FakeProc:
+    def __init__(self, stdout: bytes):
+        self._stdout = stdout
+        self.returncode = 0
+
+    async def communicate(self):
+        return self._stdout, b""
+
+
+def _install_fake_gh(monkeypatch, state: dict[str, dict], between_pages=None) -> None:
+    """Emulate ``gh api notifications?since=...`` over ``state`` (id -> thread).
+
+    Each page is computed from the state at the time it is requested;
+    ``between_pages(state)`` runs once, after the first page of the first call.
+    ``--paginate`` follows every page, otherwise only the first is returned.
+    ``--jq '.[]'`` prints one object per line, ``--jq .`` one array per page.
+    """
+    hook = [between_pages]
+
+    def listed(since: str) -> list[dict]:
+        threads = [t for t in state.values() if t["_k"] >= since]
+        return sorted(threads, key=lambda t: (t["_k"], t["id"]), reverse=True)
+
+    async def fake_exec(*argv, **kwargs):
+        query = parse_qs(urlsplit(argv[2]).query)
+        since = query["since"][0]
+        per_page = min(int(query.get("per_page", [_PAGE_SIZE])[0]), _PAGE_SIZE)
+
+        pages = []
+        offset = 0
+        while True:
+            current = listed(since)
+            page = current[offset:offset + per_page]
+            if not page:
+                break
+            pages.append([{k: v for k, v in t.items() if k != "_k"} for t in page])
+            has_next = len(current) > offset + per_page
+            if hook[0] is not None:
+                hook[0](state)
+                hook[0] = None
+            if "--paginate" not in argv or not has_next:
+                break
+            offset += per_page
+
+        jq = argv[argv.index("--jq") + 1] if "--jq" in argv else None
+        if jq == ".[]":
+            lines = [json.dumps(t) for page in pages for t in page]
+        else:
+            lines = [json.dumps(page) for page in pages]
+        return _FakeProc("\n".join(lines).encode())
+
+    monkeypatch.setattr(
+        "nerve.sources.github.asyncio.create_subprocess_exec", fake_exec,
+    )
+
+
+def _source(monkeypatch) -> GitHubSource:
+    src = GitHubSource()
+
+    async def no_enrich(notif, sem):
+        return {}
+
+    monkeypatch.setattr(src, "_enrich_notification", no_enrich)
+    return src
+
+
+@pytest.mark.asyncio
+async def test_backlog_with_updated_at_ahead_of_listing_key_is_never_skipped(monkeypatch):
+    displaced = {5, 25, 45, 65, 85}
+    threads = [
+        _thread(f"t{i:03d}", 60 * i, 60 * 119 + i if i in displaced else None)
+        for i in range(120)
+    ]
+    _install_fake_gh(monkeypatch, {t["id"]: t for t in threads})
+    src = _source(monkeypatch)
+
+    cursor = _ts(-10)
+    calls = []
+    while len(calls) < 10:
+        result = await src.fetch(cursor, limit=30)
+        calls.append(result)
+        cursor = result.next_cursor
+        if not result.records:
+            break
+
+    returned = [r.id for call in calls for r in call.records]
+    assert set(returned) == {t["id"] for t in threads}
+    assert len(returned) == len(set(returned))
+    assert len(calls[0].records) == 120
+    assert calls[0].next_cursor == _ts(60 * 119)
+
+
+@pytest.mark.asyncio
+async def test_notification_landing_between_pages_is_listed_by_the_next_call(monkeypatch):
+    state = {t["id"]: t for t in (_thread(f"t{i:03d}", 60 * i) for i in range(60))}
+
+    def between_pages(state):
+        state["x"] = _thread("x", 60 * 59 + 30)
+        state["t003"]["updated_at"] = _ts(60 * 59 + 45)
+        state["t010"]["updated_at"] = _ts(60 * 59 + 40)
+
+    _install_fake_gh(monkeypatch, state, between_pages)
+    src = _source(monkeypatch)
+
+    first = await src.fetch(_ts(-10), limit=30)
+    ids = [r.id for r in first.records]
+    assert len(ids) == 60
+    assert len(set(ids)) == 60
+    assert "x" not in ids
+    by_id = {r.id: r for r in first.records}
+    assert by_id["t010"].timestamp == _ts(60 * 59 + 40)
+    assert first.next_cursor == _ts(60 * 59)
+
+    second = await src.fetch(first.next_cursor, limit=30)
+    assert [r.id for r in second.records] == ["x"]

From 0ba59c99d3764e7dd2ffe52db904110389f25246 Mon Sep 17 00:00:00 2001
From: Groene AI <270696204+groeneai@users.noreply.github.com>
Date: Tue, 6 Oct 2026 10:08:40 +0000
Subject: [PATCH 2/5] GitHub source: list the cursor's own second again and
 split JSON Lines on \n

The cursor is the updated_at of the first listed item, at second
precision. A notification that lands after the first page is served has
a listing key at or above that value, so it can fall in the cursor's own
second. `since = cursor + 1s` then excluded it for good. `since` is now
the cursor itself; the re-listed thread of that second (normally the
first item alone) is skipped by insert_source_messages when unchanged.

str.splitlines() also breaks on U+0085, U+2028, U+2029 and a few C0/C1
separators. gh prints non-ASCII characters raw, so a U+0085 in any string
of a notification split its line, json.loads failed, and every later run
failed on the same window with the cursor held: the source stopped. The
listing is now split on "\n" only.

The test fake prints UTF-8 raw (ensure_ascii=False), like gh. New tests
cover a notification landing in the cursor's own second and a raw U+0085
in a subject title; the backlog and between-pages tests now expect the
boundary thread to be listed again.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/sources.md                        |  2 +-
 nerve/sources/github.py                | 13 ++++---
 tests/test_github_source_pagination.py | 53 ++++++++++++++++++++++----
 3 files changed, 54 insertions(+), 14 deletions(-)

diff --git a/docs/sources.md b/docs/sources.md
index cec71829c..e93ec71f0 100644
--- a/docs/sources.md
+++ b/docs/sources.md
@@ -145,7 +145,7 @@ visible rather than failing the fetch.
 - **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)
diff --git a/nerve/sources/github.py b/nerve/sources/github.py
index c51f674a0..2b16d6891 100644
--- a/nerve/sources/github.py
+++ b/nerve/sources/github.py
@@ -114,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:
@@ -144,9 +144,10 @@ 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)
 
+            # 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().splitlines()
+                for line in stdout.decode().split("\n")
                 if line.strip()
             ]
             notifications = list({n.get("id", ""): n for n in listing}.values())
@@ -287,8 +288,8 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
                     },
                 ))
 
-            # The first item came from the first page, so its `updated_at` stays
-            # below any notification that arrives while later pages are read.
+            # 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,
diff --git a/tests/test_github_source_pagination.py b/tests/test_github_source_pagination.py
index 34a00fa10..aa66f5eca 100644
--- a/tests/test_github_source_pagination.py
+++ b/tests/test_github_source_pagination.py
@@ -86,9 +86,9 @@ async def fake_exec(*argv, **kwargs):
 
         jq = argv[argv.index("--jq") + 1] if "--jq" in argv else None
         if jq == ".[]":
-            lines = [json.dumps(t) for page in pages for t in page]
+            lines = [json.dumps(t, ensure_ascii=False) for page in pages for t in page]
         else:
-            lines = [json.dumps(page) for page in pages]
+            lines = [json.dumps(page, ensure_ascii=False) for page in pages]
         return _FakeProc("\n".join(lines).encode())
 
     monkeypatch.setattr(
@@ -118,18 +118,22 @@ async def test_backlog_with_updated_at_ahead_of_listing_key_is_never_skipped(mon
 
     cursor = _ts(-10)
     calls = []
+    seen: set[str] = set()
     while len(calls) < 10:
         result = await src.fetch(cursor, limit=30)
         calls.append(result)
         cursor = result.next_cursor
-        if not result.records:
+        new = {r.id for r in result.records} - seen
+        seen |= new
+        if not new:
             break
 
-    returned = [r.id for call in calls for r in call.records]
-    assert set(returned) == {t["id"] for t in threads}
-    assert len(returned) == len(set(returned))
+    assert seen == {t["id"] for t in threads}
     assert len(calls[0].records) == 120
+    assert len({r.id for r in calls[0].records}) == 120
     assert calls[0].next_cursor == _ts(60 * 119)
+    assert [r.id for r in calls[1].records] == ["t119"]
+    assert len(calls) == 2
 
 
 @pytest.mark.asyncio
@@ -154,4 +158,39 @@ def between_pages(state):
     assert first.next_cursor == _ts(60 * 59)
 
     second = await src.fetch(first.next_cursor, limit=30)
-    assert [r.id for r in second.records] == ["x"]
+    assert [r.id for r in second.records] == ["x", "t059"]
+
+
+@pytest.mark.asyncio
+async def test_notification_landing_in_the_cursor_second_is_listed_by_the_next_call(monkeypatch):
+    state = {t["id"]: t for t in (_thread(f"t{i:03d}", 60 * i) for i in range(60))}
+
+    def between_pages(state):
+        state["x"] = _thread("x", 60 * 59)
+
+    _install_fake_gh(monkeypatch, state, between_pages)
+    src = _source(monkeypatch)
+
+    first = await src.fetch(_ts(-10), limit=30)
+    ids = [r.id for r in first.records]
+    assert len(ids) == 60
+    assert "x" not in ids
+    assert first.next_cursor == _ts(60 * 59)
+
+    second = await src.fetch(first.next_cursor, limit=30)
+    second_ids = {r.id for r in second.records}
+    assert "x" in second_ids
+    assert second_ids <= {"x", "t059"}
+
+
+@pytest.mark.asyncio
+async def test_raw_unicode_line_separators_inside_strings_are_parsed(monkeypatch):
+    threads = [_thread("a", 0), _thread("b", 60)]
+    threads[1]["subject"]["title"] = "Fix\u0085parser edge"
+    _install_fake_gh(monkeypatch, {t["id"]: t for t in threads})
+    src = _source(monkeypatch)
+
+    result = await src.fetch(_ts(-10), limit=30)
+    assert len(result.records) == 2
+    by_id = {r.id: r for r in result.records}
+    assert "Fix\u0085parser edge" in by_id["b"].summary

From 6dcf46b99d2f8d354e51f791d4798736b803b58b Mon Sep 17 00:00:00 2001
From: Groene AI <270696204+groeneai@users.noreply.github.com>
Date: Tue, 6 Oct 2026 12:26:04 +0000
Subject: [PATCH 3/5] Sources: condense and count only the records storage
 stored

insert_source_messages skips a record identical to the stored one, but
it returned only a count and SourceRunner ignored it: the runner then
condensed every fetched record and reported len(records) as ingested.
The GitHub source lists the cursor's own second again on every run, so
an idle run logged "1 records ingested" and, with condense: true, sent
the same unchanged thread to the fast model on every run.

insert_source_messages now returns the ids it stored, and the runner
keeps only those records for condensation and for records_ingested.
A failed insert still propagates and holds the cursor, as before.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/sources.md                         |  2 +-
 nerve/db/sources.py                     | 13 ++++++++-----
 nerve/sources/runner.py                 |  8 +++++---
 tests/test_channel_observation.py       |  2 +-
 tests/test_source_resurface.py          |  4 ++--
 tests/test_source_runner_persistence.py | 21 +++++++++++++++++++++
 6 files changed, 38 insertions(+), 12 deletions(-)

diff --git a/docs/sources.md b/docs/sources.md
index e93ec71f0..d480869e7 100644
--- a/docs/sources.md
+++ b/docs/sources.md
@@ -39,7 +39,7 @@ CONSUMERS (agent tools):
    - **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
 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
+4. **Persist** — Records are saved to the `source_messages` table with a configurable TTL. A record identical to the stored one is skipped; only stored records are condensed and counted as ingested.
 5. **Advance** — Source cursor is saved to SQLite after successful persistence
 
 ### Consumption (Consumer Side)
diff --git a/nerve/db/sources.py b/nerve/db/sources.py
index 5db130f9a..247df5076 100644
--- a/nerve/db/sources.py
+++ b/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,
diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py
index d0041ac56..30a57c2fb 100644
--- a/nerve/sources/runner.py
+++ b/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,8 @@ async def _ingest_batch(self, cursor: str | None) -> _BatchResult:
                 records_dropped=dropped_count,
                 error=str(e),
             )
+        # Storage skips an unchanged record; condense and count only what it stored.
+        records = [r for r in records if r.id in stored_ids]
 
         # 3. LLM-based condensation for still-long records (configurable per source)
         if self.condense:
@@ -410,9 +412,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,
         )
 
diff --git a/tests/test_channel_observation.py b/tests/test_channel_observation.py
index f98ece6b9..12bfa5630 100644
--- a/tests/test_channel_observation.py
+++ b/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
 
 
 # ---------------------------------------------------------------------- #
diff --git a/tests/test_source_resurface.py b/tests/test_source_resurface.py
index 769154f3c..e5f08c7b6 100644
--- a/tests/test_source_resurface.py
+++ b/tests/test_source_resurface.py
@@ -71,7 +71,7 @@ async def test_update_to_max_rowid_row_resurfaces_to_consumer(db):
 
     # New comment lands on the same thread -> content changes -> re-insert.
     n = await db.insert_source_messages([_rec("hot", "comment-2")], source="github")
-    assert n == 1  # the changed row was (re-)inserted
+    assert n == ["hot"]  # the changed row was (re-)inserted
 
     rowid_after = await _rowid_of(db, "github", "hot")
     assert rowid_after > parked, (
@@ -112,7 +112,7 @@ async def test_unchanged_record_does_not_resurface(db):
     rowid_before = await _rowid_of(db, "github", "t")
 
     n = await db.insert_source_messages([_rec("t", "same")], source="github")
-    assert n == 0  # nothing changed -> skipped silently
+    assert n == []  # nothing changed -> skipped silently
 
     rowid_after = await _rowid_of(db, "github", "t")
     assert rowid_after == rowid_before  # no churn, no spurious re-surface
diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py
index 6f523b041..6f5bd29d1 100644
--- a/tests/test_source_runner_persistence.py
+++ b/tests/test_source_runner_persistence.py
@@ -151,3 +151,24 @@ async def fail_once(records, source, ttl_days):
     assert await db.get_sync_cursor(SOURCE_NAME) == "new"
     rows, _ = await db.list_source_messages(source=SOURCE_NAME, limit=10)
     assert {row["id"] for row in rows} == {"a", "b"}
+
+
+@pytest.mark.asyncio
+async def test_runner_condenses_and_counts_only_records_storage_stored(db, monkeypatch):
+    runner = SourceRunner(_FixedSource([_record("a")]), db, condense=True)
+    condensed: list[list[str]] = []
+
+    async def fake_condense(records):
+        condensed.append([r.id for r in records])
+        return records
+
+    monkeypatch.setattr(runner, "_condense_long_content", fake_condense)
+
+    first = await runner.run()
+    second = await runner.run()
+
+    assert first.records_ingested == 1
+    assert second.records_ingested == 0
+    assert condensed[0] == ["a"]
+    assert all(ids == [] for ids in condensed[1:])
+    assert await db.get_sync_cursor(SOURCE_NAME) == "new"

From a214ebf8a3ba6eecb0f85651805aef678c8f6bc5 Mon Sep 17 00:00:00 2001
From: Groene AI <270696204+groeneai@users.noreply.github.com>
Date: Tue, 6 Oct 2026 14:05:01 +0000
Subject: [PATCH 4/5] docs: sources pipeline condenses after persist; GitHub
 cursor is inclusive

The ingestion list and the overview placed LLM condensation before
persistence, but the runner persists first and condenses only the
records storage stored. List condensation as its own step after
Persist, and drop "before storing" from the condense config field.

The Cursor Design note still said every inclusive source advances the
cursor by one second. That is Gmail only: the GitHub source queries the
cursor itself, so a notification that lands in the cursor's own second
is not skipped, and storage skips the unchanged thread listed again.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/sources.md | 15 +++++++--------
 1 file changed, 7 insertions(+), 8 deletions(-)

diff --git a/docs/sources.md b/docs/sources.md
index d480869e7..969e797b5 100644
--- a/docs/sources.md
+++ b/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. A record identical to the stored one is skipped; only stored records are condensed and counted as ingested.
-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)
 
@@ -372,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 |
 
@@ -787,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.
 

From a779b0fe1ca81c57b4fce509293b08923680331d Mon Sep 17 00:00:00 2001
From: Groene AI <270696204+groeneai@users.noreply.github.com>
Date: Tue, 20 Oct 2026 16:48:31 +0000
Subject: [PATCH 5/5] Sources: count a record id repeated within one batch once

ChannelSource (#422) maps a message observed twice to the same record id,
so one batch can carry two copies of an id. Storage stores one row for
them (it skips an identical copy and replaces the row for a changed one),
but the runner kept every input record whose id storage returned, so it
condensed both copies and counted 2 ingested.

Keep one record per stored id, the last copy, which is the one storage
holds. A parametrized runner test covers an identical and a changed repeat.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 nerve/sources/runner.py                 |  5 +++--
 tests/test_source_runner_persistence.py | 20 ++++++++++++++++++++
 2 files changed, 23 insertions(+), 2 deletions(-)

diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py
index 30a57c2fb..4ff23b1e9 100644
--- a/nerve/sources/runner.py
+++ b/nerve/sources/runner.py
@@ -378,8 +378,9 @@ async def _ingest_batch(self, cursor: str | None) -> _BatchResult:
                 records_dropped=dropped_count,
                 error=str(e),
             )
-        # Storage skips an unchanged record; condense and count only what it stored.
-        records = [r for r in records if r.id in stored_ids]
+        # 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:
diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py
index 6f5bd29d1..18852a765 100644
--- a/tests/test_source_runner_persistence.py
+++ b/tests/test_source_runner_persistence.py
@@ -1,3 +1,4 @@
+import dataclasses
 import sqlite3
 
 import pytest
@@ -172,3 +173,22 @@ async def fake_condense(records):
     assert condensed[0] == ["a"]
     assert all(ids == [] for ids in condensed[1:])
     assert await db.get_sync_cursor(SOURCE_NAME) == "new"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("repeat_content", ["Content a", "Content a, edited"])
+async def test_runner_condenses_and_counts_a_repeated_id_once(db, monkeypatch, repeat_content):
+    repeat = dataclasses.replace(_record("a"), content=repeat_content)
+    runner = SourceRunner(_FixedSource([_record("a"), repeat]), db, condense=True)
+    condensed: list[list[str]] = []
+
+    async def fake_condense(records):
+        condensed.append([r.content for r in records])
+        return records
+
+    monkeypatch.setattr(runner, "_condense_long_content", fake_condense)
+
+    result = await runner.run()
+
+    assert result.records_ingested == 1
+    assert condensed == [[repeat_content]]