frontend/index.html

@@ -324,8 +324,8 @@ textarea { resize: vertical; min-height: 120px; }
   function setClock() { qs('#clock').textContent = fmt(new Date()); }
   setClock(); setInterval(setClock, 1000);
 
-  async function fetchLiveSnapshot() {
+  async function fetchContextSnapshot() {
-    const res = await fetch('api/v1/live', { headers: { Accept: 'application/json' } });
+    const res = await fetch('api/v1/context', { headers: { Accept: 'application/json' } });
     if (!res.ok) throw new Error('HTTP ' + res.status);
     return res.json();
   }
@@ -552,13 +552,14 @@ textarea { resize: vertical; min-height: 120px; }
     qs('#gitea-events-status').textContent = message;
   }
 
-  function renderLiveSnapshot(snapshot) {
+  async function loadEventStream() {
-    if (snapshot.context) renderContextSnapshot(snapshot.context);
+    setEventStreamStatus('Updating…');
-    else handleContextError(new Error('Context section unavailable'));
+    try {
-    if (snapshot.events) {
+      const res = await fetch('api/v1/events', { headers: { Accept: 'application/json' } });
-      paintEventStream(snapshot.events);
+      if (!res.ok) throw new Error('HTTP ' + res.status);
+      paintEventStream(await res.json());
       setEventStreamStatus('Updated ' + fmt(new Date()));
-    } else {
+    } catch (e) {
       setEventStreamStatus('Update failed · showing last activity');
     }
   }
@@ -663,12 +664,9 @@ textarea { resize: vertical; min-height: 120px; }
   });
 
   const contextPoller = createContextPoller({
-    fetchContext: fetchLiveSnapshot,
+    fetchContext: fetchContextSnapshot,
-    onSnapshot: renderLiveSnapshot,
+    onSnapshot: renderContextSnapshot,
-    onError: error => {
+    onError: handleContextError,
-      handleContextError(error);
-      setEventStreamStatus('Update failed · showing last activity');
-    },
     isHidden: () => document.hidden,
     intervalMs: 8000,
   });
@@ -691,8 +689,11 @@ textarea { resize: vertical; min-height: 120px; }
     });
   });
   contextPoller.start();
+  loadEventStream();
+  setInterval(() => { if (!document.hidden) loadEventStream(); }, 5000);
   document.addEventListener('visibilitychange', () => {
     contextPoller.setVisible(!document.hidden);
+    if (!document.hidden) loadEventStream();
   });
 
   /* Widgets */

src/gitea_proxy.py

@@ -9,7 +9,6 @@ GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/")
 GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
 REVIEW_DIFF_MAX_BYTES = 64 * 1024
 REVIEW_DIFF_MAX_LINES = 400
-_client: httpx.AsyncClient | None = None
 
 
 def _auth() -> dict[str, str]:
@@ -19,47 +18,30 @@ def _auth() -> dict[str, str]:
     return headers
 
 
-def start_client(**kwargs) -> httpx.AsyncClient:
-    """Create the application-lifetime Gitea transport."""
-    global _client
-    _client = httpx.AsyncClient(base_url=GITEA_URL, timeout=10, **kwargs)
-    return _client
-
-
-def _get_client() -> httpx.AsyncClient:
-    if _client is None or _client.is_closed:
-        return start_client()
-    return _client
-
-
-async def stop_client() -> None:
-    global _client
-    if _client is not None and not _client.is_closed:
-        await _client.aclose()
-
-
 async def fetch(path: str) -> Any:
-    r = await _get_client().get(f"/api/v1/{path}", headers=_auth())
+    async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client:
-    r.raise_for_status()
+        r = await client.get(f"/api/v1/{path}", headers=_auth())
-    return r.json()
+        r.raise_for_status()
+        return r.json()
 
 
 async def fetch_text(path: str, max_bytes: int) -> tuple[str, bool]:
     chunks: list[bytes] = []
     size = 0
     truncated = False
-    async with _get_client().stream(
+    async with httpx.AsyncClient(base_url=GITEA_URL, timeout=10) as client:
-        "GET", f"/api/v1/{path}", headers={**_auth(), "Accept": "text/plain"}
+        async with client.stream(
-    ) as response:
+            "GET", f"/api/v1/{path}", headers={**_auth(), "Accept": "text/plain"}
-        response.raise_for_status()
+        ) as response:
-        async for chunk in response.aiter_bytes():
+            response.raise_for_status()
-            remaining = max_bytes - size
+            async for chunk in response.aiter_bytes():
-            if len(chunk) > remaining:
+                remaining = max_bytes - size
-                chunks.append(chunk[:remaining])
+                if len(chunk) > remaining:
-                truncated = True
+                    chunks.append(chunk[:remaining])
-                break
+                    truncated = True
-            chunks.append(chunk)
+                    break
-            size += len(chunk)
+                chunks.append(chunk)
+                size += len(chunk)
     return b"".join(chunks).decode("utf-8", errors="replace"), truncated
 
 
@@ -118,11 +100,11 @@ async def issues() -> list[dict]:
 
 
 async def pull_requests() -> list[dict]:
-    assigned, review_requested = await asyncio.gather(
+    assigned = await fetch(
-        fetch("repos/issues/search?state=open&assigned=true&type=pulls&limit=50"),
+        "repos/issues/search?state=open&assigned=true&type=pulls&limit=50"
-        fetch(
+    )
-            "repos/issues/search?state=open&review_requested=true&type=pulls&limit=50"
+    review_requested = await fetch(
-        ),
+        "repos/issues/search?state=open&review_requested=true&type=pulls&limit=50"
     )
     merged: dict[int, dict] = {}
     for reason, pulls in (
@@ -219,9 +201,8 @@ async def pull_review_detail(repository: str, number: int) -> dict:
     }
 
 
-async def activity_events(user: dict | None = None) -> list[dict]:
+async def activity_events() -> list[dict]:
-    if user is None:
+    user = await current_user()
-        user = await current_user()
     events = await fetch(f"users/{user['login']}/activities/feeds?limit=20")
     if events is None:
         events = []

src/main.py

@@ -1,6 +1,5 @@
 import asyncio
 import math
-from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import FastAPI, HTTPException
@@ -8,7 +7,6 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from fastapi.staticfiles import StaticFiles
 
-from src import gitea_proxy
 from src.gitea_proxy import (
     activity_events,
     current_user,
@@ -22,22 +20,12 @@ from src.models import Issue, PullRequest, Repo, User
 from src.suggestion_engine import compute
 from src.views import router as frontend_router
 
-@asynccontextmanager
+app = FastAPI(title="Stackchain Dashboard")
-async def lifespan(_app: FastAPI):
-    gitea_proxy.start_client()
-    try:
-        yield
-    finally:
-        await gitea_proxy.stop_client()
-
-
-app = FastAPI(title="Stackchain Dashboard", lifespan=lifespan)
 CONTEXT_TIMEOUT_SECONDS = 5.0
 EVENT_STREAM_TIMEOUT_SECONDS = 5.0
 READINESS_TIMEOUT_SECONDS = 5.0
 REVIEW_DETAIL_TIMEOUT_SECONDS = 5.0
 FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
-_live_snapshot_task: asyncio.Task | None = None
 
 
 class ContextPayloadError(ValueError):
@@ -47,52 +35,6 @@ class ContextPayloadError(ValueError):
 class ReadinessPayloadError(ValueError):
     """Raised when Gitea returns a structurally invalid readiness payload."""
 
-
-def _context_payload(user_data, repo_data, issues_data, prs_data) -> dict:
-    user_model = User(
-        id=user_data["id"],
-        login=user_data["login"],
-        full_name=user_data.get("full_name") or "",
-        email=user_data.get("email") or "",
-    )
-    repo_models = [
-        Repo(
-            id=r["id"], name=r["name"], full_name=r["full_name"],
-            description=r.get("description") or "", url=r["html_url"],
-            updated_at=r.get("updated_at", ""),
-        )
-        for r in (repo_data or [])[:50]
-        if isinstance(r, dict)
-        and all(field in r for field in ("id", "name", "full_name", "html_url"))
-    ]
-    issue_models = [
-        Issue(
-            id=i["id"], number=i["number"], title=i["title"], state=i["state"],
-            labels=[label.get("name", "") for label in (i.get("labels") or []) if isinstance(label, dict)],
-            assignees=[assignee.get("login", "") for assignee in (i.get("assignees") or []) if isinstance(assignee, dict)],
-            repository=i["repository"].get("full_name", "") if isinstance(i.get("repository"), dict) else "",
-            updated_at=i.get("updated_at") or "", url=i["html_url"],
-        )
-        for i in (issues_data or [])[:50]
-        if isinstance(i, dict)
-        and all(field in i for field in ("id", "number", "title", "state", "html_url"))
-    ]
-    pr_models = [
-        PullRequest(
-            id=p["id"], number=p["number"], title=p["title"], state=p["state"],
-            user=p["user"].get("login", "") if isinstance(p.get("user"), dict) else "",
-            labels=[label.get("name", "") for label in (p.get("labels") or []) if isinstance(label, dict)],
-            assignees=[assignee.get("login", "") for assignee in (p.get("assignees") or []) if isinstance(assignee, dict)],
-            work_reasons=[reason for reason in (p.get("work_reasons") or []) if reason in ("assigned_to_me", "review_requested")],
-            repository=p["repository"].get("full_name", "") if isinstance(p.get("repository"), dict) else "",
-            updated_at=p.get("updated_at") or "", url=p["html_url"],
-        )
-        for p in (prs_data or [])[:50]
-        if isinstance(p, dict)
-        and all(field in p for field in ("id", "number", "title", "state", "html_url"))
-    ]
-    return compute(user_model, repo_models, issue_models, pr_models).model_dump()
-
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -108,7 +50,7 @@ app.include_router(frontend_router)
 @app.middleware("http")
 async def prevent_live_api_caching(request, call_next):
     response = await call_next(request)
-    if request.url.path in {"/api/v1/context", "/api/v1/events", "/api/v1/live"} or (
+    if request.url.path in {"/api/v1/context", "/api/v1/events"} or (
         request.url.path.startswith("/api/v1/repos/")
         and request.url.path.endswith("/review")
     ):
@@ -203,68 +145,87 @@ async def context() -> JSONResponse:
         }, headers={
             "Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))
         } if isinstance(e, TimeoutError) else None)
-    return JSONResponse(_context_payload(user_data, repo_data, issues_data, prs_data))
+    user_model = User(id=user_data["id"], login=user_data["login"], full_name=user_data.get("full_name") or "", email=user_data.get("email") or "")
-
+    repo_models = [
-
+        Repo(id=r["id"], name=r["name"], full_name=r["full_name"], description=r.get("description") or "", url=r["html_url"], updated_at=r.get("updated_at", ""))
-async def _load_context_for_user(user_data: dict) -> dict:
+        for r in (repo_data or [])[:50]
-    repo_data, issues_data, prs_data = await asyncio.gather(
+        if isinstance(r, dict)
-        repos(), issues(), pull_requests()
+        and all(field in r for field in ("id", "name", "full_name", "html_url"))
-    )
+    ]
-    if not all(field in user_data for field in ("id", "login")):
+    issue_models = [
-        raise ContextPayloadError("Gitea current-user response did not include id and login")
+        Issue(
-    if any(
+            id=i["id"],
-        data is not None and not isinstance(data, list)
+            number=i["number"],
-        for data in (repo_data, issues_data, prs_data)
+            title=i["title"],
-    ):
+            state=i["state"],
-        raise ContextPayloadError("Gitea collection response was not a list")
+            labels=[
-    return _context_payload(user_data, repo_data, issues_data, prs_data)
+                label.get("name", "")
-
+                for label in (i.get("labels") or [])
-
+                if isinstance(label, dict)
-async def _build_live_snapshot() -> dict:
+            ],
-    user_data = await current_user()
+            assignees=[
-    if not isinstance(user_data, dict) or not user_data.get("login"):
+                assignee.get("login", "")
-        raise ContextPayloadError("Gitea current-user response was invalid")
+                for assignee in (i.get("assignees") or [])
-    context_result, events_result = await asyncio.gather(
+                if isinstance(assignee, dict)
-        _load_context_for_user(user_data),
+            ],
-        activity_events(user_data),
+            repository=(
-        return_exceptions=True,
+                i["repository"].get("full_name", "")
-    )
+                if isinstance(i.get("repository"), dict)
-    context_ok = not isinstance(context_result, BaseException)
+                else ""
-    events_ok = not isinstance(events_result, BaseException)
+            ),
-    return {
+            updated_at=i.get("updated_at") or "",
-        "context": context_result if context_ok else None,
+            url=i["html_url"],
-        "events": events_result if events_ok else None,
-        "sections": {
-            "context": "fresh" if context_ok else "temporarily unavailable",
-            "events": "fresh" if events_ok else "temporarily unavailable",
-        },
-    }
-
-
-@app.get("/api/v1/live")
-async def live_snapshot() -> JSONResponse:
-    """Return a fresh, section-aware snapshot; join only an active identical load."""
-    global _live_snapshot_task
-    if _live_snapshot_task is None or _live_snapshot_task.done():
-        _live_snapshot_task = asyncio.create_task(_build_live_snapshot())
-    task = _live_snapshot_task
-    try:
-        result = await asyncio.wait_for(task, timeout=CONTEXT_TIMEOUT_SECONDS)
-        return JSONResponse(result)
-    except TimeoutError:
-        return JSONResponse(
-            {"error": f"Gitea live snapshot timed out after {CONTEXT_TIMEOUT_SECONDS:g}s"},
-            status_code=503,
-            headers={"Retry-After": str(max(1, math.ceil(CONTEXT_TIMEOUT_SECONDS)))},
         )
-    except Exception:
+        for i in (issues_data or [])[:50]
-        return JSONResponse(
+        if isinstance(i, dict)
-            {"error": "Gitea live snapshot is temporarily unavailable"},
+        and all(
-            status_code=503,
+            field in i
+            for field in ("id", "number", "title", "state", "html_url")
         )
-    finally:
+    ]
-        if task.done() and _live_snapshot_task is task:
+    pr_models = [
-            _live_snapshot_task = None
+        PullRequest(
+            id=p["id"],
+            number=p["number"],
+            title=p["title"],
+            state=p["state"],
+            user=(
+                p["user"].get("login", "")
+                if isinstance(p.get("user"), dict)
+                else ""
+            ),
+            labels=[
+                label.get("name", "")
+                for label in (p.get("labels") or [])
+                if isinstance(label, dict)
+            ],
+            assignees=[
+                assignee.get("login", "")
+                for assignee in (p.get("assignees") or [])
+                if isinstance(assignee, dict)
+            ],
+            work_reasons=[
+                reason
+                for reason in (p.get("work_reasons") or [])
+                if reason in ("assigned_to_me", "review_requested")
+            ],
+            repository=(
+                p["repository"].get("full_name", "")
+                if isinstance(p.get("repository"), dict)
+                else ""
+            ),
+            updated_at=p.get("updated_at") or "",
+            url=p["html_url"],
+        )
+        for p in (prs_data or [])[:50]
+        if isinstance(p, dict)
+        and all(
+            field in p
+            for field in ("id", "number", "title", "state", "html_url")
+        )
+    ]
+    ctx = compute(user_model, repo_models, issue_models, pr_models)
+    return JSONResponse(ctx.model_dump())
 
 
 @app.get("/api/v1/events")

tests/test_api_cache_policy.py

@@ -12,7 +12,7 @@ async def test_live_gitea_api_responses_cannot_be_stored_by_shared_caches(monkey
     async def empty_collection():
         return []
 
-    async def empty_events(user_data=None):
+    async def empty_events():
         return []
 
     monkeypatch.setattr(main, "current_user", user)
@@ -23,7 +23,7 @@ async def test_live_gitea_api_responses_cannot_be_stored_by_shared_caches(monkey
 
     transport = httpx.ASGITransport(app=main.app)
     async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
-        for path in ("/api/v1/context", "/api/v1/events", "/api/v1/live"):
+        for path in ("/api/v1/context", "/api/v1/events"):
             response = await client.get(path)
 
             assert response.status_code == 200

tests/test_api_paths.py

@@ -15,5 +15,6 @@ def test_api_requests_resolve_inside_dashboard_subpath():
         urljoin("https://forge.alexanderwhitestone.com/dashboard/", path)
         for path in api_paths
     } == {
-        "https://forge.alexanderwhitestone.com/dashboard/api/v1/live",
+        "https://forge.alexanderwhitestone.com/dashboard/api/v1/context",
+        "https://forge.alexanderwhitestone.com/dashboard/api/v1/events",
     }

tests/test_event_stream.py

@@ -17,9 +17,8 @@ def test_dashboard_has_realtime_gitea_event_stream_widget():
     assert "Gitea event stream" in html
     assert "id=\"gitea-events\"" in html
     assert "function paintEventStream" in html
-    assert "fetch('api/v1/live'" in html
+    assert "setInterval(() => { if (!document.hidden) loadEventStream(); }, 5000)" in html
-    assert "paintEventStream(snapshot.events)" in html
+    assert "if (!document.hidden) loadEventStream();" in html
-    assert "loadEventStream" not in html
     assert "event.actor?.login" in html
     assert "event.repo?.full_name" in html
 
@@ -28,6 +27,7 @@ def test_event_stream_reports_when_activity_was_refreshed():
     html = DASHBOARD.read_text()
 
     assert 'id="gitea-events-status"' in html
+    assert "setEventStreamStatus('Updating…')" in html
     assert "setEventStreamStatus('Updated ' + fmt(new Date()))" in html
 
 
@@ -121,21 +121,6 @@ async def test_activity_events_fetches_feed_for_authenticated_user(monkeypatch):
     assert paths == ["users/timmy/activities/feeds?limit=20"]
 
 
-@pytest.mark.anyio
-async def test_activity_events_reuses_supplied_authenticated_user(monkeypatch):
-    async def unexpected_current_user():
-        raise AssertionError("current user must not be fetched twice")
-
-    async def fake_fetch(path):
-        assert path == "users/timmy/activities/feeds?limit=20"
-        return []
-
-    monkeypatch.setattr(gitea_proxy, "current_user", unexpected_current_user)
-    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
-
-    assert await gitea_proxy.activity_events({"login": "timmy"}) == []
-
-
 @pytest.mark.anyio
 async def test_activity_events_skips_malformed_feed_entries(monkeypatch):
     async def fake_current_user():

tests/test_gitea_transport.py

@@ -1,43 +0,0 @@
-import httpx
-import pytest
-
-from src import gitea_proxy
-from src import main
-
-
-@pytest.mark.anyio
-async def test_gitea_transport_is_reused_across_requests_and_closed():
-    client_ids = []
-
-    async def handler(request):
-        client_ids.append(id(gitea_proxy._client))
-        return httpx.Response(200, json={"path": request.url.path})
-
-    client = gitea_proxy.start_client(transport=httpx.MockTransport(handler))
-    try:
-        first = await gitea_proxy.fetch("user")
-        second = await gitea_proxy.fetch("user/repos")
-    finally:
-        await gitea_proxy.stop_client()
-
-    assert first == {"path": "/api/v1/user"}
-    assert second == {"path": "/api/v1/user/repos"}
-    assert client_ids == [id(client), id(client)]
-    assert client.is_closed
-
-
-@pytest.mark.anyio
-async def test_application_lifespan_opens_and_closes_gitea_transport(monkeypatch):
-    calls = []
-
-    monkeypatch.setattr(gitea_proxy, "start_client", lambda: calls.append("start"))
-
-    async def stop_client():
-        calls.append("stop")
-
-    monkeypatch.setattr(gitea_proxy, "stop_client", stop_client)
-
-    async with main.app.router.lifespan_context(main.app):
-        assert calls == ["start"]
-
-    assert calls == ["start", "stop"]

tests/test_gitea_work_search.py

@@ -55,29 +55,6 @@ async def test_pull_requests_merge_assignment_and_review_responsibilities(monkey
     assert pulls[1]["work_reasons"] == ["review_requested"]
 
 
-@pytest.mark.anyio
-async def test_pull_request_searches_start_concurrently(monkeypatch):
-    started = [asyncio.Event(), asyncio.Event()]
-    release = asyncio.Event()
-
-    async def fake_fetch(path):
-        index = 0 if "assigned=true" in path else 1
-        started[index].set()
-        await release.wait()
-        return []
-
-    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)
-    task = asyncio.create_task(gitea_proxy.pull_requests())
-    try:
-        await asyncio.wait_for(
-            asyncio.gather(*(event.wait() for event in started)), timeout=1
-        )
-    finally:
-        release.set()
-
-    assert await task == []
-
-
 @pytest.mark.anyio
 async def test_requested_review_guard_uses_only_dedicated_review_search(monkeypatch):
     requested_paths = []

tests/test_live_snapshot.py

@@ -1,144 +0,0 @@
-import asyncio
-import json
-
-import pytest
-
-from src import main
-
-
-@pytest.fixture(autouse=True)
-def reset_live_snapshot_task():
-    main._live_snapshot_task = None
-    yield
-    main._live_snapshot_task = None
-
-
-def payload(response):
-    return json.loads(response.body)
-
-
-@pytest.mark.anyio
-async def test_live_snapshot_fetches_user_once_and_updates_work_and_activity(monkeypatch):
-    calls = {"user": 0}
-
-    async def user():
-        calls["user"] += 1
-        return {"id": 1, "login": "timmy"}
-
-    async def empty():
-        return []
-
-    async def events(authenticated_user):
-        assert authenticated_user["login"] == "timmy"
-        return [{"type": "push"}]
-
-    monkeypatch.setattr(main, "current_user", user)
-    monkeypatch.setattr(main, "repos", empty)
-    monkeypatch.setattr(main, "issues", empty)
-    monkeypatch.setattr(main, "pull_requests", empty)
-    monkeypatch.setattr(main, "activity_events", events)
-
-    response = await main.live_snapshot()
-    result = payload(response)
-
-    assert calls["user"] == 1
-    assert result["context"]["user"]["login"] == "timmy"
-    assert result["events"] == [{"type": "push"}]
-    assert result["sections"] == {"context": "fresh", "events": "fresh"}
-
-
-@pytest.mark.anyio
-async def test_live_snapshot_keeps_fresh_context_when_activity_fails(monkeypatch):
-    async def user():
-        return {"id": 1, "login": "timmy"}
-
-    async def empty():
-        return []
-
-    async def failing_events(authenticated_user):
-        raise ConnectionError("secret upstream detail")
-
-    monkeypatch.setattr(main, "current_user", user)
-    monkeypatch.setattr(main, "repos", empty)
-    monkeypatch.setattr(main, "issues", empty)
-    monkeypatch.setattr(main, "pull_requests", empty)
-    monkeypatch.setattr(main, "activity_events", failing_events)
-
-    result = payload(await main.live_snapshot())
-
-    assert result["context"]["user"]["login"] == "timmy"
-    assert result["events"] is None
-    assert result["sections"] == {
-        "context": "fresh",
-        "events": "temporarily unavailable",
-    }
-    assert "secret" not in json.dumps(result)
-
-
-@pytest.mark.anyio
-async def test_live_snapshot_keeps_fresh_activity_when_work_fails(monkeypatch):
-    async def user():
-        return {"id": 1, "login": "timmy"}
-
-    async def failing_repos():
-        raise ConnectionError("work unavailable")
-
-    async def empty():
-        return []
-
-    async def events(authenticated_user):
-        return [{"type": "push"}]
-
-    monkeypatch.setattr(main, "current_user", user)
-    monkeypatch.setattr(main, "repos", failing_repos)
-    monkeypatch.setattr(main, "issues", empty)
-    monkeypatch.setattr(main, "pull_requests", empty)
-    monkeypatch.setattr(main, "activity_events", events)
-
-    result = payload(await main.live_snapshot())
-
-    assert result["context"] is None
-    assert result["events"] == [{"type": "push"}]
-    assert result["sections"] == {
-        "context": "temporarily unavailable",
-        "events": "fresh",
-    }
-
-
-@pytest.mark.anyio
-async def test_live_snapshot_coalesces_only_simultaneous_requests(monkeypatch):
-    user_calls = 0
-    release = asyncio.Event()
-
-    async def user():
-        nonlocal user_calls
-        user_calls += 1
-        return {"id": 1, "login": "timmy"}
-
-    async def blocked_repos():
-        await release.wait()
-        return []
-
-    async def empty():
-        return []
-
-    async def events(authenticated_user):
-        return []
-
-    monkeypatch.setattr(main, "current_user", user)
-    monkeypatch.setattr(main, "repos", blocked_repos)
-    monkeypatch.setattr(main, "issues", empty)
-    monkeypatch.setattr(main, "pull_requests", empty)
-    monkeypatch.setattr(main, "activity_events", events)
-
-    first = asyncio.create_task(main.live_snapshot())
-    await asyncio.sleep(0)
-    second = asyncio.create_task(main.live_snapshot())
-    await asyncio.sleep(0)
-    assert user_calls == 1
-
-    release.set()
-    await asyncio.gather(first, second)
-    await main.live_snapshot()
-
-    assert user_calls == 2

tests/test_shared_context_snapshot.py

@@ -15,15 +15,3 @@ async def test_one_context_snapshot_updates_every_context_backed_panel():
     assert "updateRepoMix(qs('#repo-mix'), fetch)" not in html
     assert "setInterval(load, 8000)" not in html
     assert "document.addEventListener('visibilitychange'" in html
-
-
-@pytest.mark.anyio
-async def test_one_live_snapshot_updates_work_and_activity_on_one_timer():
-    html = await dashboard()
-
-    assert "fetch('api/v1/live'" in html
-    assert "renderLiveSnapshot(snapshot)" in html
-    assert "renderContextSnapshot(snapshot.context)" in html
-    assert "paintEventStream(snapshot.events)" in html
-    assert "loadEventStream" not in html
-    assert "5000" not in html