frontend/index.html

@@ -1216,8 +1216,12 @@ textarea { resize: vertical; min-height: 120px; }
   }
 
   function renderLiveSnapshot(snapshot) {
-    const notificationsFresh = Array.isArray(snapshot.notifications);
+    const contextFreshness = snapshot.freshness?.sections?.context;
-    if (notificationsFresh) {
+    const eventsFreshness = snapshot.freshness?.sections?.events;
+    const notificationFreshness = snapshot.freshness?.sections?.notifications;
+    const hasNotifications = Array.isArray(snapshot.notifications);
+    const notificationsFresh = hasNotifications && !notificationFreshness?.stale;
+    if (hasNotifications) {
       if (notificationPagination.page > 1) {
         const byId = new Map(lastNotifications.map(item => [item.id, item]));
         snapshot.notifications.forEach(item => byId.set(item.id, item));
@@ -1238,19 +1242,27 @@ textarea { resize: vertical; min-height: 120px; }
     if (snapshot.context) {
       snapshot.context.notifications = lastNotifications;
       renderContextSnapshot(snapshot.context);
-      if (!notificationsFresh) markNotificationsStale();
+      if (contextFreshness?.stale) markMyWorkStale();
+      else if (!notificationsFresh) markNotificationsStale();
     } else handleContextError(new Error('Context section unavailable'));
-    if (snapshot.events) {
+    if (Array.isArray(snapshot.events)) paintEventStream(snapshot.events);
-      paintEventStream(snapshot.events);
+    if (eventsFreshness?.revalidating) {
+      setEventStreamStatus('Refreshing activity · showing last activity');
+    } else if (eventsFreshness?.stale || eventsFreshness?.degraded) {
+      const retrySeconds = Number(eventsFreshness.retry_in_seconds) || 0;
+      setEventStreamStatus('Activity refresh failed · showing last activity' +
+        (retrySeconds > 0 ? ' · retrying in ' + retrySeconds + 's' : ''));
+    } else if (Array.isArray(snapshot.events)) {
       setEventStreamStatus('Updated ' + fmt(new Date()));
     } else {
       setEventStreamStatus('Update failed · showing last activity');
     }
-    if (snapshot.freshness?.degraded && !snapshot.freshness.revalidating) {
+    // Compatibility fallback for snapshots produced before section metadata.
+    if (!eventsFreshness && snapshot.freshness?.degraded && !snapshot.freshness.revalidating) {
       const retrySeconds = Number(snapshot.freshness.retry_in_seconds) || 0;
       setEventStreamStatus('Refresh failed · showing last known data' +
         (retrySeconds > 0 ? ' · retrying in ' + retrySeconds + 's' : ''));
-    } else if (snapshot.freshness?.revalidating) {
+    } else if (!eventsFreshness && snapshot.freshness?.revalidating) {
       setEventStreamStatus('Refreshing · showing recent snapshot');
     }
   }

src/main.py

@@ -1,8 +1,10 @@
 import asyncio
 import math
 import time
+from collections.abc import Awaitable
 from contextlib import asynccontextmanager
 from pathlib import Path
+from typing import Any
 
 from fastapi import FastAPI, HTTPException, Path as PathParam, Query
 from fastapi.responses import JSONResponse
@@ -65,8 +67,17 @@ FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
 _live_snapshot_task: asyncio.Task | None = None
 _live_snapshot_value: dict | None = None
 _live_snapshot_created_at: float | None = None
-_live_snapshot_failure_count = 0
+LIVE_SNAPSHOT_SECTIONS = ("context", "events", "notifications")
-_live_snapshot_retry_at: float | None = None
+_live_section_created_at: dict[str, float | None] = {
+    section: None for section in LIVE_SNAPSHOT_SECTIONS
+}
+_live_section_failure_count: dict[str, int] = {
+    section: 0 for section in LIVE_SNAPSHOT_SECTIONS
+}
+_live_section_retry_at: dict[str, float | None] = {
+    section: None for section in LIVE_SNAPSHOT_SECTIONS
+}
+_live_snapshot_refreshing_sections: set[str] = set()
 _read_notification_ids: set[int] = set()
 
 
@@ -337,19 +348,41 @@ async def _load_context_for_user(user_data: dict) -> dict:
     return _context_payload(user_data, repo_data, issues_data, prs_data)
 
 
-async def _build_live_snapshot() -> dict:
+async def _build_live_snapshot(sections: set[str] | None = None) -> dict:
+    requested = set(sections or LIVE_SNAPSHOT_SECTIONS)
+    results: dict[str, object] = {}
+    user_data: dict | None = None
+    if requested & {"context", "events"}:
+        try:
             user_data = await current_user()
             if not isinstance(user_data, dict) or not user_data.get("login"):
                 raise ContextPayloadError("Gitea current-user response was invalid")
-    context_result, events_result, notifications_result = await asyncio.gather(
+        except Exception as exc:
-        _load_context_for_user(user_data),
+            for section in requested & {"context", "events"}:
-        activity_events(user_data),
+                results[section] = exc
-        notifications(),
+
-        return_exceptions=True,
+    loads: dict[str, Awaitable[Any]] = {}
+    if "context" in requested and "context" not in results:
+        assert user_data is not None
+        loads["context"] = _load_context_for_user(user_data)
+    if "events" in requested and "events" not in results:
+        assert user_data is not None
+        loads["events"] = activity_events(user_data)
+    if "notifications" in requested:
+        loads["notifications"] = notifications()
+    if loads:
+        loaded = await asyncio.gather(*loads.values(), return_exceptions=True)
+        results.update(zip(loads, loaded))
+
+    context_result = results.get("context")
+    events_result = results.get("events")
+    notifications_result = results.get("notifications")
+    context_ok = "context" in requested and not isinstance(context_result, BaseException)
+    events_ok = "events" in requested and not isinstance(events_result, BaseException)
+    notifications_ok = (
+        "notifications" in requested
+        and not isinstance(notifications_result, BaseException)
     )
-    context_ok = not isinstance(context_result, BaseException)
-    events_ok = not isinstance(events_result, BaseException)
-    notifications_ok = not isinstance(notifications_result, BaseException)
     notification_items = notifications_result
     notification_pagination = None
     if notifications_ok and isinstance(notifications_result, dict):
@@ -366,64 +399,88 @@ async def _build_live_snapshot() -> dict:
         "notifications": notification_items if notifications_ok else None,
         "notification_pagination": notification_pagination if notifications_ok else None,
         "sections": {
-            "context": "fresh" if context_ok else "temporarily unavailable",
+            section: "fresh" if ok else "temporarily unavailable"
-            "events": "fresh" if events_ok else "temporarily unavailable",
+            for section, ok in (
-            "notifications": "fresh" if notifications_ok else "temporarily unavailable",
+                ("context", context_ok),
+                ("events", events_ok),
+                ("notifications", notifications_ok),
+            )
+            if section in requested
         },
     }
 
 
-async def _build_live_snapshot_before_deadline() -> dict:
+async def _build_live_snapshot_before_deadline(sections: set[str]) -> dict:
     async with asyncio.timeout(CONTEXT_TIMEOUT_SECONDS):
-        return await _build_live_snapshot()
+        return await _build_live_snapshot(sections)
 
 
-def _record_live_snapshot_failure() -> None:
+def _record_live_section_failure(section: str) -> None:
-    global _live_snapshot_failure_count, _live_snapshot_retry_at
+    _live_section_failure_count[section] += 1
-    _live_snapshot_failure_count += 1
     delay = min(
         LIVE_SNAPSHOT_RETRY_MAX_SECONDS,
-        LIVE_SNAPSHOT_RETRY_BASE_SECONDS * (2 ** (_live_snapshot_failure_count - 1)),
+        LIVE_SNAPSHOT_RETRY_BASE_SECONDS
+        * (2 ** (_live_section_failure_count[section] - 1)),
     )
-    _live_snapshot_retry_at = time.monotonic() + delay
+    _live_section_retry_at[section] = time.monotonic() + delay
+
+
+def _due_live_sections(now: float) -> set[str]:
+    due = set()
+    for section in LIVE_SNAPSHOT_SECTIONS:
+        retry_at = _live_section_retry_at[section]
+        if retry_at is not None and now < retry_at:
+            continue
+        created_at = _live_section_created_at[section]
+        if created_at is None or now - created_at >= LIVE_SNAPSHOT_FRESHNESS_SECONDS:
+            due.add(section)
+    return due
 
 
 def _merge_live_snapshot(previous: dict | None, refreshed: dict) -> dict:
     if previous is None:
         return refreshed
-    merged = dict(refreshed)
+    merged = dict(previous)
-    sections = dict(refreshed.get("sections") or {})
+    sections = dict(previous.get("sections") or {})
-    for section in ("context", "events", "notifications"):
+    for section, state in (refreshed.get("sections") or {}).items():
-        if sections.get(section) == "fresh":
+        if state == "fresh":
-            continue
+            merged[section] = refreshed.get(section)
-        if previous.get(section) is not None:
+            sections[section] = "fresh"
-            merged[section] = previous[section]
-            sections[section] = "stale"
             if section == "notifications":
-                merged["notification_pagination"] = previous.get(
+                merged["notification_pagination"] = refreshed.get(
                     "notification_pagination"
                 )
+        elif previous.get(section) is not None:
+            sections[section] = "stale"
+        else:
+            merged[section] = None
+            sections[section] = "temporarily unavailable"
     merged["sections"] = sections
     return merged
 
 
-async def _refresh_live_snapshot() -> dict:
+async def _refresh_live_snapshot(sections: set[str]) -> dict:
     global _live_snapshot_value, _live_snapshot_created_at
-    global _live_snapshot_failure_count, _live_snapshot_retry_at
     try:
-        result = await _build_live_snapshot_before_deadline()
+        refreshed = await _build_live_snapshot_before_deadline(sections)
     except Exception:
-        _record_live_snapshot_failure()
+        for section in sections:
+            _record_live_section_failure(section)
         raise
-    result = _merge_live_snapshot(_live_snapshot_value, result)
+    now = time.monotonic()
+    refreshed_states = refreshed.get("sections") or {}
+    for section in sections:
+        if refreshed_states.get(section) == "fresh":
+            _live_section_created_at[section] = now
+            _live_section_failure_count[section] = 0
+            _live_section_retry_at[section] = None
+        else:
+            _record_live_section_failure(section)
+    result = _merge_live_snapshot(_live_snapshot_value, refreshed)
     result = _without_read_notifications(result)
     _live_snapshot_value = result
-    _live_snapshot_created_at = time.monotonic()
+    successful_times = [value for value in _live_section_created_at.values() if value is not None]
-    if any(state != "fresh" for state in result.get("sections", {}).values()):
+    _live_snapshot_created_at = max(successful_times) if successful_times else None
-        _record_live_snapshot_failure()
-    else:
-        _live_snapshot_failure_count = 0
-        _live_snapshot_retry_at = None
     return result
 
 
@@ -433,34 +490,47 @@ def _consume_live_snapshot_failure(task: asyncio.Task) -> None:
     task.exception()
 
 
-def _start_live_snapshot_refresh() -> asyncio.Task:
+def _start_live_snapshot_refresh(sections: set[str]) -> asyncio.Task:
     global _live_snapshot_task
+    global _live_snapshot_refreshing_sections
     if _live_snapshot_task is None or _live_snapshot_task.done():
-        _live_snapshot_task = asyncio.create_task(_refresh_live_snapshot())
+        _live_snapshot_refreshing_sections = set(sections)
+        _live_snapshot_task = asyncio.create_task(_refresh_live_snapshot(sections))
         _live_snapshot_task.add_done_callback(_consume_live_snapshot_failure)
     return _live_snapshot_task
 
 
 def _live_snapshot_payload(value: dict, *, stale: bool, revalidating: bool) -> dict:
     payload = dict(value)
-    age = (
+    now = time.monotonic()
-        max(0.0, time.monotonic() - _live_snapshot_created_at)
+    section_freshness = {}
-        if _live_snapshot_created_at is not None
+    for section in LIVE_SNAPSHOT_SECTIONS:
-        else 0.0
+        created_at = _live_section_created_at[section]
-    )
+        retry_at = _live_section_retry_at[section]
-    retry_in_seconds = (
+        age = max(0.0, now - created_at) if created_at is not None else 0.0
-        max(0.0, _live_snapshot_retry_at - time.monotonic())
+        retry_in = max(0.0, retry_at - now) if retry_at is not None else 0.0
-        if _live_snapshot_retry_at is not None
+        section_freshness[section] = {
-        else 0.0
-    )
-    payload["freshness"] = {
             "age_seconds": round(age, 3),
+            "stale": (value.get("sections") or {}).get(section) != "fresh"
+            or created_at is None
+            or age >= LIVE_SNAPSHOT_FRESHNESS_SECONDS,
+            "revalidating": revalidating
+            and section in _live_snapshot_refreshing_sections,
+            "degraded": retry_at is not None,
+            "retry_in_seconds": math.ceil(retry_in),
+        }
+    ages = [item["age_seconds"] for item in section_freshness.values()]
+    retries = [item["retry_in_seconds"] for item in section_freshness.values() if item["retry_in_seconds"]]
+    degraded = any(item["degraded"] for item in section_freshness.values())
+    payload["freshness"] = {
+        "age_seconds": max(ages, default=0.0),
         "fresh_for_seconds": LIVE_SNAPSHOT_FRESHNESS_SECONDS,
-        "stale": stale,
+        "stale": stale or any(item["stale"] for item in section_freshness.values()),
         "revalidating": revalidating,
-        "degraded": _live_snapshot_retry_at is not None,
+        "degraded": degraded,
-        "last_refresh_failed": _live_snapshot_retry_at is not None,
+        "last_refresh_failed": degraded,
-        "retry_in_seconds": math.ceil(retry_in_seconds),
+        "retry_in_seconds": min(retries, default=0),
+        "sections": section_freshness,
     }
     return payload
 
@@ -508,33 +578,22 @@ async def live_snapshot() -> JSONResponse:
     """Return a freshness-bounded snapshot and share identical upstream loads."""
     global _live_snapshot_task, _live_snapshot_value, _live_snapshot_created_at
     now = time.monotonic()
+    due_sections = _due_live_sections(now)
     if (
         _live_snapshot_value is not None
-        and _live_snapshot_retry_at is not None
+        and not due_sections
-    ):
-        if now < _live_snapshot_retry_at:
-            return JSONResponse(
-                _live_snapshot_payload(
-                    _live_snapshot_value, stale=True, revalidating=False
-                )
-            )
-        _start_live_snapshot_refresh()
-        return JSONResponse(
-            _live_snapshot_payload(
-                _live_snapshot_value, stale=True, revalidating=True
-            )
-        )
-    if (
-        _live_snapshot_value is not None
-        and _live_snapshot_created_at is not None
-        and now - _live_snapshot_created_at < LIVE_SNAPSHOT_FRESHNESS_SECONDS
     ):
         return JSONResponse(
             _live_snapshot_payload(
-                _live_snapshot_value, stale=False, revalidating=False
+                _live_snapshot_value,
+                stale=any(
+                    state != "fresh"
+                    for state in _live_snapshot_value.get("sections", {}).values()
+                ),
+                revalidating=False,
             )
         )
-    task = _start_live_snapshot_refresh()
+    task = _start_live_snapshot_refresh(due_sections)
     if _live_snapshot_value is not None:
         return JSONResponse(
             _live_snapshot_payload(

tests/test_live_snapshot.py

@@ -11,14 +11,30 @@ def reset_live_snapshot_task():
     main._live_snapshot_task = None
     main._live_snapshot_value = None
     main._live_snapshot_created_at = None
-    main._live_snapshot_failure_count = 0
+    main._live_section_created_at = {
-    main._live_snapshot_retry_at = None
+        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+    }
+    main._live_section_failure_count = {
+        section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
+    }
+    main._live_section_retry_at = {
+        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+    }
+    main._live_snapshot_refreshing_sections = set()
     yield
     main._live_snapshot_task = None
     main._live_snapshot_value = None
     main._live_snapshot_created_at = None
-    main._live_snapshot_failure_count = 0
+    main._live_section_created_at = {
-    main._live_snapshot_retry_at = None
+        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+    }
+    main._live_section_failure_count = {
+        section: 0 for section in main.LIVE_SNAPSHOT_SECTIONS
+    }
+    main._live_section_retry_at = {
+        section: None for section in main.LIVE_SNAPSHOT_SECTIONS
+    }
+    main._live_snapshot_refreshing_sections = set()
 
 
 def payload(response):
@@ -209,7 +225,7 @@ async def test_stale_snapshot_returns_immediately_while_one_refresh_revalidates(
     release_refresh = asyncio.Event()
     builds = 0
 
-    async def snapshot():
+    async def snapshot(_sections=None):
         nonlocal builds
         builds += 1
         if builds == 2:
@@ -247,7 +263,7 @@ async def test_failed_revalidation_enters_cooldown_and_keeps_last_snapshot(monke
     now = 100.0
     builds = 0
 
-    async def snapshot():
+    async def snapshot(_sections=None):
         nonlocal builds
         builds += 1
         if builds > 1:
@@ -288,7 +304,7 @@ async def test_partial_refresh_updates_fresh_sections_and_retains_failed_section
     now = 100.0
     builds = 0
 
-    async def snapshot():
+    async def snapshot(_sections=None):
         nonlocal builds
         builds += 1
         if builds == 1:
@@ -332,6 +348,70 @@ async def test_partial_refresh_updates_fresh_sections_and_retains_failed_section
     assert result["freshness"]["revalidating"] is False
 
 
+@pytest.mark.anyio
+async def test_notification_cooldown_does_not_stop_due_work_and_activity_refreshes(monkeypatch):
+    now = 100.0
+    calls = {"context": 0, "events": 0, "notifications": 0}
+    event_refresh_times = []
+    notification_outage = False
+
+    async def user():
+        return {"id": 1, "login": "timmy"}
+
+    async def work():
+        calls["context"] += 1
+        return []
+
+    async def empty_work():
+        return []
+
+    async def events(_authenticated_user):
+        calls["events"] += 1
+        event_refresh_times.append(now)
+        return [{"generation": calls["events"]}]
+
+    async def updates():
+        calls["notifications"] += 1
+        if notification_outage:
+            raise ConnectionError("notifications unavailable")
+        return []
+
+    monkeypatch.setattr(main.time, "monotonic", lambda: now)
+    monkeypatch.setattr(main, "current_user", user)
+    monkeypatch.setattr(main, "repos", work)
+    monkeypatch.setattr(main, "issues", empty_work)
+    monkeypatch.setattr(main, "pull_requests", empty_work)
+    monkeypatch.setattr(main, "activity_events", events)
+    monkeypatch.setattr(main, "notifications", updates)
+
+    await main.live_snapshot()
+    notification_outage = True
+
+    for elapsed in (
+        main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1,
+        main.LIVE_SNAPSHOT_RETRY_BASE_SECONDS,
+    ):
+        now += elapsed
+        await main.live_snapshot()
+        assert main._live_snapshot_task is not None
+        await main._live_snapshot_task
+
+    # Notification backoff is now 10 seconds. The healthy sections become due
+    # one second before that cooldown expires and must refresh independently.
+    now += main.LIVE_SNAPSHOT_FRESHNESS_SECONDS + 1
+    response = payload(await main.live_snapshot())
+    assert main._live_snapshot_task is not None
+    await main._live_snapshot_task
+    refreshed = payload(await main.live_snapshot())
+
+    assert calls == {"context": 3, "events": 3, "notifications": 3}
+    assert event_refresh_times == [100.0, 109.0, 123.0]
+    assert response["freshness"]["sections"]["context"]["revalidating"] is True
+    assert refreshed["events"] == [{"generation": 3}]
+    assert refreshed["sections"]["notifications"] == "stale"
+    assert refreshed["freshness"]["sections"]["notifications"]["retry_in_seconds"] == 1
+
+
 @pytest.mark.anyio
 async def test_retry_window_starts_one_refresh_and_success_clears_degraded_state(monkeypatch):
     now = 100.0
@@ -339,7 +419,7 @@ async def test_retry_window_starts_one_refresh_and_success_clears_degraded_state
     retry_started = asyncio.Event()
     release_retry = asyncio.Event()
 
-    async def snapshot():
+    async def snapshot(_sections=None):
         nonlocal builds
         builds += 1
         if builds == 1:
@@ -399,7 +479,7 @@ async def test_cancelling_one_waiter_does_not_cancel_the_shared_snapshot(monkeyp
     release = asyncio.Event()
     snapshot_calls = 0
 
-    async def blocked_snapshot():
+    async def blocked_snapshot(_sections=None):
         nonlocal snapshot_calls
         snapshot_calls += 1
         started.set()
@@ -437,7 +517,7 @@ async def test_shared_snapshot_deadline_cancels_upstream_work_for_all_waiters(mo
     started = asyncio.Event()
     cancelled = asyncio.Event()
 
-    async def snapshot_that_exceeds_deadline():
+    async def snapshot_that_exceeds_deadline(_sections=None):
         started.set()
         try:
             await asyncio.Event().wait()

tests/test_notification_read.py

@@ -214,7 +214,7 @@ async def test_inflight_refresh_cannot_restore_a_notification_marked_read(monkey
     async def mark(_thread_id):
         return None
 
-    async def stale_upstream_snapshot():
+    async def stale_upstream_snapshot(_sections):
         refresh_started.set()
         await release_refresh.wait()
         return {
@@ -231,7 +231,7 @@ async def test_inflight_refresh_cannot_restore_a_notification_marked_read(monkey
     monkeypatch.setattr(main, "_live_snapshot_task", None)
     monkeypatch.setattr(main, "_live_snapshot_value", {"notifications": [{"id": 42}]})
 
-    refresh = main._start_live_snapshot_refresh()
+    refresh = main._start_live_snapshot_refresh(set(main.LIVE_SNAPSHOT_SECTIONS))
     await refresh_started.wait()
     await main.read_notification(42)
     release_refresh.set()

tests/test_shared_context_snapshot.py

@@ -44,3 +44,14 @@ async def test_dashboard_announces_failed_refresh_and_retry_without_blanking_pan
     assert "snapshot.freshness?.degraded" in html
     assert "Refresh failed · showing last known data" in html
     assert "snapshot.freshness.retry_in_seconds" in html
+
+
+@pytest.mark.anyio
+async def test_dashboard_reports_each_live_section_from_its_own_freshness():
+    html = await dashboard()
+
+    assert "snapshot.freshness?.sections?.notifications" in html
+    assert "snapshot.freshness?.sections?.events" in html
+    assert "const notificationsFresh = Array.isArray(snapshot.notifications)" not in html
+    assert "Unread updates unavailable · showing last known updates" in html
+    assert "Activity refresh failed · showing last activity" in html