fix(workspace): route Postgres-backed watch/update through the shared DB instead of local wiki.db

packages/core/src/repowise/core/persistence/database.py

@@ -229,7 +229,7 @@
     return _default_db_url(repo_path)
 
 
 def create_engine(
     url: str | None = None,
     *,
     echo: bool = False,
@@ -237,6 +237,7 @@
     # Pass use_static_pool=True explicitly when creating in-memory test engines.
     use_static_pool: bool = False,
     busy_timeout_ms: int | None = None,
+    short_lived: bool = True,
 ) -> AsyncEngine:
     """Create an AsyncEngine for the given database URL.
 
@@ -249,6 +250,25 @@
                          small value for best-effort secondary writers that must
                          never stall the primary writer (issue #326). Ignored
                          for non-SQLite backends.
+        short_lived:     Whether this engine is created, used, and disposed
+                         within a single call (the pattern almost every caller
+                         follows: one CLI command, one workspace update, one
+                         background task). Defaults to True, which uses
+                         NullPool for PostgreSQL — one connection per checkout,
+                         closed on dispose, so a short-lived engine can never
+                         hold more than a single Postgres server slot, and an
+                         engine that outlives its creating event loop can never
+                         hand back a dead pooled connection to a later one
+                         (issue #2062's failure class). Pass False only for an
+                         engine stored for a process's lifetime and reused
+                         across many requests — currently just the FastAPI app
+                         and the MCP server — where SQLAlchemy's pooled
+                         AsyncAdaptedQueuePool is the correct choice and
+                         NullPool would open a fresh connection per request.
+                         Ignored for SQLite, which already always uses
+                         NullPool (or StaticPool for :memory:) regardless of
+                         this flag — SQLite has no equivalent long-lived-pool
+                         need since ``aiosqlite`` connections are cheap.
     """
     db_url = get_db_url(url)
     is_sqlite = db_url.startswith("sqlite")
@@ -265,8 +285,21 @@
         else:
             kwargs["poolclass"] = NullPool
     else:
-        # PostgreSQL — asyncpg handles its own connection pool
+        # PostgreSQL. SQLAlchemy pools these connections with
+        # AsyncAdaptedQueuePool by default — asyncpg does NOT provide its own
+        # pool here (that only happens if something calls asyncpg.create_pool,
+        # which nothing in this codebase does). Every create_engine() call in
+        # this codebase except the long-lived server/MCP engines is
+        # short-lived (create, use, dispose within one async function), so
+        # there's no reuse to gain from pooling and every pooled-but-idle
+        # connection is a Postgres server slot held for no benefit — or,
+        # worse, one that survives past a closed event loop and gets handed
+        # to a later, unrelated caller (#2062's failure class). NullPool caps
+        # a short-lived engine's footprint at exactly one connection instead
+        # of up to 15 (pool_size=5 + max_overflow=10) sitting idle.
         kwargs["pool_pre_ping"] = True
+        if short_lived:
+            kwargs["poolclass"] = NullPool
 
     engine = create_async_engine(db_url, **kwargs)
     if is_sqlite:

packages/core/src/repowise/core/workspace/update.py

@@ -51,26 +51,73 @@
 _log = logging.getLogger("repowise.workspace.update")
 
 
-def _merged_repo_excludes(
+async def _merged_repo_excludes(
     repo_path: Path,
     extra_exclude_patterns: list[str] | None = None,
 ) -> list[str]:
+    """Merge config.yaml excludes with the persisted repo settings' excludes.
+
+    Local SQLite indexes store ``settings_json`` in the repo-local
+    ``wiki.db``, read directly since this call is on the hot per-update path
+    and a raw sqlite3 read is cheaper than opening an engine for one row.
+    When a shared database is configured, that DB is the source of truth
+    instead — reading only ``config.yaml`` there would silently drop every
+    pattern the user added through the shared repository's settings, so
+    every workspace update would run with a narrower exclude set than the
+    repo was actually indexed with.
+    """
     from ..repo_config import load_repo_config
 
     patterns: list[str] = list(load_repo_config(repo_path).get("exclude_patterns") or [])
-    db_path = repo_path / ".repowise" / "wiki.db"
-    if db_path.is_file():
+
+    from ..persistence.database import get_configured_db_url
+
+    configured_url = get_configured_db_url()
+    settings_json: str | None = None
+
+    if configured_url is None:
+        db_path = repo_path / ".repowise" / "wiki.db"
+        if db_path.is_file():
+            try:
+                with sqlite3.connect(str(db_path)) as conn:
+                    row = conn.execute("SELECT settings_json FROM repositories LIMIT 1").fetchone()
+                if row and row[0]:
+                    settings_json = row[0]
+            except Exception:
+                pass
+    else:
+        from ..persistence import (
+            create_engine,
+            create_session_factory,
+            get_session,
+            init_db,
+        )
+        from ..persistence.crud import get_repository_by_path
+
+        try:
+            engine = create_engine(configured_url)
+            try:
+                await init_db(engine)
+                sf = create_session_factory(engine)
+                async with get_session(sf) as session:
+                    repo = await get_repository_by_path(session, str(repo_path))
+                    if repo is not None and repo.settings_json:
+                        settings_json = repo.settings_json
+            finally:
+                await engine.dispose()
+        except Exception:
+            pass
+
+    if settings_json:
         try:
-            with sqlite3.connect(str(db_path)) as conn:
-                row = conn.execute("SELECT settings_json FROM repositories LIMIT 1").fetchone()
-            if row and row[0]:
-                settings = _json.loads(row[0])
-                if isinstance(settings, dict):
-                    for value in settings.get("exclude_patterns") or []:
-                        if isinstance(value, str) and value not in patterns:
-                            patterns.append(value)
+            settings = _json.loads(settings_json)
+            if isinstance(settings, dict):
+                for value in settings.get("exclude_patterns") or []:
+                    if isinstance(value, str) and value not in patterns:
+                        patterns.append(value)
         except Exception:
             pass
+
     for pattern in extra_exclude_patterns or []:
         if pattern not in patterns:
             patterns.append(pattern)
@@ -305,16 +352,26 @@
     sync-check — a routine ``repowise update`` that finds nothing to do still
     counts as "verified current now". Creates the row when it is missing from
     an existing ``wiki.db`` (self-heals a corrupt/blank store — the policy the
-    CLI's ``stamp_head_commit``, now a thin wrapper over this, always had);
+    CLI's ``stamp_head_commit``, now a thin wrapper over this, always had).
+
+    When a shared database is configured (``get_configured_db_url()``), that
+    DB is the source of truth and this always runs — a repo-local ``wiki.db``
+    is not expected to exist in that mode. Otherwise (local SQLite), this is
     still a no-op when ``wiki.db`` itself is absent, so a stamp can never
-    conjure an empty database.
+    conjure an empty local database.
 
     This is the single head-commit stamper for both update paths — the CLI
     fast paths and the workspace updater used to run two implementations with
     different creation semantics.
     """
-    if not head or not (repo_path / ".repowise" / "wiki.db").is_file():
+    if not head:
+        return
+    from ..persistence.database import get_configured_db_url
+
+    configured_url = get_configured_db_url()
+    if configured_url is None and not (repo_path / ".repowise" / "wiki.db").is_file():
         return
+
     from ..persistence import (
         create_engine,
         create_session_factory,
@@ -325,7 +382,7 @@
     from ..persistence.crud import get_repository_by_path
     from ..persistence.database import resolve_db_url
 
-    url = resolve_db_url(repo_path)
+    url = configured_url or resolve_db_url(repo_path)
     engine = create_engine(url)
     try:
         await init_db(engine)
@@ -407,16 +464,14 @@
         # New commits but nothing the index cares about changed (merge/empty
         # commits, or every change excluded). Report success so the caller
         # bumps ``last_sync_commit`` instead of re-diffing forever.
-        return RepoUpdateResult(
-            alias=alias, updated=True, working_tree_paths=working_tree_paths
-        )
+        return RepoUpdateResult(alias=alias, updated=True, working_tree_paths=working_tree_paths)
 
     # Per-repo config, like the single-repo update path. The workspace-level
     # ``exclude_patterns`` (when provided) apply on top.
     from ..repo_config import load_repo_config
 
     cfg = load_repo_config(repo_path)
-    merged_excludes = _merged_repo_excludes(repo_path, exclude_patterns)
+    merged_excludes = await _merged_repo_excludes(repo_path, exclude_patterns)
 
     # Decay-only rows for idle files the anchor advance recovered (#728);
     # persisted alongside the changed rows, kept out of git_meta_map so partial
@@ -542,6 +597,36 @@
     )
 
 
+async def _has_persisted_repo_index(repo_path: Path) -> bool:
+    """Return whether this repo has persisted index data.
+
+    Local SQLite indexes are identified by their repo-local wiki.db.
+    When a shared database is configured, the repository row is the source
+    of truth instead; a repo-local wiki.db is not expected to exist.
+    """
+    from ..persistence.database import get_configured_db_url
+
+    if get_configured_db_url() is None:
+        return (repo_path / ".repowise" / "wiki.db").is_file()
+
+    from ..persistence import (
+        create_engine,
+        create_session_factory,
+        get_session,
+        init_db,
+    )
+    from ..persistence.crud import get_repository_by_path
+
+    engine = create_engine(get_configured_db_url())
+    try:
+        await init_db(engine)
+        sf = create_session_factory(engine)
+        async with get_session(sf) as session:
+            return await get_repository_by_path(session, str(repo_path)) is not None
+    finally:
+        await engine.dispose()
+
+
 async def update_single_repo_index(
     repo_path: Path,
     *,
@@ -565,7 +650,7 @@
     alias = repo_path.name
     state = read_repo_state(repo_path)
     base_ref = state.get("last_sync_commit")
-    merged_excludes = _merged_repo_excludes(repo_path, exclude_patterns)
+    merged_excludes = await _merged_repo_excludes(repo_path, exclude_patterns)
 
     # Config drift check, mirroring the single-repo update path: a changed
     # config.yaml / health-rules.json invalidates persisted health scores and
@@ -577,7 +662,9 @@
     # surprise full re-index.
     stored_fp = state.get("config_fingerprint")
     config_changed = stored_fp is not None and stored_fp != config_fingerprint(repo_path)
-    if config_changed and (repo_path / ".repowise" / "wiki.db").is_file():
+
+    has_persisted_index = await _has_persisted_repo_index(repo_path)
+    if config_changed and has_persisted_index:
         _log.info(
             "workspace_update: %s config fingerprint drifted — full re-index "
             "so health scores reflect the new config",
@@ -587,7 +674,7 @@
     if (
         not config_changed
         and base_ref
-        and (repo_path / ".repowise" / "wiki.db").is_file()
+        and has_persisted_index
         and commit_exists(repo_path, str(base_ref))
     ):
         try:
@@ -798,9 +885,7 @@
             if new_head:
                 with suppress(OSError):
                     (path / ".repowise").mkdir(parents=True, exist_ok=True)
-                    (path / ".repowise" / ".update.pending").write_text(
-                        new_head, encoding="utf-8"
-                    )
+                    (path / ".repowise" / ".update.pending").write_text(new_head, encoding="utf-8")
         return [
             RepoUpdateResult(
                 alias=alias,
@@ -848,7 +933,9 @@
                     )
                     # Record pending so the running update can roll forward.
                     with suppress(OSError):
-                        (path / ".repowise" / ".update.pending").write_text(new_head, encoding="utf-8")
+                        (path / ".repowise" / ".update.pending").write_text(
+                            new_head, encoding="utf-8"
+                        )
                     return RepoUpdateResult(
                         alias=alias,
                         updated=False,
@@ -1019,202 +1106,202 @@
         else WorkspaceIndex({})
     )
     try:
-        await _run_phases(
-            ws_config, workspace_root, changed_repos, timings, workspace_index
-        )
+        await _run_phases(ws_config, workspace_root, changed_repos, timings, workspace_index)
     finally:
         await workspace_index.close()
 
 
 async def _run_phases(
     ws_config: WorkspaceConfig,
     workspace_root: Path,
     changed_repos: list[str],
     timings: PhaseTimingRecorder,
     workspace_index: WorkspaceIndex,
 ) -> None:
     """The five cross-repo phases, over an already-open workspace index."""
     from .breaking_change import run_breaking_change_detection
     from .conformance import run_conformance_check
     from .contracts import ContractStore, load_contract_store, run_contract_extraction
     from .cross_repo import CrossRepoOverlay, run_cross_repo_analysis
     from .system_graph import (
         SystemGraph,
         _detect_boundaries_by_repo,
         run_system_graph_build,
     )
 
     # Service boundaries, detected once for the whole workspace. Contract
     # extraction and the system-graph build both need them and each used to walk
     # every repo for its own copy.
     boundaries_by_repo: dict[str, list[ServiceBoundary]] = {}
     timings.on_phase_start("boundaries", None)
     try:
         boundaries_by_repo = await asyncio.to_thread(
             _detect_boundaries_by_repo, ws_config, workspace_root
         )
     except Exception:
         _log.warning("Service boundary detection failed", exc_info=True)
     timings.on_phase_done("boundaries")
     _log_phase(
         timings,
         "boundaries",
         repos=len(boundaries_by_repo),
         boundaries=sum(len(b) for b in boundaries_by_repo.values()),
     )
 
     # Snapshot the contracts as they stand on disk BEFORE extraction overwrites
     # them. Two jobs, one read: the baseline for the breaking-change diff, and
     # the source of any rows extraction carries forward instead of recomputing.
     # Loaded here rather than inside extraction so the ordering against the
     # write is explicit and the 360 KB parse happens once.
     previous_store = load_contract_store(workspace_root) or ContractStore()
 
     # Phases 1 and 2 are independent: they read disjoint inputs (git history and
     # manifests vs source files and the symbol index), write different artifacts
     # (cross_repo_edges.json vs contracts.json), and share no mutable state —
     # boundaries_by_repo is computed above and passed in read-only. Both push
     # their blocking work through asyncio.to_thread, so gather genuinely
     # overlaps them rather than interleaving two synchronous bodies.
     # Each coroutine times itself, so running them together does not cost the
     # ability to say which one dominates — the question this instrumentation
     # exists to answer, and one a single duration for the pair would destroy.
     async def _timed(phase: str, coro: Awaitable[Any]) -> Any:
         timings.on_phase_start(phase, None)
         try:
             return await coro
         finally:
             timings.on_phase_done(phase)
 
     overlay_result, store_result = await asyncio.gather(
-        _timed("cross_repo_analysis", run_cross_repo_analysis(ws_config, workspace_root, changed_repos)),
+        _timed(
+            "cross_repo_analysis", run_cross_repo_analysis(ws_config, workspace_root, changed_repos)
+        ),
         _timed(
             "contract_extraction",
             run_contract_extraction(
                 ws_config,
                 workspace_root,
                 changed_repos,
                 boundaries_by_repo or None,
                 previous_store,
                 workspace_index,
             ),
         ),
         return_exceptions=True,
     )
 
     # Cancellation and interpreter-level exits are not phase failures — the old
     # `except Exception` let them propagate by construction, and gather's
     # return_exceptions=True does not. Re-raise them before anything downstream
     # treats a cancelled run as a completed one.
     for result in (overlay_result, store_result):
         if isinstance(result, BaseException) and not isinstance(result, Exception):
             raise result
 
     overlay = CrossRepoOverlay()
     if isinstance(overlay_result, Exception):
         _log.warning("Cross-repo analysis failed", exc_info=overlay_result)
     else:
         overlay = overlay_result
 
     store = ContractStore()
     extraction_ok = not isinstance(store_result, Exception)
     if isinstance(store_result, Exception):
         _log.warning("Contract extraction failed", exc_info=store_result)
     else:
         store = store_result
 
     _log_phase(
         timings,
         "cross_repo_analysis",
         co_changes=len(overlay.co_changes),
         package_deps=len(overlay.package_deps),
     )
     _log_phase(
         timings,
         "contract_extraction",
         contracts=len(store.contracts),
         links=len(store.contract_links),
         # Repos actually walked this run — the number that separates a real
         # speed-up from a phase that got faster by doing less than it claims.
         # Derived from the stamp rather than summing extraction_stats, which
         # also carries the counters of the repos this run skipped.
         repos_extracted=sum(
             1 for p in store.repo_provenance.values() if p.get("extracted_at") == store.generated_at
         ),
         repos_reused=sum(
             1 for p in store.repo_provenance.values() if p.get("extracted_at") != store.generated_at
         ),
     )
 
     # System graph — the normalized service-granular structure every workspace
     # view reads. Built last so it folds in the contracts and overlay above.
     system_graph: SystemGraph | None = None
     timings.on_phase_start("system_graph", None)
     try:
         system_graph = await run_system_graph_build(
             ws_config, workspace_root, store, overlay, boundaries_by_repo or None
         )
     except Exception:
         _log.warning("System graph build failed", exc_info=True)
     timings.on_phase_done("system_graph")
     _log_phase(
         timings,
         "system_graph",
         nodes=len(system_graph.nodes) if system_graph else 0,
         edges=len(system_graph.edges) if system_graph else 0,
     )
 
     # Breaking-change guard — diff the previous (on-disk) contracts against the
     # freshly extracted set and persist the impacted-consumer report.
     #
     # Gated on the SHAPE of the result, not on whether an exception escaped.
     # Diffing a populated baseline against an empty set reports every contract
     # in the workspace as removed, and extraction has more than one way to hand
     # back nothing: it raises, or it returns an empty store early because fewer
     # than two repos are currently indexed (a repo unmounted, renamed, or having
     # just lost its .repowise/). Both would publish themselves as the largest
     # breaking change the workspace has ever seen. An empty result is only
     # trustworthy when the baseline is empty too.
     if extraction_ok and (store.contracts or not previous_store.contracts):
         report = None
         timings.on_phase_start("breaking_change", None)
         try:
             report = run_breaking_change_detection(workspace_root, previous_store, store)
         except Exception:
             _log.warning("Breaking-change detection failed", exc_info=True)
         timings.on_phase_done("breaking_change")
         _log_phase(
             timings,
             "breaking_change",
             breaking=len(report.changes) if report else 0,
         )
     else:
         _log.warning(
             "Skipping breaking-change detection: extraction produced no contracts "
             "(ok=%s) while the previous artifact holds %d — diffing these would "
             "report the whole workspace as removed",
             extraction_ok,
             len(previous_store.contracts),
         )
 
     # Conformance + cycles — check declared dependency rules and detect circular
     # service dependencies over the freshly-built system graph.
     if system_graph is not None:
         conformance = None
         timings.on_phase_start("conformance", None)
         try:
             conformance = run_conformance_check(ws_config, workspace_root, system_graph)
         except Exception:
             _log.warning("Conformance check failed", exc_info=True)
         timings.on_phase_done("conformance")
         _log_phase(
             timings,
             "conformance",
             rules=conformance.rules_evaluated if conformance else 0,
             violations=len(conformance.violations) if conformance else 0,
             # The true total, not the capped list — a workspace with 500 cycles
             # must not report the same number as one with 50.
             cycles=conformance.total_cycles if conformance else 0,
         )
 
     _log.info("Cross-repo phase timings (seconds): %s", timings.timings)

packages/server/src/repowise/server/app.py

@@ -184,7 +184,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             pass  # Fall back to default
 
-    engine = create_engine(db_url)
+    engine = create_engine(db_url, short_lived=False)
     await init_db(engine)
     session_factory = create_session_factory(engine)
 
@@ -327,6 +327,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
                     app.state.workspace_fts[repo_id] = fts
                     continue
 
+                # NullPool is irrelevant here since SQLite ignores create_engine()'s
+                # short_lived flag — but this engine is stored in app.state for the
+                # server's lifetime (workspace_engines/workspace_sessions), matching the
+                # long-lived pattern in the two engines above. If this URL is ever made
+                # configurable (e.g. pointed at a shared REPOWISE_DB_URL for workspace
+                # members), it must pass short_lived=False too.
                 repo_engine = create_engine(f"sqlite+aiosqlite:///{db_url_posix}")
                 await init_db(repo_engine)
                 repo_sf = create_session_factory(repo_engine)

packages/server/src/repowise/server/mcp_server/_server.py

@@ -527,7 +527,7 @@ async def _lifespan(server: FastMCP):
     db_url = resolve_db_url(_state._repo_path)
 
     _log.info("repowise MCP: initialising database…")
-    engine = create_engine(db_url)
+    engine = create_engine(db_url, short_lived=False)
     await init_db(engine)
 
     _state._session_factory = async_sessionmaker(

tests/unit/server/mcp/test_config.py

@@ -62,8 +62,9 @@ async def fake_init_db(engine) -> None:
     async def fake_load_vector_stores(repo_path: str | None) -> None:
         return None
 
-    def fake_create_engine(url: str) -> DummyEngine:
+    def fake_create_engine(url: str, **kwargs) -> DummyEngine:
         captured["url"] = url
+        captured["short_lived"] = kwargs.get("short_lived")
         return DummyEngine()
 
     monkeypatch.setenv("REPOWISE_DB_URL", "sqlite+aiosqlite:///tmp/from-cli.db")
@@ -88,6 +89,7 @@ def fake_create_engine(url: str) -> DummyEngine:
     try:
         async with mcp_server._lifespan(mcp_server.mcp):
             assert captured["url"] == "sqlite+aiosqlite:///tmp/from-cli.db"
+            assert captured["short_lived"] is False
     finally:
         _state._repo_path = original_repo_path
         _state._vector_store = original_vector_store