diff --git a/changelogs/v0.7.0/briancastelino_2026-07-09.log b/changelogs/v0.7.0/briancastelino_2026-07-09.log
new file mode 100644
index 00000000..87a37e58
--- /dev/null
+++ b/changelogs/v0.7.0/briancastelino_2026-07-09.log
@@ -0,0 +1,143 @@
+## Stop broad Graph Chat queries from freezing the whole app (issue #114)
+
+### Context
+A single Graph Chat question that triggered a broad graph read (deep BFS
+traversal, unfiltered `triples/find`, or a heavy GraphQL resolver) could make
+the entire OntoBricks app unresponsive for every user until the query finished
+or the browser gave up. Two root causes compounded each other:
+
+1. **Event-loop starvation.** The internal `/dtwin/...` graph routes the chat
+ agent calls over loopback ran their blocking SQL *directly on the single
+ uvicorn event loop*. While one slow query ran, no other request — health
+ checks, other users' pages, other chat turns — could be serviced.
+2. **Unbounded reads.** Nothing capped how long a graph read could run or how
+ many triples it could return, so a runaway query pinned a DB session (and,
+ through the loopback agent, the event loop) indefinitely.
+
+The fix bounds every graph read server-side, offloads the blocking work off the
+event loop, auto-sizes the worker pool to the instance, and — when the app is
+genuinely saturated — tells the user why and what to do about it, with an admin
+knob to tune the bounds.
+
+### Changes
+1. `src/back/core/query_limits.py` (new)
+ Central resolver for the two graph-read bounds — statement timeout and Graph
+ Chat result cap — with admin-override → env-var → built-in-default precedence
+ and clamping so a misconfiguration can't disable the guard.
+2. `src/back/core/graphdb/lakebase/LakebaseFlatStore.py`
+ `execute_query` sets a per-read `SET statement_timeout` (ms) from
+ `get_graph_query_timeout_s()` so a runaway read is cancelled server-side
+ instead of pinning the pooled connection.
+3. `src/back/core/graphdb/lakebase/LakebaseBase.py`
+ The DDL/migration `_cursor` resets `SET statement_timeout = 0` so schema work
+ on a reused pooled connection isn't cancelled by a prior read's bound.
+4. `src/back/core/databricks/SQLWarehouse.py`
+ `execute_query` gained an optional `statement_timeout_s`: when positive it
+ brackets the query with `SET STATEMENT_TIMEOUT` and resets it to `0` on the
+ pooled session afterwards.
+5. `src/back/core/triplestore/delta/DeltaTripleStore.py`
+ Routes bounded graph reads through `SQLWarehouse.execute_query` with the
+ resolved timeout; full-graph dumps stay unbounded.
+6. `src/api/routers/internal/dtwin.py`
+ Offloaded the blocking DB work in `triples/find`, `graphql/execute`,
+ `neighbors`, and `sync/stats` into `run_blocking` (extract-function of the
+ blocking body); `triples/find` now clamps `limit` to the result cap. Added
+ `_resource_pressure_payload()` and merged it into both chat responses
+ (`/assistant/chat` and the SSE `done` event).
+7. `src/back/core/helpers/DatabricksHelpers.py` (+ `__init__.py`)
+ Auto-sizes the blocking `ThreadPoolExecutor` from vCPU count (`cpu*4`, floored
+ at the historical 20, still overridable via `ONTOBRICKS_THREAD_POOL_SIZE`),
+ tracks in-flight tasks in `run_blocking`, and exposes `get_blocking_pool_stats()`
+ with a `saturated` flag.
+8. `src/agents/agent_dtwin_chat/tools.py`
+ The loopback HTTP client timeout is now derived from the server statement
+ timeout plus a margin, so a bounded query is cancelled server-side (clean
+ error the LLM can react to) before the client aborts.
+9. `src/back/objects/session/GlobalConfigService.py`
+ Persists/applies `graph_query_timeout_s` and `graph_chat_result_cap`
+ (get/set + `_empty()` keys + re-apply on config load; `0` = unset).
+10. `src/back/objects/domain/SettingsService.py`
+ `get_graph_limits_result` / `save_graph_limits_result` (admin-gated).
+11. `src/api/routers/internal/settings.py`
+ `GET /settings/graph-limits` and `POST /settings/save-graph-limits`.
+12. `src/front/templates/settings.html`, `src/front/static/config/js/settings.js`
+ Admin-only "Graph read limits" inputs on the Graph DB tab (load + save).
+13. `src/front/static/query/js/query-chat.js`
+ Inline advisory bubble + global toast when the server flags resource pressure.
+14. `docs/issues/graph-chat-event-loop-hang.md` (new)
+ Bug write-up: symptom, root cause, and remediation.
+
+### Files modified
+- `src/back/core/query_limits.py`
+- `src/back/core/graphdb/lakebase/LakebaseFlatStore.py`
+- `src/back/core/graphdb/lakebase/LakebaseBase.py`
+- `src/back/core/databricks/SQLWarehouse.py`
+- `src/back/core/triplestore/delta/DeltaTripleStore.py`
+- `src/api/routers/internal/dtwin.py`
+- `src/api/routers/internal/settings.py`
+- `src/back/core/helpers/DatabricksHelpers.py`
+- `src/back/core/helpers/__init__.py`
+- `src/agents/agent_dtwin_chat/tools.py`
+- `src/back/objects/session/GlobalConfigService.py`
+- `src/back/objects/domain/SettingsService.py`
+- `src/front/templates/settings.html`
+- `src/front/static/config/js/settings.js`
+- `src/front/static/query/js/query-chat.js`
+- `docs/issues/graph-chat-event-loop-hang.md`
+- `tests/units/core/test_query_limits.py` (new)
+- `tests/units/core/test_graph_query_bounds.py` (new)
+
+### Related work
+Contributor @ulsmo (issue #114 comment) confirmed the bug and traced the
+triggers on a 1–5M-triple graph to two performance bottlenecks this change
+complements but does not replace: #112 (indexes dropped/not applied during
+Lakeflow sync → unindexed `_sync` tables → exploding query times) and PR #115
+(optimizes the `expand_uri_aliases` / `get_triples_for_subjects` path that
+`triples/find` invokes). This change is the resilience layer: it bounds and
+offloads reads so a slow query degrades to a clean per-request cancel instead of
+an app-wide freeze. Trade-off: on an unindexed large graph a legitimate read may
+now hit the default 60s timeout and be cancelled; the admin knob (≤900s) covers
+that until #112/#115 land.
+
+### Review fixups (Copilot, PR #116)
+- **[High] Lakebase statement_timeout leak.** The pool hands out
+ `autocommit=True` connections, so the per-read `SET statement_timeout` in
+ `LakebaseFlatStore.execute_query` persisted to the next borrower and could
+ cancel a bulk COPY/INSERT or DDL mid-write. Now wrapped in `try/finally` with
+ `RESET statement_timeout` so the bound is scoped to the read (safe in
+ autocommit — a timeout cancel leaves no open transaction to abort).
+- **[Med] Settings input validation.** `save-graph-limits` `_opt_int` now raises
+ `ValidationError` on non-integer input instead of letting `int()` bubble as a
+ 500.
+- **[Med] Config/UI vs runtime consistency.** `GlobalConfigService`
+ get/set for `graph_query_timeout_s` and `graph_chat_result_cap` now route the
+ persisted value through the central clamp (`back.core.query_limits`) so the
+ Settings UI never shows or re-saves an out-of-range value that differs from the
+ enforced effective bound.
+- Added `TestLakebaseStatementTimeoutReset` (reset on success and on error) to
+ `tests/units/core/test_graph_query_bounds.py`.
+
+### Rebase onto `develop` (v0.7.0, PR #116)
+Retargeted from `master` to the `develop` integration branch. `develop` had
+already migrated `triplestore` → `graphdb` and added its own `run_blocking`
+offloading for the graph routes, so the redundant offloading half of this change
+was dropped in favour of develop's, and only the unique bounding/advisory layer
+was kept:
+- `triplestore/delta/DeltaTripleStore.py` was deleted on `develop`; the bounded
+ read was ported to `graphdb/delta/DeltaFlatStore.execute_query`.
+- `dtwin.py` kept only the resource-pressure advisory (`_resource_pressure_payload`
+ in the chat + SSE responses) and the `get_graph_chat_result_cap()` clamp on
+ `triples/find`; the `run_blocking` wrapping is now inherited from `develop`.
+- `settings.js` kept only `loadGraphLimits()` + the save handler; the Graph DB tab
+ refactor is `develop`'s.
+- Changelog moved from `v0.6.1/` to `v0.7.0/` to match the `develop` version line.
+- `TestDeltaStoreBoundsReads` updated to target `DeltaFlatStore`.
+
+### Tests
+`uv` is not installed in this dev environment and system Python lacks `psycopg`,
+so the full `uv run pytest -q -m "not scenario"` suite could not be run here.
+Ran the targeted subset with system Python + pytest:
+`python -m pytest tests/units/core/test_query_limits.py tests/units/core/test_graph_query_bounds.py tests/units/core/test_sql_warehouse.py -q`
+→ **41 passed** (includes the pre-existing `test_sql_warehouse.py` and the two new
+Lakebase timeout-reset tests — no regression). Full-suite run still required in a
+`uv`/`psycopg` environment before merge.
diff --git a/docs/issues/graph-chat-event-loop-hang.md b/docs/issues/graph-chat-event-loop-hang.md
new file mode 100644
index 00000000..8a56e670
--- /dev/null
+++ b/docs/issues/graph-chat-event-loop-hang.md
@@ -0,0 +1,114 @@
+# [BUG]: Graph Chat heavy query blocks the event loop and freezes the app until redeploy
+
+> Tracked as GitHub issue #114. Formatted to match `.github/ISSUE_TEMPLATE/bug.yml`.
+
+## Is there an existing issue for this?
+
+- [x] I have searched the existing issues
+
+## Current Behavior
+
+Asking Graph Chat a broad question (e.g. *"What are the top 10 violations?"*) hangs
+the request and then freezes the **entire** app: no other request — from any user —
+completes until the app is redeployed.
+
+Root cause: the Graph Chat agent runs in a worker thread and makes **synchronous
+loopback HTTP calls** back into the same FastAPI process. The internal `/dtwin/...`
+routes it hits (`dtwin_triples_find`, `dtwin_graphql_execute`, `dtwin_neighbors`,
+`sync/stats`) are declared `async def` but execute **blocking DB work directly on the
+asyncio event loop** — notably the recursive BFS CTE in
+`TripleStoreBackend.bfs_traversal`. That query joins on
+`(t.subject = b.entity OR t.object = b.entity)` with no `LIMIT` and no DB
+`statement_timeout`, so a broad seed explodes into a multi-minute / effectively
+unbounded query. Because it runs on the event loop, the single uvicorn worker is
+frozen for the whole query, stalling every other request. The agent's httpx client
+timeout fires client-side (`triples/find error: timed out`) but does **not** cancel
+the server-side query, so the loop stays blocked and the Lakebase connection pool
+(`_POOL_MAX_SIZE = 4`) gets pinned.
+
+This violates the project's own guidance in `src/.coding_rules.md`
+("synchronous I/O in an async handler blocks the event loop … wrap it in
+`asyncio.to_thread`"). The helper already exists: `DatabricksHelpers.run_blocking()`.
+
+## Expected Behavior
+
+- Slow/broad questions time out **gracefully** with an error message the agent can
+ react to, cancelled server-side by a `statement_timeout`.
+- The app stays responsive to all other users/requests while a heavy Graph Chat
+ query runs (blocking DB work is offloaded off the event loop).
+- Under sustained load, the app advises upgrading the Databricks App instance size
+ instead of silently freezing.
+
+## Steps To Reproduce
+
+1. Create / load a domain and build its Knowledge Graph (Lakebase backend).
+2. Open the **Graph Chat** tab.
+3. Ask a broad question, e.g. *"What are the top 10 violations?"*.
+4. Observe the request hang; then observe that all other pages/requests also stall
+ until the app is redeployed.
+
+## Cloud
+
+
+
+## Browser
+
+
+
+## OntoBricks Version
+
+0.6.1
+
+## Relevant log output
+
+```shell
+WARNING agents.agent_graph_chat_tools | tools._error:71 | agent_dbx_chat: triples/find error: timed out
+INFO uvicorn.access | 'POST /dtwin/graphql/execute HTTP/1.1' 200
+# ... after the timeout, no further requests are served until redeploy
+```
+
+## Additional Context
+
+- Graph backend: Lakebase Postgres (the same class of bug affects the Delta/SQL
+ warehouse backend, which likewise has no per-statement cancellation — only a
+ 30s socket timeout).
+- Deployed logs reference `agents.agent_graph_chat_tools` / `agent_dbx_chat`, which
+ are renamed to `agent_dtwin_chat` on current `main`; the architectural bug is
+ identical on `main`.
+
+## Related issues (query-performance layer)
+
+This issue is the **resilience / graceful-degradation** half of the problem. The
+contributor **@ulsmo** confirmed the bug and traced the *triggers* on a
+1–5M-triple graph to two underlying performance bottlenecks — this fix bounds
+their blast radius but does not replace them:
+
+- **#112** — Indexes dropped/not applied during Lakeflow sync, leaving `_sync`
+ tables unindexed and causing exploding query times. A root cause of the slow
+ reads that trip this hang.
+- **#115** — `Perf(graphdb): Optimize finding subjects with local id's` (PR):
+ optimizes `describe_entity` alias expansion on large ID sets — the same
+ `expand_uri_aliases` / `get_triples_for_subjects` path the `triples/find` route
+ invokes.
+
+These are complementary: even with #112 and #115 fixed, an unbounded read can
+still starve the single event loop, so the bounding + offloading here is still
+required.
+
+## Proposed Fix
+
+1. Offload the blocking DB work in the internal `/dtwin` graph routes via
+ `run_blocking(...)` so it never runs on the event loop.
+2. Add a configurable graph-read `statement_timeout` on **both** backends
+ (Lakebase `SET statement_timeout`, warehouse `SET STATEMENT_TIMEOUT`) so a runaway
+ query is cancelled server-side.
+3. Auto-tune the blocking thread pool from the instance size and surface an
+ admin-configurable advisory recommending an instance-size upgrade under pressure.
+
+### Caveat / trade-off
+
+On an **unindexed** large graph (see #112), a legitimate `describe` / `triples/find`
+can now exceed the default 60s `statement_timeout` and be **cancelled** — turning a
+*silent app-wide freeze* into a *clean, per-request "query cancelled" error*. This
+is the intended trade-off; the admin timeout knob (up to 900s) lets operators raise
+the bound until #112/#115 fix the underlying query speed.
diff --git a/src/agents/agent_dtwin_chat/tools.py b/src/agents/agent_dtwin_chat/tools.py
index 155bf6bd..174ec075 100644
--- a/src/agents/agent_dtwin_chat/tools.py
+++ b/src/agents/agent_dtwin_chat/tools.py
@@ -47,7 +47,8 @@
logger = get_logger(__name__)
-_HTTP_TIMEOUT = 120
+_HTTP_TIMEOUT = 120 # fallback floor when the graph timeout can't be resolved
+_HTTP_TIMEOUT_MARGIN_S = 30
_MAX_DEPTH = 1
_SPARQL_DANGEROUS = re.compile(
r"\b(DROP|DELETE|INSERT|CREATE|CLEAR|LOAD|COPY|MOVE|ADD)\b",
@@ -61,8 +62,22 @@
def _client(ctx: ToolContext):
- """Build a sync HTTP client bound to the loopback OntoBricks URL."""
- return loopback_client(ctx, timeout=_HTTP_TIMEOUT)
+ """Build a sync HTTP client bound to the loopback OntoBricks URL.
+
+ The timeout is derived from the server-side graph statement timeout plus a
+ margin, so a bounded query is cancelled *server-side* (returning a clean
+ error the LLM can react to) before the loopback client gives up on a query
+ that is still running.
+ """
+ try:
+ from back.core.query_limits import get_graph_query_timeout_s
+
+ timeout = max(
+ _HTTP_TIMEOUT, get_graph_query_timeout_s() + _HTTP_TIMEOUT_MARGIN_S
+ )
+ except Exception: # noqa: BLE001
+ timeout = _HTTP_TIMEOUT
+ return loopback_client(ctx, timeout=timeout)
def _registry_params(ctx: ToolContext) -> dict:
diff --git a/src/api/routers/internal/dtwin.py b/src/api/routers/internal/dtwin.py
index e15629bf..9fa59adb 100644
--- a/src/api/routers/internal/dtwin.py
+++ b/src/api/routers/internal/dtwin.py
@@ -1832,6 +1832,37 @@ async def get_inferred_triples(
_CHAT_MIN_LIMIT = 5
_CHAT_MAX_LIMIT = 100
+# Surfaced to the Graph Chat UI (inline + global toast) when the blocking
+# thread pool is saturated, so users understand why responses are slow and
+# admins know the actionable remedy.
+_UPGRADE_INSTANCE_ADVICE = (
+ "OntoBricks is under heavy load - the request worker pool is saturated, so "
+ "responses may be slow. If this happens often, upgrade the Databricks App "
+ "instance size (Apps UI -> Compute) for more concurrency."
+)
+
+
+def _resource_pressure_payload() -> dict:
+ """Return a resource-pressure advisory when the blocking pool is saturated.
+
+ Sampled around a Graph Chat turn so the UI can nudge the user toward a
+ larger Databricks App instance instead of silently appearing to hang.
+ Never raises - pressure detection must not break the chat response.
+ """
+ try:
+ from back.core.helpers import get_blocking_pool_stats
+
+ stats = get_blocking_pool_stats()
+ except Exception: # noqa: BLE001
+ return {"resource_pressure": False}
+ if stats.get("saturated"):
+ return {
+ "resource_pressure": True,
+ "resource_advice": _UPGRADE_INSTANCE_ADVICE,
+ "pool_stats": stats,
+ }
+ return {"resource_pressure": False}
+
def _chat_cache(session_mgr: SessionManager) -> dict:
"""Return the Graph Chat session cache, creating an empty one if absent."""
@@ -2098,6 +2129,7 @@ async def dtwin_assistant_chat(
"tools": tool_calls,
"iterations": agent_result.iterations,
"usage": agent_result.usage,
+ **_resource_pressure_payload(),
}
@@ -2251,6 +2283,7 @@ async def _generate():
"iterations": agent_result.iterations,
"usage": agent_result.usage,
"success": agent_result.success,
+ **_resource_pressure_payload(),
}) + "\n\n"
break
@@ -2481,12 +2514,13 @@ async def dtwin_triples_find(
published as a version.
"""
from back.core.helpers import sql_escape
+ from back.core.query_limits import get_graph_chat_result_cap
if not entity_type and not search:
raise ValidationError("Provide at least entity_type or search")
depth = max(1, min(int(depth or 1), 10))
- limit = max(1, min(int(limit or 1000), 10000))
+ limit = max(1, min(int(limit or 1000), get_graph_chat_result_cap()))
offset = max(0, int(offset or 0))
domain = get_domain(session_mgr)
diff --git a/src/api/routers/internal/settings.py b/src/api/routers/internal/settings.py
index 68d5ae58..c99702f6 100644
--- a/src/api/routers/internal/settings.py
+++ b/src/api/routers/internal/settings.py
@@ -642,6 +642,45 @@ async def save_registry_cache_ttl(
)
+@router.get("/graph-limits")
+async def get_graph_limits(
+ session_mgr: SessionManager = Depends(get_session_manager),
+ settings: Settings = Depends(get_settings),
+):
+ """Get effective graph-read bounds (statement timeout + chat result cap)."""
+ return config_service.get_graph_limits_result(session_mgr, settings)
+
+
+@router.post("/save-graph-limits")
+async def save_graph_limits(
+ request: Request,
+ session_mgr: SessionManager = Depends(get_session_manager),
+ settings: Settings = Depends(get_settings),
+):
+ """Save graph-read bounds (admin only, stored globally). ``0`` = unset."""
+ data = await request.json()
+
+ def _opt_int(key: str):
+ if key not in data or data[key] is None or data[key] == "":
+ return None
+ try:
+ return int(data[key])
+ except (TypeError, ValueError) as exc:
+ raise ValidationError(f"{key} must be an integer") from exc
+
+ email, _display_name, user_token, _user_role, _user_domain_role = (
+ _settings_request_identity(request)
+ )
+ return config_service.save_graph_limits_result(
+ _opt_int("graph_query_timeout_s"),
+ _opt_int("graph_chat_result_cap"),
+ email,
+ user_token,
+ session_mgr,
+ settings,
+ )
+
+
@router.get("/edit-lock-ttl")
async def get_edit_lock_ttl(
session_mgr: SessionManager = Depends(get_session_manager),
diff --git a/src/back/core/databricks/SQLWarehouse.py b/src/back/core/databricks/SQLWarehouse.py
index 67f00073..bfe48cd5 100644
--- a/src/back/core/databricks/SQLWarehouse.py
+++ b/src/back/core/databricks/SQLWarehouse.py
@@ -137,15 +137,36 @@ def test_connection(self) -> Tuple[bool, str]:
except Exception as exc:
return False, f"Connection failed: {exc}"
- def execute_query(self, query: str) -> List[Dict[str, Any]]:
- """Execute *query* and return rows as a list of dicts."""
+ def execute_query(
+ self, query: str, statement_timeout_s: Optional[int] = None
+ ) -> List[Dict[str, Any]]:
+ """Execute *query* and return rows as a list of dicts.
+
+ When *statement_timeout_s* is a positive integer the query is bounded
+ by a session ``STATEMENT_TIMEOUT`` so the warehouse cancels a runaway
+ statement server-side. The bound is reset before the (pooled)
+ connection is returned so it never leaks to the next borrower. Used by
+ the graph read path (:class:`DeltaTripleStore`); left unset for the
+ build pipeline and full-graph dumps which may legitimately run longer.
+ """
self._require_warehouse()
+ bounded = bool(statement_timeout_s and int(statement_timeout_s) > 0)
try:
with self._borrow() as conn:
with conn.cursor() as cur:
- cur.execute(query)
- columns = [desc[0] for desc in cur.description]
- return [dict(zip(columns, row)) for row in cur.fetchall()]
+ if bounded:
+ cur.execute(
+ f"SET STATEMENT_TIMEOUT = {int(statement_timeout_s)}"
+ )
+ try:
+ cur.execute(query)
+ columns = [desc[0] for desc in cur.description]
+ return [dict(zip(columns, row)) for row in cur.fetchall()]
+ finally:
+ if bounded:
+ # 0 disables the per-session bound (workspace default
+ # applies) so the recycled connection is unaffected.
+ cur.execute("SET STATEMENT_TIMEOUT = 0")
except Exception as exc:
logger.exception("Error executing query: %s", exc)
raise
diff --git a/src/back/core/graphdb/delta/DeltaFlatStore.py b/src/back/core/graphdb/delta/DeltaFlatStore.py
index 24955a2d..e639c8b8 100644
--- a/src/back/core/graphdb/delta/DeltaFlatStore.py
+++ b/src/back/core/graphdb/delta/DeltaFlatStore.py
@@ -259,6 +259,21 @@ def optimize_table(self, table_name: str) -> None:
self._client.execute_statement(f"OPTIMIZE {table_name}")
def execute_query(self, query: str) -> List[Dict[str, Any]]:
+ """Execute a graph read query, bounded by the graph statement timeout.
+
+ Routed through :meth:`SQLWarehouse.execute_query` with a
+ ``statement_timeout_s`` so a runaway traversal (e.g. the recursive BFS
+ CTE) is cancelled server-side instead of pinning a warehouse session
+ and, via the loopback chat agent, the event loop. The unbounded
+ ``client.execute_query`` is intentionally reserved for full-graph dumps
+ (``query_triples``) and the build pipeline.
+ """
+ from back.core.query_limits import get_graph_query_timeout_s
+
+ timeout_s = get_graph_query_timeout_s()
+ sql_service = getattr(self._client, "sql", None)
+ if sql_service is not None and hasattr(sql_service, "execute_query"):
+ return sql_service.execute_query(query, statement_timeout_s=timeout_s)
return self._client.execute_query(query)
def get_inferred_triple_count(self, table_name: str) -> int:
diff --git a/src/back/core/graphdb/lakebase/LakebaseBase.py b/src/back/core/graphdb/lakebase/LakebaseBase.py
index bfd2f430..5da209c5 100644
--- a/src/back/core/graphdb/lakebase/LakebaseBase.py
+++ b/src/back/core/graphdb/lakebase/LakebaseBase.py
@@ -215,4 +215,8 @@ def _cursor(self) -> Iterator[Any]:
with pool.connection() as conn:
with conn.cursor(row_factory=dict_row) as cur:
cur.execute(f'SET search_path TO "{self._schema}", public')
+ # DDL / migration work may legitimately exceed the bounded
+ # graph-read timeout set by ``execute_query`` on this (pooled,
+ # reused) connection — clear it so schema changes aren't cancelled.
+ cur.execute("SET statement_timeout = 0")
yield cur
diff --git a/src/back/core/graphdb/lakebase/LakebaseFlatStore.py b/src/back/core/graphdb/lakebase/LakebaseFlatStore.py
index 8e12d8e5..071083b2 100644
--- a/src/back/core/graphdb/lakebase/LakebaseFlatStore.py
+++ b/src/back/core/graphdb/lakebase/LakebaseFlatStore.py
@@ -472,15 +472,32 @@ def _require_pg():
return _require_psycopg()
def execute_query(self, query: str) -> List[Dict[str, Any]]:
+ from back.core.query_limits import get_graph_query_timeout_s
+
_, dict_row = self._require_pg()
pool = self._pool()
+ timeout_ms = get_graph_query_timeout_s() * 1000
with pool.connection() as conn:
with conn.cursor(row_factory=dict_row) as cur:
cur.execute(f'SET search_path TO "{self._schema}", public')
- cur.execute(query)
- if cur.description:
- return [dict(row) for row in cur.fetchall()]
- return []
+ # Bound graph reads so a runaway traversal is cancelled
+ # server-side instead of pinning the connection (and, via the
+ # loopback agent, the event loop).
+ #
+ # The pool hands out ``autocommit=True`` connections, so this is
+ # a *session* GUC that would otherwise persist to the next
+ # borrower (e.g. a bulk COPY/INSERT or DDL path) and cancel it
+ # mid-write. Reset it in ``finally`` so the bound is scoped to
+ # this read only. In autocommit mode a statement_timeout cancel
+ # leaves no open transaction to abort, so the RESET runs cleanly.
+ cur.execute(f"SET statement_timeout = {int(timeout_ms)}")
+ try:
+ cur.execute(query)
+ if cur.description:
+ return [dict(row) for row in cur.fetchall()]
+ return []
+ finally:
+ cur.execute("RESET statement_timeout")
def find_subjects_by_patterns(
self, table_name: str, like_patterns: List[str]
diff --git a/src/back/core/helpers/DatabricksHelpers.py b/src/back/core/helpers/DatabricksHelpers.py
index e7120a72..7707b0f1 100644
--- a/src/back/core/helpers/DatabricksHelpers.py
+++ b/src/back/core/helpers/DatabricksHelpers.py
@@ -1,5 +1,6 @@
import asyncio
import os
+import threading
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from typing import Any, Callable, Dict, Tuple
@@ -11,11 +12,62 @@
logger = get_logger(__name__)
+# Floor kept at the historical default so small instances never *lose*
+# capacity; larger Databricks App instances (more vCPUs) scale up automatically.
+_BLOCKING_POOL_MIN = 20
+
+
+def _resolve_blocking_pool_size() -> int:
+ """Resolve the blocking-pool worker count.
+
+ Priority: explicit ``ONTOBRICKS_THREAD_POOL_SIZE`` > instance-size
+ derivation (``vCPUs * 4``, floored at :data:`_BLOCKING_POOL_MIN`). This lets
+ a bigger Databricks App instance get more blocking-worker capacity without a
+ config change, while an operator can still pin an exact value.
+ """
+ explicit = os.getenv("ONTOBRICKS_THREAD_POOL_SIZE", "").strip()
+ if explicit:
+ try:
+ value = int(explicit)
+ if value > 0:
+ return value
+ except ValueError:
+ logger.warning("Ignoring non-integer ONTOBRICKS_THREAD_POOL_SIZE=%r", explicit)
+ cpu = os.cpu_count() or 2
+ return max(_BLOCKING_POOL_MIN, cpu * 4)
+
+
+_BLOCKING_POOL_SIZE = _resolve_blocking_pool_size()
_BLOCKING_POOL = ThreadPoolExecutor(
- max_workers=int(os.getenv("ONTOBRICKS_THREAD_POOL_SIZE", "20")),
+ max_workers=_BLOCKING_POOL_SIZE,
thread_name_prefix="ob-blocking",
)
+# In-flight blocking tasks (submitted but not yet finished). Used to detect
+# thread-pool saturation so the app can advise an instance-size upgrade instead
+# of silently queueing (and appearing to hang).
+_inflight_lock = threading.Lock()
+_inflight_blocking = 0
+_peak_inflight_blocking = 0
+
+
+def get_blocking_pool_stats() -> Dict[str, Any]:
+ """Return blocking thread-pool utilisation.
+
+ ``saturated`` is true when every worker is busy, i.e. further blocking work
+ is queued rather than running — the signal used to surface the
+ resource-pressure advisory.
+ """
+ with _inflight_lock:
+ active = _inflight_blocking
+ peak = _peak_inflight_blocking
+ return {
+ "max_workers": _BLOCKING_POOL_SIZE,
+ "active": active,
+ "peak": peak,
+ "saturated": active >= _BLOCKING_POOL_SIZE,
+ }
+
def make_volume_file_service(domain, settings=None):
"""Return :class:`VolumeFileService` using host/token from ``get_databricks_host_and_token``."""
@@ -48,18 +100,28 @@ class DatabricksHelpers:
async def run_blocking(func: Callable, *args: Any, **kwargs: Any) -> Any:
"""Run a blocking function in a sized thread pool.
- Uses a dedicated :class:`ThreadPoolExecutor` (default 20 threads,
- configurable via ``ONTOBRICKS_THREAD_POOL_SIZE``) instead of the
- default asyncio executor so that concurrent blocking work does not
- starve the event loop.
+ Uses a dedicated :class:`ThreadPoolExecutor` (sized from the instance's
+ vCPU count, floored at 20, overridable via ``ONTOBRICKS_THREAD_POOL_SIZE``)
+ instead of the default asyncio executor so that concurrent blocking work
+ does not starve the event loop. In-flight tasks are tracked so
+ :func:`get_blocking_pool_stats` can report saturation.
Usage in an ``async def`` route handler::
result = await run_blocking(client.execute_query, sql)
"""
+ global _inflight_blocking, _peak_inflight_blocking
loop = asyncio.get_running_loop()
call = partial(func, *args, **kwargs) if kwargs else partial(func, *args)
- return await loop.run_in_executor(_BLOCKING_POOL, call)
+ with _inflight_lock:
+ _inflight_blocking += 1
+ if _inflight_blocking > _peak_inflight_blocking:
+ _peak_inflight_blocking = _inflight_blocking
+ try:
+ return await loop.run_in_executor(_BLOCKING_POOL, call)
+ finally:
+ with _inflight_lock:
+ _inflight_blocking -= 1
@staticmethod
def _resolve_registry_cfg(domain, settings) -> Dict[str, str]:
diff --git a/src/back/core/helpers/__init__.py b/src/back/core/helpers/__init__.py
index 9a37fd63..d16effc9 100644
--- a/src/back/core/helpers/__init__.py
+++ b/src/back/core/helpers/__init__.py
@@ -3,6 +3,7 @@
from back.core.helpers.DatabricksHelpers import ( # noqa: F401
DatabricksHelpers,
effective_uc_version_path,
+ get_blocking_pool_stats,
make_volume_file_service,
)
from back.core.helpers.SQLHelpers import SQLHelpers # noqa: F401
@@ -64,6 +65,7 @@
"get_triplestore_sql_credentials",
"get_databricks_host_and_token",
"make_volume_file_service",
+ "get_blocking_pool_stats",
"require_serving_llm",
"effective_uc_version_path",
"sql_escape",
diff --git a/src/back/core/query_limits.py b/src/back/core/query_limits.py
new file mode 100644
index 00000000..8e86662f
--- /dev/null
+++ b/src/back/core/query_limits.py
@@ -0,0 +1,112 @@
+"""Runtime-tunable bounds for graph *read* queries.
+
+Two knobs are resolved here and consumed by the graph store backends and the
+Graph Chat routes so a single broad question can no longer pin a DB connection
+or an event-loop worker indefinitely:
+
+* **statement timeout** — how long a graph read query may run before the
+ database cancels it server-side (Lakebase ``SET statement_timeout`` /
+ warehouse ``SET STATEMENT_TIMEOUT``).
+* **chat result cap** — the hard ceiling on triples returned by the
+ session-aware ``/dtwin/triples/find`` route the agent calls.
+
+Resolution order for each knob (first hit wins):
+
+1. **admin override** — persisted in the registry global config and applied via
+ :func:`set_graph_query_timeout_override` / :func:`set_graph_chat_result_cap_override`
+ (Settings → Graph DB, and re-applied whenever the settings blob is loaded).
+2. **environment variable** — ``ONTOBRICKS_GRAPH_QUERY_TIMEOUT_S`` /
+ ``ONTOBRICKS_GRAPH_CHAT_RESULT_CAP``.
+3. **built-in default**.
+
+These are deliberately independent of the generic (registry / build-pipeline)
+SQL paths: only graph reads are bounded, so long-running builds and full-graph
+dumps are unaffected.
+"""
+
+from __future__ import annotations
+
+import os
+import threading
+from typing import Optional
+
+from back.core.logging import get_logger
+
+logger = get_logger(__name__)
+
+DEFAULT_GRAPH_QUERY_TIMEOUT_S = 60
+DEFAULT_GRAPH_CHAT_RESULT_CAP = 10_000
+
+# Sane bounds so a misconfiguration can't disable the guard entirely or set a
+# value that would itself cause problems.
+_MIN_TIMEOUT_S = 5
+_MAX_TIMEOUT_S = 900
+_MIN_RESULT_CAP = 100
+_MAX_RESULT_CAP = 100_000
+
+_ENV_TIMEOUT = "ONTOBRICKS_GRAPH_QUERY_TIMEOUT_S"
+_ENV_RESULT_CAP = "ONTOBRICKS_GRAPH_CHAT_RESULT_CAP"
+
+_lock = threading.Lock()
+_override_timeout_s: Optional[int] = None
+_override_result_cap: Optional[int] = None
+
+
+def _env_int(name: str) -> Optional[int]:
+ raw = os.getenv(name, "").strip()
+ if not raw:
+ return None
+ try:
+ value = int(raw)
+ except ValueError:
+ logger.warning("Ignoring non-integer %s=%r", name, raw)
+ return None
+ return value if value > 0 else None
+
+
+def _clamp(value: int, lo: int, hi: int) -> int:
+ return max(lo, min(int(value), hi))
+
+
+def get_graph_query_timeout_s() -> int:
+ """Return the effective graph-read statement timeout in seconds."""
+ with _lock:
+ override = _override_timeout_s
+ if override is not None:
+ return override
+ env = _env_int(_ENV_TIMEOUT)
+ if env is not None:
+ return _clamp(env, _MIN_TIMEOUT_S, _MAX_TIMEOUT_S)
+ return DEFAULT_GRAPH_QUERY_TIMEOUT_S
+
+
+def set_graph_query_timeout_override(seconds: Optional[int]) -> None:
+ """Set (or clear, when ``seconds`` is falsy) the admin timeout override."""
+ global _override_timeout_s
+ with _lock:
+ if not seconds:
+ _override_timeout_s = None
+ else:
+ _override_timeout_s = _clamp(int(seconds), _MIN_TIMEOUT_S, _MAX_TIMEOUT_S)
+
+
+def get_graph_chat_result_cap() -> int:
+ """Return the effective hard cap on triples returned to the chat agent."""
+ with _lock:
+ override = _override_result_cap
+ if override is not None:
+ return override
+ env = _env_int(_ENV_RESULT_CAP)
+ if env is not None:
+ return _clamp(env, _MIN_RESULT_CAP, _MAX_RESULT_CAP)
+ return DEFAULT_GRAPH_CHAT_RESULT_CAP
+
+
+def set_graph_chat_result_cap_override(count: Optional[int]) -> None:
+ """Set (or clear, when ``count`` is falsy) the admin result-cap override."""
+ global _override_result_cap
+ with _lock:
+ if not count:
+ _override_result_cap = None
+ else:
+ _override_result_cap = _clamp(int(count), _MIN_RESULT_CAP, _MAX_RESULT_CAP)
diff --git a/src/back/objects/domain/SettingsService.py b/src/back/objects/domain/SettingsService.py
index f8a650e1..1027559d 100644
--- a/src/back/objects/domain/SettingsService.py
+++ b/src/back/objects/domain/SettingsService.py
@@ -1405,6 +1405,64 @@ def save_registry_cache_ttl_result(
raise InfrastructureError("Failed to save registry cache TTL", detail=msg)
return {"success": True, "registry_cache_ttl": max(10, int(ttl))}
+ @staticmethod
+ def get_graph_limits_result(
+ session_mgr: SessionManager,
+ settings: Settings,
+ ) -> Dict[str, Any]:
+ """Return the effective graph-read bounds for the Settings UI.
+
+ ``graph_query_timeout_s`` bounds a single graph read (Lakebase /
+ warehouse ``statement_timeout``); ``graph_chat_result_cap`` bounds the
+ triples returned to the Graph Chat agent. Both resolve admin override →
+ env var → built-in default.
+ """
+ _, host, token, registry_cfg = SettingsService._resolve_context(
+ session_mgr, settings
+ )
+ return {
+ "success": True,
+ "graph_query_timeout_s": global_config_service.get_graph_query_timeout_s(
+ host, token, registry_cfg
+ ),
+ "graph_chat_result_cap": global_config_service.get_graph_chat_result_cap(
+ host, token, registry_cfg
+ ),
+ }
+
+ @staticmethod
+ def save_graph_limits_result(
+ graph_query_timeout_s: Optional[int],
+ graph_chat_result_cap: Optional[int],
+ email: str,
+ user_token: str,
+ session_mgr: SessionManager,
+ settings: Settings,
+ ) -> Dict[str, Any]:
+ """Persist admin-set graph-read bounds (``0``/``None`` = unset)."""
+ SettingsService.require_admin_error(email, user_token, session_mgr, settings)
+
+ _, host, token, registry_cfg = SettingsService._resolve_context(
+ session_mgr, settings
+ )
+ if graph_query_timeout_s is not None:
+ ok, msg = global_config_service.set_graph_query_timeout_s(
+ host, token, registry_cfg, int(graph_query_timeout_s)
+ )
+ if not ok:
+ raise InfrastructureError(
+ "Failed to save graph query timeout", detail=msg
+ )
+ if graph_chat_result_cap is not None:
+ ok, msg = global_config_service.set_graph_chat_result_cap(
+ host, token, registry_cfg, int(graph_chat_result_cap)
+ )
+ if not ok:
+ raise InfrastructureError(
+ "Failed to save graph chat result cap", detail=msg
+ )
+ return SettingsService.get_graph_limits_result(session_mgr, settings)
+
@staticmethod
def get_edit_lock_ttl_result(
session_mgr: SessionManager,
diff --git a/src/back/objects/session/GlobalConfigService.py b/src/back/objects/session/GlobalConfigService.py
index f67ba457..aa26cd47 100644
--- a/src/back/objects/session/GlobalConfigService.py
+++ b/src/back/objects/session/GlobalConfigService.py
@@ -19,6 +19,12 @@
from typing import Any, Dict, Optional, Tuple
from back.core.logging import get_logger
+from back.core.query_limits import (
+ get_graph_chat_result_cap as _effective_result_cap,
+ get_graph_query_timeout_s as _effective_timeout_s,
+ set_graph_chat_result_cap_override,
+ set_graph_query_timeout_override,
+)
from back.objects.registry.registry_cache import set_registry_cache_ttl
logger = get_logger(__name__)
@@ -91,6 +97,17 @@ def load(
self._cache_ts = now
if "registry_cache_ttl" in data:
set_registry_cache_ttl(int(data["registry_cache_ttl"]))
+ # Apply the persisted graph-read bounds so admin overrides
+ # survive a cold restart (0 / unset clears the override, so
+ # the env var / built-in default applies).
+ if "graph_query_timeout_s" in data:
+ set_graph_query_timeout_override(
+ int(data["graph_query_timeout_s"] or 0) or None
+ )
+ if "graph_chat_result_cap" in data:
+ set_graph_chat_result_cap_override(
+ int(data["graph_chat_result_cap"] or 0) or None
+ )
logger.info(
"Loaded global config (backend=%s)", store.backend
)
@@ -365,6 +382,70 @@ def set_registry_cache_ttl(
set_registry_cache_ttl(ttl)
return self._save(host, token, registry_cfg, {"registry_cache_ttl": ttl})
+ def get_graph_query_timeout_s(
+ self, host: str, token: str, registry_cfg: Dict[str, str]
+ ) -> int:
+ """Return the effective graph-read statement timeout (seconds).
+
+ The persisted admin value is re-applied through the central clamp in
+ :mod:`back.core.query_limits` so the Settings UI shows the same
+ (bounded) value the database actually enforces — never a stale
+ out-of-range number.
+ """
+ val = self.get(host, token, registry_cfg, "graph_query_timeout_s", "")
+ s = str(val).strip() if val is not None else ""
+ if s.isdigit():
+ set_graph_query_timeout_override(int(s) or None)
+ return _effective_timeout_s()
+
+ def set_graph_query_timeout_s(
+ self,
+ host: str,
+ token: str,
+ registry_cfg: Dict[str, str],
+ seconds: int,
+ ) -> Tuple[bool, str]:
+ """Persist and apply the graph-read statement timeout (``0`` = unset)."""
+ seconds = max(0, int(seconds))
+ set_graph_query_timeout_override(seconds or None)
+ # Persist the clamped effective value (not the raw input) so config and
+ # the Settings UI can never show a timeout the database won't honour.
+ to_save = 0 if seconds <= 0 else _effective_timeout_s()
+ return self._save(
+ host, token, registry_cfg, {"graph_query_timeout_s": to_save}
+ )
+
+ def get_graph_chat_result_cap(
+ self, host: str, token: str, registry_cfg: Dict[str, str]
+ ) -> int:
+ """Return the effective Graph Chat triple result cap.
+
+ Re-applies the persisted admin value through the central clamp so the
+ Settings UI shows the same bounded cap that is actually enforced.
+ """
+ val = self.get(host, token, registry_cfg, "graph_chat_result_cap", "")
+ s = str(val).strip() if val is not None else ""
+ if s.isdigit():
+ set_graph_chat_result_cap_override(int(s) or None)
+ return _effective_result_cap()
+
+ def set_graph_chat_result_cap(
+ self,
+ host: str,
+ token: str,
+ registry_cfg: Dict[str, str],
+ count: int,
+ ) -> Tuple[bool, str]:
+ """Persist and apply the Graph Chat triple result cap (``0`` = unset)."""
+ count = max(0, int(count))
+ set_graph_chat_result_cap_override(count or None)
+ # Persist the clamped effective value so config/UI stay consistent with
+ # the enforced cap.
+ to_save = 0 if count <= 0 else _effective_result_cap()
+ return self._save(
+ host, token, registry_cfg, {"graph_chat_result_cap": to_save}
+ )
+
def get_edit_lock_ttl_s(
self, host: str, token: str, registry_cfg: Dict[str, str]
) -> Optional[int]:
@@ -449,6 +530,9 @@ def _empty() -> Dict[str, Any]:
"navbar_logo": "",
"use_cloud_fetch": True,
"registry_cache_ttl": 300,
+ # 0 = unset → env var / built-in default from back.core.query_limits.
+ "graph_query_timeout_s": 0,
+ "graph_chat_result_cap": 0,
"graph_engine_config": {},
}
diff --git a/src/front/static/config/js/settings.js b/src/front/static/config/js/settings.js
index 6a768f2b..51f7ebe8 100644
--- a/src/front/static/config/js/settings.js
+++ b/src/front/static/config/js/settings.js
@@ -1319,6 +1319,28 @@ document.addEventListener('DOMContentLoaded', function () {
console.log('Graph DB heavy refresh failed', e);
} finally {
applyGraphDbEnginePanels();
+ await loadGraphLimits();
+ }
+ }
+
+ /** Populate the graph-read bound inputs (statement timeout + chat cap). */
+ async function loadGraphLimits() {
+ const timeoutEl = document.getElementById('graphQueryTimeoutS');
+ const capEl = document.getElementById('graphChatResultCap');
+ if (!timeoutEl && !capEl) return;
+ try {
+ const resp = await fetch('/settings/graph-limits', { credentials: 'same-origin' });
+ if (!resp.ok) return;
+ const data = await resp.json();
+ if (!data || !data.success) return;
+ if (timeoutEl && typeof data.graph_query_timeout_s === 'number') {
+ timeoutEl.value = String(data.graph_query_timeout_s);
+ }
+ if (capEl && typeof data.graph_chat_result_cap === 'number') {
+ capEl.value = String(data.graph_chat_result_cap);
+ }
+ } catch (e) {
+ console.log('Graph limits load failed', e);
}
}
@@ -3512,7 +3534,33 @@ document.addEventListener('DOMContentLoaded', function () {
}
}
- // 3c. Save the Databricks graph-analytics job toggle. An explicit "off"
+ // 3c. Save graph-read bounds (statement timeout + chat result cap;
+ // 0/blank leaves the env-var / built-in default in force).
+ const gTimeoutInput = document.getElementById('graphQueryTimeoutS');
+ const gCapInput = document.getElementById('graphChatResultCap');
+ if (gTimeoutInput || gCapInput) {
+ const body = {};
+ if (gTimeoutInput) {
+ const v = parseInt(gTimeoutInput.value, 10);
+ body.graph_query_timeout_s = isNaN(v) ? 0 : Math.max(0, v);
+ }
+ if (gCapInput) {
+ const v = parseInt(gCapInput.value, 10);
+ body.graph_chat_result_cap = isNaN(v) ? 0 : Math.max(0, v);
+ }
+ try {
+ const resp = await fetch('/settings/save-graph-limits', {
+ method: 'POST',
+ headers: { 'Content-Type': 'application/json' },
+ credentials: 'same-origin',
+ body: JSON.stringify(body)
+ });
+ const r = await resp.json();
+ if (!r.success) errors.push('Graph limits: ' + (r.message || 'save failed'));
+ } catch (e) { errors.push('Graph limits: ' + e.message); }
+ }
+
+ // 3d. Save the Databricks graph-analytics job toggle. An explicit "off"
// is sent as readily as an "on", because unchecking has to persist a
// value that overrides the env-var default — but only once the checkbox
// is known to hold the stored state. Posting an unhydrated box would
diff --git a/src/front/static/query/js/query-chat.js b/src/front/static/query/js/query-chat.js
index 9c88b389..5c58a632 100644
--- a/src/front/static/query/js/query-chat.js
+++ b/src/front/static/query/js/query-chat.js
@@ -305,6 +305,41 @@
container.scrollTop = container.scrollHeight;
}
+ /**
+ * Inline advisory shown in the chat when the server reports resource
+ * pressure (blocking thread pool saturated). Nudges the user toward a
+ * larger Databricks App instance instead of appearing to hang.
+ */
+ function appendResourceAdvisory(text) {
+ const container = messagesEl();
+ if (!container) return;
+ const div = document.createElement('div');
+ div.className = 'alert alert-warning py-2 px-3 small graph-chat-resource-advisory d-flex align-items-start';
+ const icon = document.createElement('i');
+ icon.className = 'bi bi-exclamation-triangle-fill me-2 mt-1';
+ const span = document.createElement('span');
+ span.textContent = text;
+ div.appendChild(icon);
+ div.appendChild(span);
+ container.appendChild(div);
+ container.scrollTop = container.scrollHeight;
+ }
+
+ /**
+ * Surface the resource-pressure advisory both inline (in the chat) and via
+ * the global NotificationCenter toast, when the server flags it.
+ */
+ function maybeShowResourcePressure(event) {
+ if (!event || !event.resource_pressure) return;
+ const msg = event.resource_advice ||
+ 'OntoBricks is under heavy load; responses may be slow. Consider ' +
+ 'upgrading the Databricks App instance size.';
+ appendResourceAdvisory(msg);
+ if (typeof window.showNotification === 'function') {
+ try { window.showNotification(msg, 'warning'); } catch (_) { /* best effort */ }
+ }
+ }
+
function showThinking() {
const container = messagesEl();
if (!container) return;
@@ -515,6 +550,7 @@
role: 'assistant',
content: reply,
});
+ maybeShowResourcePressure(doneEvent);
} else {
errorStreamingBubble(bubble, 'Stream ended without a final response.');
}
diff --git a/src/front/templates/settings.html b/src/front/templates/settings.html
index b40ae044..fa452882 100644
--- a/src/front/templates/settings.html
+++ b/src/front/templates/settings.html
@@ -923,6 +923,36 @@
Global
+
+
+
+
+
+
+
+ Max time a single graph read (e.g. a Graph Chat traversal) may run before the
+ database cancels it — applies to both Lakebase and the SQL warehouse. 0
+ uses the built-in default. Prevents a broad query from freezing the app.
+
+
+
+
+
+
+ Hard ceiling on triples returned to the Graph Chat agent per query.
+ 0 uses the built-in default.
+
+
+
+
+
diff --git a/tests/units/core/test_graph_query_bounds.py b/tests/units/core/test_graph_query_bounds.py
new file mode 100644
index 00000000..75b6b5c5
--- /dev/null
+++ b/tests/units/core/test_graph_query_bounds.py
@@ -0,0 +1,181 @@
+"""Regression tests for the Graph Chat event-loop-hang fix.
+
+Covers the two guards that stop a broad Graph Chat query from freezing the app:
+
+* the SQL warehouse graph-read path issues (and resets) a session
+ ``STATEMENT_TIMEOUT`` when bounded, and never touches it otherwise;
+* the blocking thread pool auto-sizes from the instance's vCPU count and reports
+ saturation for the resource-pressure advisory.
+"""
+
+import importlib
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+
+from back.core.databricks.DatabricksAuth import DatabricksAuth
+from back.core.databricks.SQLWarehouse import SQLWarehouse
+
+# Import the *module* explicitly: the ``back.core.helpers`` package re-exports a
+# ``DatabricksHelpers`` class under the same name, which otherwise shadows the
+# submodule when accessed as an attribute.
+dh = importlib.import_module("back.core.helpers.DatabricksHelpers")
+
+
+def _make_conn(description, rows):
+ cur = MagicMock()
+ cur.description = description
+ cur.fetchall.return_value = rows
+ conn = MagicMock()
+ conn.__enter__ = Mock(return_value=conn)
+ conn.__exit__ = Mock(return_value=False)
+ conn.cursor.return_value.__enter__ = Mock(return_value=cur)
+ conn.cursor.return_value.__exit__ = Mock(return_value=False)
+ return conn, cur
+
+
+def _executed(cur):
+ return [str(c.args[0]) for c in cur.execute.call_args_list]
+
+
+class TestWarehouseStatementTimeout:
+ @patch("databricks.sql.connect")
+ def test_bounded_sets_and_resets_timeout(self, mock_connect, monkeypatch):
+ monkeypatch.delenv("DATABRICKS_APP_PORT", raising=False)
+ conn, cur = _make_conn([("id",)], [(1,)])
+ mock_connect.return_value = conn
+ sw = SQLWarehouse(
+ DatabricksAuth(host="https://h.databricks.com", token="t", warehouse_id="wh-1")
+ )
+
+ rows = sw.execute_query("SELECT id FROM t", statement_timeout_s=45)
+
+ assert rows == [{"id": 1}]
+ stmts = _executed(cur)
+ assert any("STATEMENT_TIMEOUT = 45" in s for s in stmts)
+ # Reset to 0 so the pooled connection doesn't leak the bound.
+ assert any("STATEMENT_TIMEOUT = 0" in s for s in stmts)
+
+ @patch("databricks.sql.connect")
+ def test_unbounded_leaves_timeout_untouched(self, mock_connect, monkeypatch):
+ monkeypatch.delenv("DATABRICKS_APP_PORT", raising=False)
+ conn, cur = _make_conn([("id",)], [(1,)])
+ mock_connect.return_value = conn
+ sw = SQLWarehouse(
+ DatabricksAuth(host="https://h.databricks.com", token="t", warehouse_id="wh-1")
+ )
+
+ sw.execute_query("SELECT id FROM t")
+
+ assert all("STATEMENT_TIMEOUT" not in s for s in _executed(cur))
+
+
+class TestDeltaStoreBoundsReads:
+ def test_execute_query_passes_timeout_to_warehouse(self, monkeypatch):
+ from back.core.graphdb.delta.DeltaFlatStore import DeltaFlatStore
+
+ monkeypatch.setattr(
+ "back.core.query_limits.get_graph_query_timeout_s", lambda: 42
+ )
+ sql_service = MagicMock()
+ sql_service.execute_query.return_value = [{"n": 1}]
+ client = MagicMock()
+ client.sql = sql_service
+
+ store = object.__new__(DeltaFlatStore)
+ store._client = client
+ out = store.execute_query("SELECT 1")
+
+ assert out == [{"n": 1}]
+ sql_service.execute_query.assert_called_once_with(
+ "SELECT 1", statement_timeout_s=42
+ )
+
+
+class TestLakebaseStatementTimeoutReset:
+ """The pool hands out ``autocommit=True`` connections, so a read's
+ ``statement_timeout`` must be reset or it leaks to the next borrower
+ (a bulk write/DDL) and cancels it mid-flight."""
+
+ def _store(self, cur):
+ from back.core.graphdb.lakebase.LakebaseFlatStore import LakebaseFlatStore
+
+ store = object.__new__(LakebaseFlatStore)
+ store._schema = "s"
+ conn = MagicMock()
+ conn.__enter__ = Mock(return_value=conn)
+ conn.__exit__ = Mock(return_value=False)
+ conn.cursor.return_value.__enter__ = Mock(return_value=cur)
+ conn.cursor.return_value.__exit__ = Mock(return_value=False)
+ pool = MagicMock()
+ pool.connection.return_value = conn
+ store._pool = lambda: pool
+ store._require_pg = lambda: (None, dict)
+ return store
+
+ def test_resets_after_success(self, monkeypatch):
+ monkeypatch.setattr(
+ "back.core.query_limits.get_graph_query_timeout_s", lambda: 30
+ )
+ cur = MagicMock()
+ cur.description = [("s",)]
+ cur.fetchall.return_value = [{"s": "x"}]
+
+ out = self._store(cur).execute_query("SELECT 1")
+
+ assert out == [{"s": "x"}]
+ stmts = _executed(cur)
+ assert any("SET statement_timeout = 30000" in s for s in stmts)
+ assert stmts[-1] == "RESET statement_timeout"
+
+ def test_resets_even_when_query_raises(self, monkeypatch):
+ monkeypatch.setattr(
+ "back.core.query_limits.get_graph_query_timeout_s", lambda: 30
+ )
+ cur = MagicMock()
+ cur.description = None
+
+ def _exec(sql, *a, **k):
+ if sql == "BOOM":
+ raise RuntimeError("cancelled by statement_timeout")
+
+ cur.execute.side_effect = _exec
+
+ with pytest.raises(RuntimeError):
+ self._store(cur).execute_query("BOOM")
+
+ assert "RESET statement_timeout" in _executed(cur)
+
+
+class TestBlockingPoolAutoTune:
+ def test_explicit_env_size_wins(self, monkeypatch):
+ monkeypatch.setenv("ONTOBRICKS_THREAD_POOL_SIZE", "7")
+ assert dh._resolve_blocking_pool_size() == 7
+
+ def test_derives_from_cpu_count(self, monkeypatch):
+ monkeypatch.delenv("ONTOBRICKS_THREAD_POOL_SIZE", raising=False)
+ monkeypatch.setattr(dh.os, "cpu_count", lambda: 16)
+ assert dh._resolve_blocking_pool_size() == 64
+
+ def test_floored_at_minimum(self, monkeypatch):
+ monkeypatch.delenv("ONTOBRICKS_THREAD_POOL_SIZE", raising=False)
+ monkeypatch.setattr(dh.os, "cpu_count", lambda: 1)
+ assert dh._resolve_blocking_pool_size() == dh._BLOCKING_POOL_MIN
+
+ def test_invalid_env_falls_back_to_cpu(self, monkeypatch):
+ monkeypatch.setenv("ONTOBRICKS_THREAD_POOL_SIZE", "abc")
+ monkeypatch.setattr(dh.os, "cpu_count", lambda: 8)
+ assert dh._resolve_blocking_pool_size() == 32
+
+
+class TestBlockingPoolStats:
+ def test_stats_shape_and_idle(self, monkeypatch):
+ monkeypatch.setattr(dh, "_inflight_blocking", 0)
+ stats = dh.get_blocking_pool_stats()
+ assert set(stats) >= {"max_workers", "active", "peak", "saturated"}
+ assert stats["active"] == 0
+ assert stats["saturated"] is False
+
+ def test_saturation_flag(self, monkeypatch):
+ monkeypatch.setattr(dh, "_inflight_blocking", dh._BLOCKING_POOL_SIZE)
+ assert dh.get_blocking_pool_stats()["saturated"] is True
diff --git a/tests/units/core/test_query_limits.py b/tests/units/core/test_query_limits.py
new file mode 100644
index 00000000..71d7b975
--- /dev/null
+++ b/tests/units/core/test_query_limits.py
@@ -0,0 +1,82 @@
+"""Unit tests for graph read-query bounds (:mod:`back.core.query_limits`).
+
+Pins the resolution order (admin override > env var > built-in default) and the
+clamping that keeps a misconfiguration from disabling the guard. These bounds
+are what stop a broad Graph Chat traversal from freezing the app.
+"""
+
+import back.core.query_limits as ql
+
+
+def _reset_overrides():
+ ql.set_graph_query_timeout_override(None)
+ ql.set_graph_chat_result_cap_override(None)
+
+
+class TestGraphQueryTimeout:
+ def test_builtin_default_when_unset(self, monkeypatch):
+ monkeypatch.delenv("ONTOBRICKS_GRAPH_QUERY_TIMEOUT_S", raising=False)
+ _reset_overrides()
+ assert ql.get_graph_query_timeout_s() == ql.DEFAULT_GRAPH_QUERY_TIMEOUT_S
+
+ def test_env_var_used_when_no_override(self, monkeypatch):
+ _reset_overrides()
+ monkeypatch.setenv("ONTOBRICKS_GRAPH_QUERY_TIMEOUT_S", "90")
+ assert ql.get_graph_query_timeout_s() == 90
+
+ def test_admin_override_beats_env(self, monkeypatch):
+ monkeypatch.setenv("ONTOBRICKS_GRAPH_QUERY_TIMEOUT_S", "90")
+ ql.set_graph_query_timeout_override(120)
+ try:
+ assert ql.get_graph_query_timeout_s() == 120
+ finally:
+ _reset_overrides()
+
+ def test_override_clamped_to_bounds(self, monkeypatch):
+ monkeypatch.delenv("ONTOBRICKS_GRAPH_QUERY_TIMEOUT_S", raising=False)
+ try:
+ ql.set_graph_query_timeout_override(10_000_000)
+ assert ql.get_graph_query_timeout_s() == ql._MAX_TIMEOUT_S
+ ql.set_graph_query_timeout_override(1)
+ assert ql.get_graph_query_timeout_s() == ql._MIN_TIMEOUT_S
+ finally:
+ _reset_overrides()
+
+ def test_zero_override_clears(self, monkeypatch):
+ monkeypatch.delenv("ONTOBRICKS_GRAPH_QUERY_TIMEOUT_S", raising=False)
+ ql.set_graph_query_timeout_override(120)
+ ql.set_graph_query_timeout_override(0)
+ try:
+ assert ql.get_graph_query_timeout_s() == ql.DEFAULT_GRAPH_QUERY_TIMEOUT_S
+ finally:
+ _reset_overrides()
+
+ def test_non_integer_env_ignored(self, monkeypatch):
+ _reset_overrides()
+ monkeypatch.setenv("ONTOBRICKS_GRAPH_QUERY_TIMEOUT_S", "not-a-number")
+ assert ql.get_graph_query_timeout_s() == ql.DEFAULT_GRAPH_QUERY_TIMEOUT_S
+
+
+class TestGraphChatResultCap:
+ def test_builtin_default_when_unset(self, monkeypatch):
+ monkeypatch.delenv("ONTOBRICKS_GRAPH_CHAT_RESULT_CAP", raising=False)
+ _reset_overrides()
+ assert ql.get_graph_chat_result_cap() == ql.DEFAULT_GRAPH_CHAT_RESULT_CAP
+
+ def test_admin_override_beats_env(self, monkeypatch):
+ monkeypatch.setenv("ONTOBRICKS_GRAPH_CHAT_RESULT_CAP", "5000")
+ ql.set_graph_chat_result_cap_override(2500)
+ try:
+ assert ql.get_graph_chat_result_cap() == 2500
+ finally:
+ _reset_overrides()
+
+ def test_override_clamped_to_bounds(self, monkeypatch):
+ monkeypatch.delenv("ONTOBRICKS_GRAPH_CHAT_RESULT_CAP", raising=False)
+ try:
+ ql.set_graph_chat_result_cap_override(10_000_000)
+ assert ql.get_graph_chat_result_cap() == ql._MAX_RESULT_CAP
+ ql.set_graph_chat_result_cap_override(1)
+ assert ql.get_graph_chat_result_cap() == ql._MIN_RESULT_CAP
+ finally:
+ _reset_overrides()