diff --git a/docs/sources.md b/docs/sources.md index 8120cbb34..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 -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=` with `Z` suffix (not `+00:00` — the `+` in a URL query string is decoded as a space, breaking the filter) +- **Subsequent runs:** Uses `since=` (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. 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/github.py b/nerve/sources/github.py index c54806a20..2b16d6891 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), @@ -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, diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py index d0041ac56..4ff23b1e9 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,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, ) 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_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..aa66f5eca --- /dev/null +++ b/tests/test_github_source_pagination.py @@ -0,0 +1,196 @@ +"""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, ensure_ascii=False) for page in pages for t in page] + else: + lines = [json.dumps(page, ensure_ascii=False) 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 = [] + seen: set[str] = set() + while len(calls) < 10: + result = await src.fetch(cursor, limit=30) + calls.append(result) + cursor = result.next_cursor + new = {r.id for r in result.records} - seen + seen |= new + if not new: + break + + 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 +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", "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 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..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 @@ -151,3 +152,43 @@ 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" + + +@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]]