fix(workspace): reconcile listed repos with the configured (postgres) DB

packages/server/src/repowise/server/routers/workspace.py

@@ -12,8 +12,12 @@
 from pathlib import Path
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from sqlalchemy import func, select
 
 from repowise.core.docs_mode import resolve_docs_mode
+from repowise.core.persistence import crud
+from repowise.core.persistence.database import get_session
+from repowise.core.persistence.models import GitMetadata, GraphNode, HealthFileMetric, Page
 from repowise.server.deps import (
     get_cross_repo_enricher,
     get_workspace_config,
@@ -58,17 +62,46 @@
         raise HTTPException(status_code=404, detail="Not running in workspace mode")
 
 
-def _query_top_language(db_path: Path) -> str:
-    """Return the most common language across graph_nodes in a repo's wiki.db."""
-    if not db_path.exists():
+async def _query_top_language(
+    db_path: Path, session_factory=None, repo_path: Path | None = None
+) -> str:
+    """Return the most common language across graph_nodes for a repo.
+
+    Reads the repo-local ``wiki.db`` (SQLite) when present, otherwise falls
+    back to the configured database (e.g. PostgreSQL) keyed by ``local_path``.
+    """
+    if db_path and Path(db_path).exists():
+        try:
+            with sqlite3.connect(str(db_path)) as conn:
+                row = conn.execute(
+                    "SELECT language, COUNT(*) AS cnt FROM graph_nodes "
+                    "WHERE node_type = 'file' AND language IS NOT NULL AND language != '' "
+                    "GROUP BY language ORDER BY cnt DESC LIMIT 1"
+                ).fetchone()
+                return row[0] if row else "unknown"
+        except Exception:
+            return "unknown"
+
+    if session_factory is None or repo_path is None:
         return "unknown"
     try:
-        with sqlite3.connect(str(db_path)) as conn:
-            row = conn.execute(
-                "SELECT language, COUNT(*) AS cnt FROM graph_nodes "
-                "WHERE node_type = 'file' AND language IS NOT NULL AND language != '' "
-                "GROUP BY language ORDER BY cnt DESC LIMIT 1"
-            ).fetchone()
+        async with get_session(session_factory) as session:
+            repo = await crud.get_repository_by_path(session, str(Path(repo_path).resolve()))
+            if repo is None:
+                return "unknown"
+            result = await session.execute(
+                select(GraphNode.language, func.count(GraphNode.id))
+                .where(
+                    GraphNode.repository_id == repo.id,
+                    GraphNode.node_type == "file",
+                    GraphNode.language.is_not(None),
+                    GraphNode.language != "",
+                )
+                .group_by(GraphNode.language)
+                .order_by(func.count(GraphNode.id).desc())
+                .limit(1)
+            )
+            row = result.first()
             return row[0] if row else "unknown"
     except Exception:
         return "unknown"
@@ -100,8 +133,17 @@
     )
 
 
-def _query_repo_stats(db_path: Path) -> dict:
-    """Query basic stats from a repo's wiki.db using raw sqlite3.
+async def _query_repo_stats(
+    db_path: Path,
+    session_factory=None,
+    repo_path: Path | None = None,
+) -> dict:
+    """Query basic stats for a workspace repo.
+
+    Reads the repo-local ``wiki.db`` (SQLite) when present. When the repo is
+    indexed into the configured database instead — e.g. PostgreSQL, where there
+    is no per-repo ``.repowise/wiki.db`` file — falls back to querying the
+    configured session factory keyed by the repo's absolute ``local_path``.
 
     Returns a dict with repo_id, file_count, symbol_count, page_count,
     doc_coverage_pct, hotspot_count, health_score, status, docs_enabled, and
@@ -122,22 +164,36 @@
         "docs_enabled": False,
         "docs_skip_reason": None,
     }
+
+    repo_root = Path(repo_path).resolve() if repo_path else None
     # Surface docs lifecycle from state.json so the UI shows a coherent
     # picture even when only indexing (no docs) ran.
     try:
         import json as _json
 
-        state_path = db_path.parent / "state.json"
+        state_path = (db_path.parent if db_path else Path(repo_root or ".")) / "state.json"
         if state_path.is_file():
             state = _json.loads(state_path.read_text(encoding="utf-8"))
             result["docs_enabled"] = resolve_docs_mode(state) != "none"
             result["docs_skip_reason"] = state.get("docs_skip_reason")
     except Exception:
         pass
-    if not db_path.exists():
-        if not db_path.parent.parent.is_dir():
+
+    # Prefer the repo-local SQLite store; fall back to the configured DB.
+    if db_path and db_path.exists():
+        _query_repo_stats_from_sqlite(db_path, result)
+    elif session_factory is not None and repo_root is not None:
+        if not repo_root.is_dir():
             result["status"] = "missing_dir"
-        return result
+            return result
+        await _query_repo_stats_from_db(session_factory, str(repo_root), result)
+    elif db_path and not db_path.parent.parent.is_dir():
+        result["status"] = "missing_dir"
+    return result
+
+
+def _query_repo_stats_from_sqlite(db_path: Path, result: dict) -> None:
+    """Populate *result* from a repo-local SQLite wiki.db."""
     result["status"] = "indexed"
     try:
         conn = sqlite3.connect(str(db_path))
@@ -148,7 +204,7 @@
         if row:
             result["repo_id"] = row[0]
 
-        # file count (graph_nodes) 
+        # file count (graph_nodes)
         row = c.execute("SELECT COUNT(*) FROM graph_nodes WHERE node_type = 'file'").fetchone()
         result["file_count"] = row[0] if row else 0
 
@@ -165,9 +221,7 @@
         result["doc_coverage_pct"] = round(float(row[0] or 0.0) * 100, 1)
 
         # hotspot count — use the canonical is_hotspot flag, matching the rest
-        # of the codebase (module_health, extract-demo-data). The earlier
-        # churn_percentile >= 90 predicate never matched: churn_percentile is
-        # stored on a 0.0-1.0 scale and only scaled to 0-100 at the API layer.
+        # of the codebase (module_health, extract-demo-data).
         try:
             row = c.execute("SELECT COUNT(*) FROM git_metadata WHERE is_hotspot = 1").fetchone()
             result["hotspot_count"] = row[0] if row else 0
@@ -181,7 +235,89 @@
     # Read once here rather than per-endpoint: both the workspace listing and
     # the graph need the canonical score, and this used to be a second sweep.
     result["health_score"] = read_repo_health_score(db_path)
-    return result
+
+
+async def _query_repo_stats_from_db(session_factory, local_path: str, result: dict) -> None:
+    """Populate *result* from the configured (e.g. PostgreSQL) database.
+
+    The repo row is keyed by ``local_path`` in the primary DB; counts are
+    aggregated from the same tables the SQLite path reads.
+    """
+    try:
+        async with get_session(session_factory) as session:
+            repo = await crud.get_repository_by_path(session, local_path)
+            if repo is None:
+                return
+            result["repo_id"] = repo.id
+            result["status"] = "indexed"
+
+            # file count (graph_nodes)
+            result["file_count"] = (
+                await session.execute(
+                    select(func.count(GraphNode.id)).where(
+                        GraphNode.repository_id == repo.id,
+                        GraphNode.node_type == "file",
+                    )
+                )
+            ).scalar_one() or 0
+
+            # symbol count
+            result["symbol_count"] = (
+                await session.execute(
+                    select(func.coalesce(func.sum(GraphNode.symbol_count), 0)).where(
+                        GraphNode.repository_id == repo.id
+                    )
+                )
+            ).scalar_one() or 0
+
+            # page count
+            result["page_count"] = (
+                await session.execute(
+                    select(func.count(Page.id)).where(Page.repository_id == repo.id)
+                )
+            ).scalar_one() or 0
+
+            # doc coverage (avg confidence * 100)
+            avg_conf = (
+                await session.execute(
+                    select(func.avg(Page.confidence)).where(Page.repository_id == repo.id)
+                )
+            ).scalar_one()
+            result["doc_coverage_pct"] = round(float(avg_conf or 0.0) * 100, 1)
+
+            # hotspot count — canonical is_hotspot flag.
+            try:
+                result["hotspot_count"] = (
+                    await session.execute(
+                        select(func.count(GitMetadata.id)).where(
+                            GitMetadata.repository_id == repo.id,
+                            GitMetadata.is_hotspot.is_(True),
+                        )
+                    )
+                ).scalar_one() or 0
+            except Exception:
+                result["hotspot_count"] = 0  # table or column may not exist
+    except Exception:
+        _log.debug("Failed to query stats from db for %s", local_path, exc_info=True)
+        return
+
+    # Canonical health score from the health_file_metrics table, matching
+    # read_repo_health_score's NLOC-weighted average.
+    try:
+        async with get_session(session_factory) as session:
+            row = (
+                await session.execute(
+                    select(
+                        func.sum(HealthFileMetric.score * func.max(HealthFileMetric.nloc, 1)),
+                        func.sum(func.max(HealthFileMetric.nloc, 1)),
+                    ).where(HealthFileMetric.repository_id == result["repo_id"])
+                )
+            ).first()
+            if row and row[1]:
+                avg = float(row[0]) / float(row[1])
+                result["health_score"] = max(0.0, min(100.0, round(avg * 10.0, 1)))
+    except Exception:
+        result["health_score"] = None
 
 
 # ---------------------------------------------------------------------------
@@ -205,14 +341,15 @@
 
     ws_root = getattr(request.app.state, "workspace_root", None)
     ws_root_path = Path(ws_root) if ws_root else None
+    session_factory = getattr(request.app.state, "session_factory", None)
 
     repo_entries = []
     for r in ws_config.repos:
         stats: dict = {}
         if ws_root_path:
             repo_path = (ws_root_path / r.path).resolve()
             db_path = repo_path / ".repowise" / "wiki.db"
-            stats = _query_repo_stats(db_path)
+            stats = await _query_repo_stats(db_path, session_factory, repo_path)
         repo_entries.append(
             WorkspaceRepoEntry(
                 alias=r.alias,
@@ -254,7 +391,9 @@
 async def get_contracts(
     ws_config=Depends(get_workspace_config),
     enricher=Depends(get_cross_repo_enricher),
-    contract_type: str | None = Query(None, description="Filter: http, grpc, socket, topic, or data"),
+    contract_type: str | None = Query(
+        None, description="Filter: http, grpc, socket, topic, or data"
+    ),
     repo: str | None = Query(None, description="Filter by repo alias"),
     role: str | None = Query(None, description="Filter: provider or consumer"),
     limit: int = Query(200, ge=1, le=1000),
@@ -394,104 +533,105 @@
 
 
 @router.get("/graph", response_model=WorkspaceGraphResponse)
 async def get_workspace_graph(
     request: Request,
     ws_config=Depends(get_workspace_config),
     enricher=Depends(get_cross_repo_enricher),
 ):
     """Cross-repo graph: repos as mega-nodes, contracts/co-changes as edges."""
     _require_workspace(ws_config)
 
     ws_root = getattr(request.app.state, "workspace_root", None)
     ws_root_path = Path(ws_root) if ws_root else None
+    session_factory = getattr(request.app.state, "session_factory", None)
 
     # Build nodes from repo metadata
     repo_id_map: dict[str, str] = {}  # alias → repo_id
     nodes: list[WorkspaceGraphNode] = []
     for r in ws_config.repos:
         stats: dict = {}
         top_language = "unknown"
         if ws_root_path:
             repo_path = (ws_root_path / r.path).resolve()
             db_path = repo_path / ".repowise" / "wiki.db"
-            stats = _query_repo_stats(db_path)
-            top_language = _query_top_language(db_path)
+            stats = await _query_repo_stats(db_path, session_factory, repo_path)
+            top_language = await _query_top_language(db_path, session_factory, repo_path)
             health_score, health_score_source = _resolve_graph_health_score(stats)
         else:
             health_score = 0.0
             health_score_source = "derived"
         rid = stats.get("repo_id") or r.alias
         repo_id_map[r.alias] = rid
         nodes.append(
             WorkspaceGraphNode(
                 repo_id=rid,
                 name=r.alias,
                 file_count=stats.get("file_count", 0),
                 coverage_pct=stats.get("doc_coverage_pct", 0.0),
                 health_score=health_score,
                 health_score_source=health_score_source,
                 top_language=top_language,
             )
         )
 
     # Build edges
     edges: list[WorkspaceGraphEdge] = []
     seen_edges: set[tuple[str, str, str]] = set()
 
     if enricher is not None:
         # Contract-based edges: each link connects two repos
         links = list(getattr(enricher, "_contract_links", []))
         for lk in links:
             p_repo = lk.get("provider_repo", "")
             c_repo = lk.get("consumer_repo", "")
             if not p_repo or not c_repo or p_repo == c_repo:
                 continue
             p_id = repo_id_map.get(p_repo, p_repo)
             c_id = repo_id_map.get(c_repo, c_repo)
             key = (min(p_id, c_id), max(p_id, c_id), "contract")
             if key in seen_edges:
                 continue
             seen_edges.add(key)
             edges.append(
                 WorkspaceGraphEdge(
                     source=p_id,
                     target=c_id,
                     type="contract",
                     strength=lk.get("confidence", 0.8),
                     label=lk.get("contract_type"),
                 )
             )
 
         # Co-change-based edges: aggregate per repo-pair
         co_changes = list(getattr(enricher, "_co_changes", []))
         pair_strengths: dict[tuple[str, str], list[float]] = {}
         for cc in co_changes:
             s_repo = cc.get("source_repo", "")
             t_repo = cc.get("target_repo", "")
             if not s_repo or not t_repo or s_repo == t_repo:
                 continue
             s_id = repo_id_map.get(s_repo, s_repo)
             t_id = repo_id_map.get(t_repo, t_repo)
             pair_key = (min(s_id, t_id), max(s_id, t_id))
             pair_strengths.setdefault(pair_key, []).append(cc.get("strength", 0.0))
 
         for (src, tgt), strengths in pair_strengths.items():
             key = (src, tgt, "co_change")
             if key in seen_edges:
                 continue
             seen_edges.add(key)
             avg_strength = sum(strengths) / len(strengths)
             edges.append(
                 WorkspaceGraphEdge(
                     source=src,
                     target=tgt,
                     type="co_change",
                     strength=round(avg_strength, 3),
                     label=f"{len(strengths)} co-changes",
                 )
             )
 
     return WorkspaceGraphResponse(nodes=nodes, edges=edges)
 
 
 # ---------------------------------------------------------------------------
@@ -748,178 +888,198 @@
 
 
 @router.post("/sync", status_code=202)
 async def sync_workspace(
     request: Request,
     repo_alias: str | None = Query(
         None,
         description="If set, only sync this repo alias (still fans through the job system).",
     ),
     full_resync: bool = Query(False, description="Trigger a full resync instead of incremental."),
     ws_config=Depends(get_workspace_config),
 ):
     """Fan out a sync to every (stale or unindexed) repo in the workspace.
 
     Returns one :class:`WorkspaceSyncResult` per attempted repo with the
     decision (accepted / skipped / error) so the web UI can render
     progress without polling for each repo individually.
 
     Uses the same job machinery as ``POST /api/repos/{id}/sync`` so the
     scheduler, cost ledger, and live-progress hooks all work without
     special cases.
     """
     from repowise.server.schemas import (
         WorkspaceSyncResponse,
         WorkspaceSyncResult,
     )
 
     _require_workspace(ws_config)
 
     from sqlalchemy import select
 
     from repowise.core.persistence import crud
     from repowise.core.persistence.database import get_session
     from repowise.core.persistence.models import GenerationJob
     from repowise.server.routers.repos import _launch_job_task
 
     ws_root = getattr(request.app.state, "workspace_root", None)
     if ws_root is None:
         raise HTTPException(status_code=500, detail="Workspace root missing on app state")
     ws_root_path = Path(ws_root)
 
     results: list[WorkspaceSyncResult] = []
 
     # Resolve aliases → entries to operate on.
     if repo_alias is not None:
         entry = ws_config.get_repo(repo_alias)
         if entry is None:
             raise HTTPException(
                 status_code=404,
                 detail=f"Unknown repo alias '{repo_alias}' in workspace.",
             )
         entries = [entry]
     else:
         entries = list(ws_config.repos)
 
     for entry in entries:
         repo_path = (ws_root_path / entry.path).resolve()
         db_path = repo_path / ".repowise" / "wiki.db"
 
-        # Not indexed yet → establish the repo-local database and run the
+        # A repo is "indexed" when a Repository row exists for its resolved
+        # path in the configured database — not when a local .repowise/wiki.db
+        # file happens to exist. Under PostgreSQL (REPOWISE_DB_URL) repos are
+        # indexed into the shared DB with no per-repo wiki.db, so a file check
+        # alone would wrongly treat every Postgres-indexed repo as "not indexed"
+        # and re-index it on every sync (issue #1034).
+        already_indexed = False
+        if db_path.exists():
+            already_indexed = True
+        else:
+            try:
+                async with get_session(request.app.state.session_factory) as session:
+                    already_indexed = (
+                        await crud.get_repository_by_path(
+                            session, str(repo_path.resolve())
+                        )
+                    ) is not None
+            except Exception:
+                already_indexed = False
+
+        # Not indexed yet → establish the repository database and run the
         # first full index through the same job machinery, instead of
         # bouncing the user to the CLI.
-        if not db_path.exists():
+        if not already_indexed:
             try:
                 from repowise.server.repo_db import ensure_repo_registration
                 from repowise.server.routers.repos import _enqueue_index_job
 
                 factory, new_repo_id = await ensure_repo_registration(
                     request.app.state,
                     local_path=str(repo_path),
                     name=entry.alias,
                 )
                 job_id = await _enqueue_index_job(request, factory, new_repo_id)
             except Exception as exc:
                 results.append(
                     WorkspaceSyncResult(
                         alias=entry.alias,
                         status="error",
                         reason=f"first-time index failed to start: {exc}",
                     )
                 )
                 continue
             results.append(
                 WorkspaceSyncResult(
                     alias=entry.alias,
                     repo_id=new_repo_id,
                     status="accepted" if job_id else "skipped",
                     reason=None if job_id else "a job is already running for this repo",
                     job_id=job_id,
                 )
             )
             continue
 
         # Discover repo_id from the per-repo DB.
         try:
             with sqlite3.connect(str(db_path)) as conn:
                 row = conn.execute("SELECT id FROM repositories LIMIT 1").fetchone()
         except Exception as exc:
             results.append(
                 WorkspaceSyncResult(
                     alias=entry.alias,
                     status="error",
                     reason=f"could not read repo id: {exc}",
                 )
             )
             continue
 
         if not row:
             results.append(
                 WorkspaceSyncResult(
                     alias=entry.alias,
                     status="error",
                     reason="repository row missing in wiki.db",
                 )
             )
             continue
         repo_id = row[0]
 
         session_factory = resolve_session_factory(request.app.state, repo_id)
 
         async def _create_job(
             *,
             session_factory=session_factory,
             repo_id: str | None = repo_id,
         ) -> tuple[str | None, str | None]:
             """Create a pending job row, returning (job_id, error)."""
             try:
                 async with get_session(session_factory) as session:
                     # Prevent concurrent runs on the same repo.
                     active = await session.execute(
                         select(GenerationJob.id)
                         .where(GenerationJob.repository_id == repo_id)
                         .where(GenerationJob.status.in_(["pending", "running"]))
                         .limit(1)
                     )
                     if active.scalar_one_or_none() is not None:
                         return None, "a sync is already running for this repo"
 
                     config_data = {"mode": "full_resync"} if full_resync else None
                     job = await crud.upsert_generation_job(
                         session,
                         repository_id=repo_id,
                         status="pending",
                         config=config_data,
                     )
                     await session.commit()
                     return job.id, None
             except Exception as exc:
                 return None, str(exc)
 
         job_id, err = await _create_job()
         if err is not None:
             results.append(
                 WorkspaceSyncResult(
                     alias=entry.alias,
                     repo_id=repo_id,
                     status="skipped" if "already running" in err else "error",
                     reason=err,
                 )
             )
             continue
 
         _launch_job_task(request, job_id, repo_id)
         results.append(
             WorkspaceSyncResult(
                 alias=entry.alias,
                 repo_id=repo_id,
                 job_id=job_id,
                 status="accepted",
             )
         )
 
     return WorkspaceSyncResponse(
         results=results,
         accepted=sum(1 for r in results if r.status == "accepted"),
         skipped=sum(1 for r in results if r.status == "skipped"),
         errors=sum(1 for r in results if r.status == "error"),
     )