Make indexing progress observable and stuck indexes recoverable

apps/api/requirements.txt

@@ -66,5 +66,7 @@ pyyaml>=6.0
 # Testing
 pytest>=8.0.0
 pytest-asyncio>=0.23.0
+# Exercises the Redis code paths (streams, consumer groups, XAUTOCLAIM) without a server.
+fakeredis>=2.20.0
 # pyproject.toml's addopts hardcodes --cov, so pytest cannot start without this.
 pytest-cov>=4.1.0

apps/api/src/api/routes/repos.py

@@ -6,7 +6,7 @@
 import json
 import logging
 
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from sqlalchemy.orm import Session
 
@@ -25,6 +25,9 @@
 router = APIRouter()
 logger = logging.getLogger(__name__)
 
+# Ceiling on a single SSE progress subscription; clients reconnect if they still care.
+_PROGRESS_STREAM_TIMEOUT_SECONDS = 15 * 60
+
 
 def run_indexing_task(repo_id: str):
     """Run async indexing in a new event loop with fresh DB session."""
@@ -151,6 +154,7 @@ async def get_repository(
 @router.get("/{repo_id}/progress")
 async def get_indexing_progress(
     repo_id: str,
+    request: Request,
     db: Session = Depends(get_db),
 ):
     """Stream indexing progress updates via SSE."""
@@ -162,14 +166,37 @@ async def get_indexing_progress(
         raise HTTPException(status_code=404, detail="Repository not found")
 
     async def progress_stream():
-        """Generate SSE events for progress updates."""
+        """
+        SSE progress events.
+
+        Bounded and disconnect-aware: the previous version looped forever with no client
+        check and no ceiling, so a repository stuck mid-index held an open response (and
+        its database session) indefinitely, once per browser tab.
+        """
         indexing_service = IndexingService(db)
+        deadline = asyncio.get_event_loop().time() + _PROGRESS_STREAM_TIMEOUT_SECONDS
+        last_payload = None
 
         while True:
+            if await request.is_disconnected():
+                break
+
             progress = await indexing_service.get_progress(repo_id)
-            yield f"data: {json.dumps(progress)}\n\n"
 
-            if progress["status"] in ["completed", "failed"]:
+            # Only emit on change; a 1s poll of an unchanged value is pure noise.
+            payload = json.dumps(progress, sort_keys=True)
+            if payload != last_payload:
+                last_payload = payload
+                yield f"data: {payload}\n\n"
+
+            if progress.get("status") in ("completed", "failed"):
+                break
+
+            if asyncio.get_event_loop().time() > deadline:
+                yield (
+                    'data: {"type": "timeout", "message": '
+                    '"Progress stream timed out; reconnect to continue watching."}\n\n'
+                )
                 break
 
             await asyncio.sleep(1)

apps/api/src/core/progress.py

@@ -0,0 +1,159 @@
+"""
+Shared indexing-progress store.
+
+WHY THIS EXISTS
+IndexingService wrote progress into a per-instance dict. The SSE endpoint constructs a
+fresh IndexingService per request, so it never saw those writes and always fell through to
+the database branch, which hardcodes current_step="Unknown" and a percentage of 0 or 100.
+The per-file progress the indexer computes was therefore unreachable by any client, and
+the progress bar could only ever show 0% or 100%.
+
+A per-process dict cannot fix this either: indexing runs in a background task and, under
+more than one worker, in a different process from the request that wants to read it.
+
+DESIGN
+A Redis stream per repository (progress:{repo_id}), capped with MAXLEN so it cannot grow
+without bound. Redis is optional throughout this codebase, so there is an in-memory
+fallback that is correct within a single process -- which is the local-dev shape. The
+fallback is explicitly not cross-process, and callers can tell which backend is live via
+`backend`.
+
+Streams rather than a plain key because the SSE endpoint wants *history*: a client that
+connects late should be able to replay what it missed rather than only seeing the current
+value, and XRANGE gives that for free.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from collections import defaultdict, deque
+from threading import Lock
+from typing import Any, Deque, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+# Enough to replay a long index without unbounded growth. Each entry is small.
+_STREAM_MAXLEN = 500
+_MEMORY_MAXLEN = 500
+
+
+def _stream_key(repo_id: str) -> str:
+    return f"progress:{repo_id}"
+
+
+class ProgressStore:
+    """
+    Publishes and reads indexing progress.
+
+    Safe to construct per request; the Redis client it wraps is the shared singleton.
+    """
+
+    def __init__(self, redis_client=None):
+        self._redis = redis_client
+        # Class-level so every instance in a process shares the fallback, which is the
+        # whole point -- a per-instance dict is the bug this replaces.
+        self._lock = _MEMORY_LOCK
+        self._memory = _MEMORY
+
+    @property
+    def backend(self) -> str:
+        return "redis" if self._redis is not None else "memory"
+
+    async def publish(
+        self,
+        repo_id: str,
+        status: str,
+        step: str,
+        percent: float,
+        files_processed: int = 0,
+        total_files: int = 0,
+    ) -> None:
+        """Record a progress event. Never raises -- progress must not break indexing."""
+        event = {
+            "repo_id": repo_id,
+            "status": status,
+            "current_step": step,
+            "progress_percent": round(float(percent), 2),
+            "files_processed": int(files_processed),
+            "total_files": int(total_files),
+            "at": time.time(),
+        }
+
+        if self._redis is not None:
+            try:
+                await self._redis.xadd(
+                    _stream_key(repo_id),
+                    {"event": json.dumps(event)},
+                    maxlen=_STREAM_MAXLEN,
+                    approximate=True,
+                )
+                return
+            except Exception as exc:
+                # Fall through to memory rather than losing the event or failing the
+                # index. Logged at debug because a flapping Redis would otherwise emit
+                # one warning per parsed file.
+                logger.debug("Progress publish to Redis failed, using memory: %s", exc)
+
+        with self._lock:
+            self._memory[repo_id].append(event)
+
+    async def latest(self, repo_id: str) -> Optional[Dict[str, Any]]:
+        """Most recent event, or None if nothing has been published."""
+        if self._redis is not None:
+            try:
+                entries = await self._redis.xrevrange(_stream_key(repo_id), count=1)
+                if entries:
+                    return _decode(entries[0])
+            except Exception as exc:
+                logger.debug("Progress read from Redis failed, using memory: %s", exc)
+
+        with self._lock:
+            events = self._memory.get(repo_id)
+            return dict(events[-1]) if events else None
+
+    async def history(self, repo_id: str, limit: int = 100) -> List[Dict[str, Any]]:
+        """Oldest-first events, so a late subscriber can replay what it missed."""
+        if self._redis is not None:
+            try:
+                entries = await self._redis.xrange(_stream_key(repo_id), count=limit)
+                return [_decode(e) for e in entries]
+            except Exception as exc:
+                logger.debug("Progress history from Redis failed, using memory: %s", exc)
+
+        with self._lock:
+            events = self._memory.get(repo_id) or []
+            return [dict(e) for e in list(events)[-limit:]]
+
+    async def clear(self, repo_id: str) -> None:
+        """Drop a repository's progress, e.g. when it is deleted or re-indexed."""
+        if self._redis is not None:
+            try:
+                await self._redis.delete(_stream_key(repo_id))
+            except Exception as exc:
+                logger.debug("Progress clear in Redis failed: %s", exc)
+
+        with self._lock:
+            self._memory.pop(repo_id, None)
+
+
+def _decode(entry) -> Dict[str, Any]:
+    """
+    Turn one stream entry into the event dict.
+
+    Redis returns (id, {field: value}); values are bytes unless the client decodes
+    responses, and this codebase does not configure that, so both are handled.
+    """
+    _entry_id, fields = entry
+    raw = fields.get(b"event") or fields.get("event") or "{}"
+    if isinstance(raw, bytes):
+        raw = raw.decode("utf-8", errors="replace")
+    try:
+        return json.loads(raw)
+    except json.JSONDecodeError:
+        return {}
+
+
+_MEMORY: Dict[str, Deque[Dict[str, Any]]] = defaultdict(lambda: deque(maxlen=_MEMORY_MAXLEN))
+_MEMORY_LOCK = Lock()

apps/api/src/dependencies.py

@@ -159,3 +159,10 @@ def get_graph_store():
     from src.core.graph.neo4j_store import Neo4jGraphStore
 
     return Neo4jGraphStore(driver, database=settings.neo4j_database)
+
+
+def get_progress_store():
+    """Shared indexing-progress store (Redis stream, memory fallback)."""
+    from src.core.progress import ProgressStore
+
+    return ProgressStore(redis_client=get_redis_client())

apps/api/src/main.py

@@ -21,7 +21,7 @@
     get_session_factory,
     get_vector_store,
 )
-from src.models.database import init_db
+from src.models.database import init_db, reap_stuck_indexing
 from src.models.migrations import run_pending_migrations
 
 # Configure logging
@@ -53,6 +53,8 @@ async def lifespan(app: FastAPI):
     engine = get_db_engine()
     init_db(engine)
     run_pending_migrations(engine)
+    # Anything still mid-index belongs to a process that no longer exists.
+    reap_stuck_indexing(engine)
     logger.info("Database initialized")
 
     # Initialize vector store

apps/api/src/models/database.py

@@ -4,6 +4,7 @@
 """
 
 import enum
+import logging
 import uuid
 from datetime import datetime, timezone
 
@@ -22,6 +23,8 @@
 from sqlalchemy import Enum as SQLEnum
 from sqlalchemy.orm import declarative_base, relationship
 
+logger = logging.getLogger(__name__)
+
 Base = declarative_base()
 
 
@@ -412,6 +415,45 @@ class GraphNodeInteraction(Base):
     )
 
 
+def reap_stuck_indexing(engine) -> int:
+    """
+    Fail repositories left mid-index by a process that died.
+
+    Only FAILED and COMPLETED are terminal. A container killed during CLONING, PARSING or
+    EMBEDDING left the row in that state forever, and the only self-healing path in
+    repos.py and seed_demo.py triggers on FAILED -- so a stuck repository could never be
+    retried through the API, and in demo mode that bricked the deployment.
+
+    Runs at startup, where anything still in a transient state provably has no live
+    indexer, because the process that owned it is gone.
+    """
+    from sqlalchemy import text as _text
+
+    transient = ("cloning", "parsing", "embedding")
+    with engine.begin() as connection:
+        placeholders = ", ".join(f":s{i}" for i in range(len(transient)))
+        params = {f"s{i}": value for i, value in enumerate(transient)}
+        # SQLEnum persists the member NAME, so compare case-insensitively to be safe
+        # across both spellings.
+        result = connection.execute(
+            _text(
+                "UPDATE repositories SET status = 'FAILED', "
+                "indexing_error = COALESCE(indexing_error, "
+                "'Indexing was interrupted before completing. Re-import to retry.') "
+                f"WHERE LOWER(status) IN ({placeholders})"
+            ),
+            params,
+        )
+        reaped = result.rowcount or 0
+
+    if reaped:
+        logger.warning(
+            "Marked %d repository(ies) as failed: they were mid-index when the previous "
+            "process exited. They can now be re-indexed.", reaped
+        )
+    return reaped
+
+
 def init_db(engine):
     """Create all tables."""
     Base.metadata.create_all(bind=engine)