feat: prioritize recent-window backfill before full history

11 files changed, 391 insertions, 64 deletions

diff --git a/API.md b/API.md
index 5b8f0e9..3362270 100644
--- a/API.md
+++ b/API.md
@@ -45,6 +45,7 @@ Contribution ranges:
 - Contribution read endpoints return zero-filled days, so clients do not need to patch missing dates.
 - Public contribution reads also register the actor for background indexing. A first lookup may therefore return an empty graph while the scheduler catches up.
 - Incremental indexing and historical backfill are separate. An actor can be recently indexed without being fully backfilled yet.
+- The service prioritizes a recent visible history window first, then continues deep-history backfill afterward.
 
 Background polling:
 
@@ -112,6 +113,7 @@ Behavior notes:
 - This endpoint also registers the actor for background indexing and updates the actor's `last_requested_at` timestamp.
 - The response is always immediate; it does not wait for SourceHut polling to finish.
 - Historical backfill runs in bounded background batches and may take multiple scheduler passes to complete.
+- The recent visible window is prioritized before full-history backfill so clients can show a useful graph sooner.
 
 Example by year:
 
@@ -135,6 +137,9 @@ Response `200 OK`:
   "is_indexed": false,
   "last_polled_at": null,
   "indexing_state": "pending",
+  "is_recent_window_backfilled": false,
+  "recent_backfill_state": "in_progress",
+  "recent_backfill_completed_at": null,
   "is_backfilled": false,
   "backfill_state": "in_progress",
   "backfill_completed_at": null,
@@ -153,6 +158,9 @@ Response fields:
 - `is_indexed` boolean: whether the service has already completed at least one successful recent/incremental poll for this actor
 - `last_polled_at` string or `null`: most recent successful poll time, if any
 - `indexing_state` string: one of `pending`, `indexed`, or `error`
+- `is_recent_window_backfilled` boolean: whether the prioritized recent history window has completed backfill
+- `recent_backfill_state` string: one of `pending`, `in_progress`, `completed`, or `error`
+- `recent_backfill_completed_at` string or `null`: when recent-window backfill completed, if it has
 - `is_backfilled` boolean: whether historical backfill has completed for this actor
 - `backfill_state` string: one of `pending`, `in_progress`, `completed`, or `error`
 - `backfill_completed_at` string or `null`: when full historical backfill completed, if it has
@@ -169,6 +177,8 @@ Indexing state semantics:
 
 Backfill state semantics:
 
+- recent-window fields:
+  - represent the prioritized visible-history window for client UX
 - `pending`: the actor has not started historical backfill yet
 - `in_progress`: historical backfill is actively progressing in bounded background batches
 - `completed`: historical backfill has completed for all supported services
@@ -226,6 +236,9 @@ Response `200 OK`:
   "is_indexed": true,
   "last_polled_at": "2026-04-11T18:05:00Z",
   "indexing_state": "indexed",
+  "is_recent_window_backfilled": true,
+  "recent_backfill_state": "completed",
+  "recent_backfill_completed_at": "2026-04-11T18:02:00Z",
   "is_backfilled": false,
   "backfill_state": "in_progress",
   "backfill_completed_at": null,
@@ -245,6 +258,9 @@ Response fields:
 - `is_indexed` boolean
 - `last_polled_at` string or `null`
 - `indexing_state` string
+- `is_recent_window_backfilled` boolean
+- `recent_backfill_state` string
+- `recent_backfill_completed_at` string or `null`
 - `is_backfilled` boolean
 - `backfill_state` string
 - `backfill_completed_at` string or `null`
diff --git a/README.md b/README.md
index b5939f6..d422338 100644
--- a/README.md
+++ b/README.md
@@ -14,7 +14,7 @@ The current V1 is intentionally narrow and production-oriented:
 
 ## What It Does
 
-The service collects SourceHut activity from one or more sr.ht GraphQL services, turns those records into a canonical event shape, aggregates activity by day, and returns zero-filled calendar ranges so the client never has to patch missing dates. It performs both recent incremental polling and bounded historical backfill.
+The service collects SourceHut activity from one or more sr.ht GraphQL services, turns those records into a canonical event shape, aggregates activity by day, and returns zero-filled calendar ranges so the client never has to patch missing dates. It performs recent incremental polling, prioritizes a recent visible-history window for faster UX, and then continues bounded historical backfill.
 
 Example use cases:
 
@@ -208,6 +208,9 @@ Example response:
   "is_indexed": true,
   "last_polled_at": "2026-04-11T18:05:00Z",
   "indexing_state": "indexed",
+  "is_recent_window_backfilled": true,
+  "recent_backfill_state": "completed",
+  "recent_backfill_completed_at": "2026-04-11T18:02:00Z",
   "is_backfilled": false,
   "backfill_state": "in_progress",
   "backfill_completed_at": null,
@@ -235,6 +238,9 @@ Example response:
   "is_indexed": true,
   "last_polled_at": "2026-04-11T18:05:00Z",
   "indexing_state": "indexed",
+  "is_recent_window_backfilled": true,
+  "recent_backfill_state": "completed",
+  "recent_backfill_completed_at": "2026-04-11T18:02:00Z",
   "is_backfilled": false,
   "backfill_state": "in_progress",
   "backfill_completed_at": null,
@@ -326,6 +332,7 @@ The SourceHut-specific assumptions are isolated to the service modules:
 - `git.sr.ht` polling assumes the actor's repositories are discoverable through the SourceHut GraphQL API
 - scheduled polling runs in-process, so it is not a distributed scheduler
 - newly requested actors are indexed asynchronously, so the first public read may be empty until a scheduler or manual poll runs
+- the recent visible-history window is prioritized first, but deep-history backfill can still take many scheduler passes for active users
 - full historical backfill can take many scheduler passes for active users because it runs in bounded batches
 - alias management is config-driven; there is no alias CRUD API yet
 - current deployment model is trusted-operator V1, not a public multi-tenant service
diff --git a/alembic/versions/20260411_0004_recent_backfill_scope.py b/alembic/versions/20260411_0004_recent_backfill_scope.py
new file mode 100644
index 0000000..8d29987
--- /dev/null
+++ b/alembic/versions/20260411_0004_recent_backfill_scope.py
@@ -0,0 +1,89 @@
+"""recent backfill scope and actor fields"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy import inspect
+
+
+revision = "20260411_0004"
+down_revision = "20260411_0003"
+branch_labels = None
+depends_on = None
+
+
+def _table_names() -> set[str]:
+    return set(inspect(op.get_bind()).get_table_names())
+
+
+def _column_names(table_name: str) -> set[str]:
+    return {column["name"] for column in inspect(op.get_bind()).get_columns(table_name)}
+
+
+def _service_backfill_needs_upgrade() -> bool:
+    if "service_backfill_states" not in _table_names():
+        return False
+    columns = _column_names("service_backfill_states")
+    if "scope" not in columns:
+        return True
+    unique_constraints = {
+        constraint["name"]
+        for constraint in inspect(op.get_bind()).get_unique_constraints("service_backfill_states")
+    }
+    return "uq_service_backfill_state_actor_service_scope" not in unique_constraints
+
+
+def _upgrade_service_backfill_states() -> None:
+    op.execute(
+        sa.text(
+            """
+            CREATE TABLE service_backfill_states__alembic_new (
+                id INTEGER NOT NULL PRIMARY KEY,
+                actor VARCHAR(255) NOT NULL,
+                service VARCHAR(32) NOT NULL,
+                scope VARCHAR(16) NOT NULL,
+                cursor_json JSON,
+                status VARCHAR(32) NOT NULL,
+                started_at DATETIME,
+                completed_at DATETIME,
+                last_error TEXT,
+                updated_at DATETIME NOT NULL,
+                CONSTRAINT uq_service_backfill_state_actor_service_scope UNIQUE (actor, service, scope)
+            )
+            """
+        )
+    )
+    op.execute(
+        sa.text(
+            """
+            INSERT INTO service_backfill_states__alembic_new
+                (id, actor, service, scope, cursor_json, status, started_at, completed_at, last_error, updated_at)
+            SELECT id, actor, service, 'full', cursor_json, status, started_at, completed_at, last_error, updated_at
+            FROM service_backfill_states
+            """
+        )
+    )
+    op.execute(sa.text("DROP TABLE service_backfill_states"))
+    op.execute(sa.text("ALTER TABLE service_backfill_states__alembic_new RENAME TO service_backfill_states"))
+
+
+def upgrade() -> None:
+    if "tracked_actors" in _table_names():
+        columns = _column_names("tracked_actors")
+        if "recent_backfill_status" not in columns:
+            op.add_column("tracked_actors", sa.Column("recent_backfill_status", sa.String(length=32), nullable=True))
+            op.execute(sa.text("UPDATE tracked_actors SET recent_backfill_status = 'pending' WHERE recent_backfill_status IS NULL"))
+        if "recent_backfill_started_at" not in columns:
+            op.add_column("tracked_actors", sa.Column("recent_backfill_started_at", sa.DateTime(timezone=True), nullable=True))
+        if "recent_backfill_completed_at" not in columns:
+            op.add_column("tracked_actors", sa.Column("recent_backfill_completed_at", sa.DateTime(timezone=True), nullable=True))
+        if "last_recent_backfill_error" not in columns:
+            op.add_column("tracked_actors", sa.Column("last_recent_backfill_error", sa.Text(), nullable=True))
+
+    if _service_backfill_needs_upgrade():
+        _upgrade_service_backfill_states()
+
+
+def downgrade() -> None:
+    pass
diff --git a/src/srht_contrib/jobs/poller.py b/src/srht_contrib/jobs/poller.py
index 45992e4..c41f3d4 100644
--- a/src/srht_contrib/jobs/poller.py
+++ b/src/srht_contrib/jobs/poller.py
@@ -18,6 +18,9 @@ from srht_contrib.utils.repositories import canonicalize_repository_name
 
 logger = logging.getLogger(__name__)
 SYNC_OVERLAP = timedelta(hours=24)
+RECENT_BACKFILL_DAYS = 365
+RECENT_BACKFILL_BATCHES_PER_SERVICE = 5
+FULL_BACKFILL_BATCHES_PER_SERVICE = 1
 
 
 class PollerService:
@@ -63,7 +66,12 @@ class PollerService:
     def track_actor_request(self, db: Session, actor: str, *, update_last_requested: bool = True) -> TrackedActor:
         tracked_actor = db.scalar(select(TrackedActor).where(TrackedActor.actor == actor))
         if tracked_actor is None:
-            tracked_actor = TrackedActor(actor=actor, is_active=True, backfill_status="pending")
+            tracked_actor = TrackedActor(
+                actor=actor,
+                is_active=True,
+                recent_backfill_status="pending",
+                backfill_status="pending",
+            )
             db.add(tracked_actor)
 
         tracked_actor.is_active = True
@@ -177,32 +185,98 @@ class PollerService:
 
     def _run_backfill_batches(self, db: Session, actor: str) -> int:
         tracked_actor = self.track_actor_request(db, actor, update_last_requested=False)
-        if tracked_actor.backfill_status == "completed":
+        if tracked_actor.recent_backfill_status == "completed" and tracked_actor.backfill_status == "completed":
             return 0
 
-        if tracked_actor.backfill_started_at is None:
-            tracked_actor.backfill_started_at = datetime.now(tz=UTC)
-        tracked_actor.backfill_status = "in_progress"
-        tracked_actor.last_backfill_error = None
+        now = datetime.now(tz=UTC)
+        recent_since = now - timedelta(days=RECENT_BACKFILL_DAYS)
         db.add(tracked_actor)
         db.flush()
         total_inserted = 0
 
-        services = [
-            (self.todo_service.service_name, self.todo_service.fetch_backfill_batch),
-            (self.git_service.service_name, self.git_service.fetch_backfill_batch),
+        recent_services = [
+            (self.todo_service.service_name, self.todo_service.fetch_recent_backfill_batch),
+            (self.git_service.service_name, self.git_service.fetch_recent_backfill_batch),
         ]
-        all_complete = True
+        if tracked_actor.recent_backfill_status != "completed":
+            if tracked_actor.recent_backfill_started_at is None:
+                tracked_actor.recent_backfill_started_at = now
+            tracked_actor.recent_backfill_status = "in_progress"
+            tracked_actor.last_recent_backfill_error = None
+            total_inserted += self._run_backfill_scope(
+                db,
+                actor=actor,
+                scope="recent",
+                services=recent_services,
+                batches_per_service=RECENT_BACKFILL_BATCHES_PER_SERVICE,
+                since=recent_since,
+            )
+            recent_statuses = db.scalars(
+                select(ServiceBackfillState.status)
+                .where(ServiceBackfillState.actor == actor)
+                .where(ServiceBackfillState.scope == "recent")
+            ).all()
+            if recent_statuses and all(status == "completed" for status in recent_statuses):
+                tracked_actor.recent_backfill_status = "completed"
+                tracked_actor.recent_backfill_completed_at = datetime.now(tz=UTC)
+                tracked_actor.last_recent_backfill_error = None
+            db.add(tracked_actor)
+            db.flush()
+
+        if tracked_actor.recent_backfill_status == "completed" and tracked_actor.backfill_status != "completed":
+            if tracked_actor.backfill_started_at is None:
+                tracked_actor.backfill_started_at = now
+            tracked_actor.backfill_status = "in_progress"
+            tracked_actor.last_backfill_error = None
+            total_inserted += self._run_backfill_scope(
+                db,
+                actor=actor,
+                scope="full",
+                services=[
+                    (self.todo_service.service_name, self.todo_service.fetch_backfill_batch),
+                    (self.git_service.service_name, self.git_service.fetch_backfill_batch),
+                ],
+                batches_per_service=FULL_BACKFILL_BATCHES_PER_SERVICE,
+                since=None,
+            )
+            full_statuses = db.scalars(
+                select(ServiceBackfillState.status)
+                .where(ServiceBackfillState.actor == actor)
+                .where(ServiceBackfillState.scope == "full")
+            ).all()
+            if full_statuses and all(status == "completed" for status in full_statuses):
+                tracked_actor.backfill_status = "completed"
+                tracked_actor.backfill_completed_at = datetime.now(tz=UTC)
+                tracked_actor.last_backfill_error = None
+            db.add(tracked_actor)
+            db.flush()
+
+        return total_inserted
+
+    def _run_backfill_scope(
+        self,
+        db: Session,
+        *,
+        actor: str,
+        scope: str,
+        services,
+        batches_per_service: int,
+        since: datetime | None,
+    ) -> int:
+        tracked_actor = self.track_actor_request(db, actor, update_last_requested=False)
+        total_inserted = 0
         for service_name, fetcher in services:
             state = db.scalar(
                 select(ServiceBackfillState)
                 .where(ServiceBackfillState.actor == actor)
                 .where(ServiceBackfillState.service == service_name)
+                .where(ServiceBackfillState.scope == scope)
             )
             if state is None:
                 state = ServiceBackfillState(
                     actor=actor,
                     service=service_name,
+                    scope=scope,
                     cursor_json=None,
                     status="pending",
                     started_at=None,
@@ -216,47 +290,56 @@ class PollerService:
             if state.status == "completed":
                 continue
 
-            all_complete = False
             if state.started_at is None:
                 state.started_at = datetime.now(tz=UTC)
             state.status = "in_progress"
             state.updated_at = datetime.now(tz=UTC)
-            try:
-                result = fetcher(actor=actor, cursor_state=state.cursor_json)
-                inserted = self._insert_events(db, result.events)
-                total_inserted += inserted
-                state.cursor_json = copy.deepcopy(result.cursor_state)
-                state.last_error = None
-                state.updated_at = datetime.now(tz=UTC)
-                if result.complete:
-                    state.status = "completed"
-                    state.completed_at = datetime.now(tz=UTC)
-                    logger.info("Backfill complete for service=%s actor=%s inserted=%s", service_name, actor, inserted)
-                else:
-                    logger.info("Backfill batch complete for service=%s actor=%s inserted=%s", service_name, actor, inserted)
-            except Exception as exc:
-                state.status = "error"
-                state.last_error = str(exc)
-                state.updated_at = datetime.now(tz=UTC)
-                tracked_actor.backfill_status = "error"
-                tracked_actor.last_backfill_error = str(exc)
-                db.add(state)
-                db.add(tracked_actor)
-                db.flush()
-                raise
+
+            for _ in range(batches_per_service):
+                try:
+                    if since is None:
+                        result = fetcher(actor=actor, cursor_state=state.cursor_json)
+                    else:
+                        result = fetcher(actor=actor, cursor_state=state.cursor_json, since=since)
+                    inserted = self._insert_events(db, result.events)
+                    total_inserted += inserted
+                    state.cursor_json = copy.deepcopy(result.cursor_state)
+                    state.last_error = None
+                    state.updated_at = datetime.now(tz=UTC)
+                    if result.complete:
+                        state.status = "completed"
+                        state.completed_at = datetime.now(tz=UTC)
+                        logger.info(
+                            "Backfill complete for scope=%s service=%s actor=%s inserted=%s",
+                            scope,
+                            service_name,
+                            actor,
+                            inserted,
+                        )
+                        break
+                    logger.info(
+                        "Backfill batch complete for scope=%s service=%s actor=%s inserted=%s",
+                        scope,
+                        service_name,
+                        actor,
+                        inserted,
+                    )
+                except Exception as exc:
+                    state.status = "error"
+                    state.last_error = str(exc)
+                    state.updated_at = datetime.now(tz=UTC)
+                    if scope == "recent":
+                        tracked_actor.recent_backfill_status = "error"
+                        tracked_actor.last_recent_backfill_error = str(exc)
+                    else:
+                        tracked_actor.backfill_status = "error"
+                        tracked_actor.last_backfill_error = str(exc)
+                    db.add(state)
+                    db.add(tracked_actor)
+                    db.flush()
+                    raise
 
             db.add(state)
             db.flush()
 
-        completed = db.scalars(
-            select(ServiceBackfillState.status).where(ServiceBackfillState.actor == actor)
-        ).all()
-        if completed and all(status == "completed" for status in completed):
-            tracked_actor.backfill_status = "completed"
-            tracked_actor.backfill_completed_at = datetime.now(tz=UTC)
-            tracked_actor.last_backfill_error = None
-        elif tracked_actor.backfill_status != "error":
-            tracked_actor.backfill_status = "in_progress"
-        db.add(tracked_actor)
-        db.flush()
         return total_inserted
diff --git a/src/srht_contrib/models.py b/src/srht_contrib/models.py
index a99a306..475bd22 100644
--- a/src/srht_contrib/models.py
+++ b/src/srht_contrib/models.py
@@ -71,6 +71,10 @@ class TrackedActor(Base):
     last_polled_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     last_poll_status: Mapped[str | None] = mapped_column(String(32), nullable=True)
     last_poll_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    recent_backfill_status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending")
+    recent_backfill_started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    recent_backfill_completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    last_recent_backfill_error: Mapped[str | None] = mapped_column(Text, nullable=True)
     backfill_status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending")
     backfill_started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     backfill_completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
@@ -79,11 +83,12 @@ class TrackedActor(Base):
 
 class ServiceBackfillState(Base):
     __tablename__ = "service_backfill_states"
-    __table_args__ = (UniqueConstraint("actor", "service", name="uq_service_backfill_state_actor_service"),)
+    __table_args__ = (UniqueConstraint("actor", "service", "scope", name="uq_service_backfill_state_actor_service_scope"),)
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
     actor: Mapped[str] = mapped_column(String(255), nullable=False)
     service: Mapped[str] = mapped_column(String(32), nullable=False)
+    scope: Mapped[str] = mapped_column(String(16), nullable=False, default="full")
     cursor_json: Mapped[dict | None] = mapped_column(JSON, nullable=True)
     status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending")
     started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
diff --git a/src/srht_contrib/schemas.py b/src/srht_contrib/schemas.py
index 6c77f6b..98d2a74 100644
--- a/src/srht_contrib/schemas.py
+++ b/src/srht_contrib/schemas.py
@@ -28,6 +28,9 @@ class ContributionIndexMetadata(BaseModel):
     is_indexed: bool
     last_polled_at: datetime | None = None
     indexing_state: Literal["pending", "indexed", "error"]
+    is_recent_window_backfilled: bool = False
+    recent_backfill_state: Literal["pending", "in_progress", "completed", "error"] = "pending"
+    recent_backfill_completed_at: datetime | None = None
     is_backfilled: bool = False
     backfill_state: Literal["pending", "in_progress", "completed", "error"] = "pending"
     backfill_completed_at: datetime | None = None
diff --git a/src/srht_contrib/services/aggregator.py b/src/srht_contrib/services/aggregator.py
index 1c8af86..4973f16 100644
--- a/src/srht_contrib/services/aggregator.py
+++ b/src/srht_contrib/services/aggregator.py
@@ -62,6 +62,12 @@ class ContributionAggregator:
             is_indexed=calendar.is_indexed,
             last_polled_at=calendar.last_polled_at,
             indexing_state=calendar.indexing_state,
+            is_recent_window_backfilled=calendar.is_recent_window_backfilled,
+            recent_backfill_state=calendar.recent_backfill_state,
+            recent_backfill_completed_at=calendar.recent_backfill_completed_at,
+            is_backfilled=calendar.is_backfilled,
+            backfill_state=calendar.backfill_state,
+            backfill_completed_at=calendar.backfill_completed_at,
         )
 
     def _index_metadata(self, db: Session, actor: str) -> ContributionIndexMetadata:
@@ -81,6 +87,9 @@ class ContributionAggregator:
             is_indexed=is_indexed,
             last_polled_at=tracked_actor.last_polled_at if tracked_actor is not None else None,
             indexing_state=indexing_state,
+            is_recent_window_backfilled=(tracked_actor.recent_backfill_status == "completed") if tracked_actor is not None else False,
+            recent_backfill_state=(tracked_actor.recent_backfill_status if tracked_actor is not None else "pending"),
+            recent_backfill_completed_at=tracked_actor.recent_backfill_completed_at if tracked_actor is not None else None,
             is_backfilled=(tracked_actor.backfill_status == "completed") if tracked_actor is not None else False,
             backfill_state=(tracked_actor.backfill_status if tracked_actor is not None else "pending"),
             backfill_completed_at=tracked_actor.backfill_completed_at if tracked_actor is not None else None,
diff --git a/src/srht_contrib/services/git.py b/src/srht_contrib/services/git.py
index ebb823e..bf7493b 100644
--- a/src/srht_contrib/services/git.py
+++ b/src/srht_contrib/services/git.py
@@ -120,6 +120,24 @@ class GitIngestionService:
         return repositories
 
     def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+        return self._fetch_backfill_batch(actor, cursor_state, since=None)
+
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return self._fetch_backfill_batch(actor, cursor_state, since=since)
+
+    def _fetch_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None,
+        *,
+        since: datetime | None,
+    ) -> BackfillBatchResult:
         state = {
             "discovery_cursor": None,
             "discovery_complete": False,
@@ -186,13 +204,18 @@ class GitIngestionService:
         log_page = repository.get("log") or {}
         commits = log_page.get("results") or []
         next_cursor = log_page.get("cursor")
-        events = [
-            normalized
-            for commit in commits
-            if isinstance(commit, dict)
-            for normalized in [self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit)]
-            if normalized is not None
-        ]
+        events: list[NormalizedEvent] = []
+        stop_repository = False
+        for commit in commits:
+            if not isinstance(commit, dict):
+                continue
+            commit_time = parse_datetime((commit.get("author") or {}).get("time"))
+            if since is not None and commit_time < since:
+                stop_repository = True
+                break
+            normalized = self._normalize_commit(actor=actor, repo_name=repo_name, commit=commit)
+            if normalized is not None:
+                events.append(normalized)
         logger.info(
             "git backfill actor=%s repository=%s commits=%s next_cursor=%s",
             actor,
@@ -200,7 +223,7 @@ class GitIngestionService:
             len(commits),
             bool(next_cursor),
         )
-        if next_cursor:
+        if next_cursor and not stop_repository:
             state["current_repository"]["cursor"] = next_cursor
         else:
             state["current_repository"] = None
diff --git a/src/srht_contrib/services/todo.py b/src/srht_contrib/services/todo.py
index b7209b1..9f2ece6 100644
--- a/src/srht_contrib/services/todo.py
+++ b/src/srht_contrib/services/todo.py
@@ -296,6 +296,24 @@ class TodoIngestionService:
         return TodoPollResult(events=tracker_events, cursor=cursor_time)
 
     def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
+        return self._fetch_backfill_batch(actor, cursor_state, since=None)
+
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return self._fetch_backfill_batch(actor, cursor_state, since=since)
+
+    def _fetch_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None,
+        *,
+        since: datetime | None,
+    ) -> BackfillBatchResult:
         state = {
             "trackers_cursor": None,
             "tracker_queue": [],
@@ -349,10 +367,20 @@ class TodoIngestionService:
             tracker = data.get("tracker") or {}
             tickets_page = tracker.get("tickets") or {}
             tickets = tickets_page.get("results") or []
-            current_tracker["tickets_cursor"] = tickets_page.get("cursor")
-            current_tracker["pending_tickets"].extend(
-                [{"id": int(ticket["id"]), "ref": str(ticket.get("ref") or ticket["id"])} for ticket in tickets if isinstance(ticket, dict)]
-            )
+            next_tickets_cursor = tickets_page.get("cursor")
+            stop_tracker_paging = False
+            for ticket in tickets:
+                if not isinstance(ticket, dict):
+                    continue
+                if since is not None:
+                    updated = parse_datetime(ticket["updated"])
+                    if updated < since:
+                        stop_tracker_paging = True
+                        continue
+                current_tracker["pending_tickets"].append(
+                    {"id": int(ticket["id"]), "ref": str(ticket.get("ref") or ticket["id"])}
+                )
+            current_tracker["tickets_cursor"] = None if stop_tracker_paging else next_tickets_cursor
             logger.info(
                 "todo backfill tracker=%s ticket page count=%s next_cursor=%s pending_tickets=%s",
                 current_tracker.get("name") or current_tracker.get("id"),
@@ -391,6 +419,7 @@ class TodoIngestionService:
         )
 
         events: list[NormalizedEvent] = []
+        stop_ticket_paging = False
         for event in page_events:
             if not isinstance(event, dict):
                 continue
@@ -402,6 +431,9 @@ class TodoIngestionService:
                 "tracker": {"name": current_tracker.get("name")},
             }
             occurred_at = parse_datetime(event["created"])
+            if since is not None and occurred_at < since:
+                stop_ticket_paging = True
+                continue
             for change in event.get("changes") or []:
                 if not isinstance(change, dict):
                     continue
@@ -415,7 +447,7 @@ class TodoIngestionService:
                 if normalized is not None:
                     events.append(normalized)
 
-        if next_cursor:
+        if next_cursor and not stop_ticket_paging:
             state["current_ticket"]["cursor"] = next_cursor
         else:
             state["current_ticket"] = None
diff --git a/tests/test_contributions_api.py b/tests/test_contributions_api.py
index f2b8564..15af7d8 100644
--- a/tests/test_contributions_api.py
+++ b/tests/test_contributions_api.py
@@ -48,6 +48,8 @@ def test_contributions_api_returns_zero_filled_range(client: TestClient, db_sess
     assert response.status_code == 200
     assert response.json()["is_indexed"] is True
     assert response.json()["indexing_state"] == "indexed"
+    assert response.json()["is_recent_window_backfilled"] is False
+    assert response.json()["recent_backfill_state"] == "pending"
     assert response.json()["days"] == [
         {"date": "2026-03-28", "count": 0, "score": 0.0},
         {"date": "2026-03-29", "count": 0, "score": 0.0},
@@ -63,6 +65,9 @@ def test_public_read_registers_actor_for_lazy_indexing(client: TestClient, db_se
     assert response.status_code == 200
     assert response.json()["is_indexed"] is False
     assert response.json()["indexing_state"] == "pending"
+    assert response.json()["is_recent_window_backfilled"] is False
+    assert response.json()["recent_backfill_state"] == "pending"
+    assert response.json()["recent_backfill_completed_at"] is None
     assert response.json()["is_backfilled"] is False
     assert response.json()["backfill_state"] == "pending"
     assert response.json()["backfill_completed_at"] is None
@@ -110,6 +115,8 @@ def test_contribution_stats_api(client: TestClient, db_session) -> None:
     assert response.json()["current_streak"] == 2
     assert response.json()["is_indexed"] is True
     assert response.json()["indexing_state"] == "indexed"
+    assert response.json()["is_recent_window_backfilled"] is False
+    assert response.json()["recent_backfill_state"] == "pending"
     assert response.json()["is_backfilled"] is False
 
 
diff --git a/tests/test_ingestion.py b/tests/test_ingestion.py
index 0cb2abc..75e3710 100644
--- a/tests/test_ingestion.py
+++ b/tests/test_ingestion.py
@@ -41,6 +41,15 @@ class RecordingTodoService:
     def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
         return BackfillBatchResult(events=[], cursor_state=None, complete=True)
 
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return BackfillBatchResult(events=[], cursor_state=None, complete=True)
+
 
 class EmptyGitService:
     service_name = "git"
@@ -61,6 +70,15 @@ class EmptyGitService:
     def fetch_backfill_batch(self, actor: str, cursor_state: dict | None = None) -> BackfillBatchResult:
         return BackfillBatchResult(events=[], cursor_state=None, complete=True)
 
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return BackfillBatchResult(events=[], cursor_state=None, complete=True)
+
 
 class BackfillingTodoService:
     service_name = "todo"
@@ -82,6 +100,15 @@ class BackfillingTodoService:
         )
         return BackfillBatchResult(events=[event], cursor_state=None, complete=True)
 
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return self.fetch_backfill_batch(actor, cursor_state)
+
 
 class QueueShrinkingTodoService:
     service_name = "todo"
@@ -100,6 +127,15 @@ class QueueShrinkingTodoService:
         state["tracker_queue"].pop(0)
         return BackfillBatchResult(events=[], cursor_state=state, complete=False)
 
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return self.fetch_backfill_batch(actor, cursor_state)
+
 
 class QueueShrinkingGitService:
     service_name = "git"
@@ -128,6 +164,15 @@ class QueueShrinkingGitService:
         state["repository_queue"].pop(0)
         return BackfillBatchResult(events=[], cursor_state=state, complete=False)
 
+    def fetch_recent_backfill_batch(
+        self,
+        actor: str,
+        cursor_state: dict | None = None,
+        *,
+        since: datetime,
+    ) -> BackfillBatchResult:
+        return self.fetch_backfill_batch(actor, cursor_state)
+
 
 def make_settings(**overrides) -> Settings:
     values = {
@@ -507,14 +552,18 @@ def test_poll_marks_backfill_complete_and_persists_service_state(db_session) ->
 
     tracked_actor = db_session.scalar(select(TrackedActor).where(TrackedActor.actor == "~ccleberg"))
     service_states = db_session.scalars(
-        select(ServiceBackfillState).where(ServiceBackfillState.actor == "~ccleberg").order_by(ServiceBackfillState.service)
+        select(ServiceBackfillState)
+        .where(ServiceBackfillState.actor == "~ccleberg")
+        .order_by(ServiceBackfillState.scope, ServiceBackfillState.service)
     ).all()
 
     assert inserted == 1
     assert tracked_actor is not None
+    assert tracked_actor.recent_backfill_status == "completed"
+    assert tracked_actor.recent_backfill_completed_at is not None
     assert tracked_actor.backfill_status == "completed"
     assert tracked_actor.backfill_completed_at is not None
-    assert [state.service for state in service_states] == ["git", "todo"]
+    assert [f"{state.scope}:{state.service}" for state in service_states] == ["full:git", "full:todo", "recent:git", "recent:todo"]
     assert all(state.status == "completed" for state in service_states)
 
 
@@ -525,7 +574,9 @@ def test_backfill_cursor_state_shrinks_across_repeated_polls(db_session) -> None
     first_states = {
         state.service: state.cursor_json
         for state in db_session.scalars(
-            select(ServiceBackfillState).where(ServiceBackfillState.actor == "~ccleberg")
+            select(ServiceBackfillState)
+            .where(ServiceBackfillState.actor == "~ccleberg")
+            .where(ServiceBackfillState.scope == "full")
         ).all()
     }
 
@@ -533,7 +584,9 @@ def test_backfill_cursor_state_shrinks_across_repeated_polls(db_session) -> None
     second_states = {
         state.service: state.cursor_json
         for state in db_session.scalars(
-            select(ServiceBackfillState).where(ServiceBackfillState.actor == "~ccleberg")
+            select(ServiceBackfillState)
+            .where(ServiceBackfillState.actor == "~ccleberg")
+            .where(ServiceBackfillState.scope == "full")
         ).all()
     }