fix+feat(nexus/graph): cross-graph Composer traversal + query result caching (#1059)

* fix(nexus/graph): resolve cross-graph traversal in the Composer query compiler

A multi-hop column path that crosses a named-graph boundary (e.g. an ExtractedItem
in one graph -> chunk_of -> PDFPaperFile in another) returned empty cells: the whole
WHERE body was wrapped in a single `GRAPH ?g { ... }`, which binds one graph for every
triple, so no single graph satisfies the path.

Pick the scoping strategy per query:
- 1 graph selected  -> the original single `GRAPH ?g` form (unchanged, fastest).
- >1 graph selected -> each triple is wrapped in its own `VALUES ?gK GRAPH ?gK`, so a
  row's path may span graphs. Per-triple GRAPH keeps Jena/TDB2 on its native quad index;
  a `FROM`-union default graph is also correct but ~10x slower on TDB2 (dynamic dataset).

Single-graph output is byte-identical (all golden tests unchanged). Adds multi-graph
unit + integration coverage.

* feat(nexus/graph): cache Composer query results with a bypass toggle

Executing the page SPARQL is the expensive part of a Composer query (seconds on large
cross-graph views) and the data changes infrequently, so memoize it.

Backend: GraphQueryService caches the page rows (new `page_cache_key`, keyed on the full
spec + workspace + page window) and the count, via the existing naas-abi-core CacheService
(FS tier -> shared across workers, survives restart). TTL 5 min, override with env
`NEXUS_QUERY_CACHE_TTL_SECONDS`. A request `force_refresh` flag bypasses both caches
(superset of `force_count_refresh`). Cache failures degrade to a live query.

Frontend: `forceRefresh` in the query client/types, a persisted `bypassCache` toggle in
use-explore surfaced as an "Always refresh" checkbox in the Composer toolbar; the Refresh
button now always fetches fresh.

Adds service tests: cache hit skips the store, force_refresh bypasses, no-cache default
recomputes, distinct page key per sort.

* feat(nexus/graph): cache column discovery (the "add column" dropdown)

Column discovery fires ~5 SPARQL queries per call (datatype props, relations, relation
target classes, incoming relations + their source classes) and was uncached — the main
Composer latency when picking a grain / adding a column. The discovered columns change
rarely, so memoize them.

GraphQueryService.discover_columns now caches its result via the same CacheService (new
`columns_cache_key` over workspace + grain graphs + class URIs + type-resolution graphs),
using the injected `columns_cache` (5-min TTL, shared with the query/count cache). Cache
miss/failure falls through to a live discovery. DiscoveredColumn/TargetClassData round-trip
through JSON. Adds cache-hit + serialization-round-trip tests.

* refactor(nexus/graph): read query-cache TTL from Settings, not raw env

The cache TTL was read with a bare `os.environ.get("NEXUS_QUERY_CACHE_TTL_SECONDS")`,
bypassing the app's pydantic-settings config (and that var was never set anywhere, so it
was effectively a hardcoded 300). Move it to `Settings.graph_query_cache_ttl_seconds`
(env `GRAPH_QUERY_CACHE_TTL_SECONDS`, default 300) like every other tunable.

Also make `0` genuinely disable caching: `timedelta(0)` is falsy, so the CacheService would
treat it as "no expiry" (infinite cache) — instead, TTL <= 0 now injects no cache at all.

* refactor(nexus/graph): back the query cache with the engine's multi-tier CacheService

Use the engine's configured CacheService (`ABIModule.get_instance().engine.services.cache`)
instead of a one-off FS cache, so the Composer query/column cache benefits from the hot
(Redis) + cold (FS/object-storage) tiers. `_resolve_query_cache()` prefers the engine cache
and falls back to a local FS cache if none is available; None when TTL <= 0.

Reads are a hot→cold read-through; writes go to cold (durable) AND, when a hot tier exists,
to hot (fast subsequent reads) — the CacheService's default set_json is cold-only, so hot is
populated explicitly. (Deployments enable the hot tier by adding a redis adapter under
`services.cache` in config; without it the engine cache defaults to a single cold FS tier.)

Author: Dr0p42

libs/naas-abi/naas_abi/apps/nexus/apps/api/app/core/config.py

@@ -346,6 +346,10 @@ class Settings(BaseSettings):
     frontend_url: str = "http://localhost:3000"
     websocket_path: str = "/ws/socket.io"
 
+    # Graph (Composer) query cache: TTL for cached page rows / count / column discovery.
+    # Env: GRAPH_QUERY_CACHE_TTL_SECONDS. 0 disables caching (always live).
+    graph_query_cache_ttl_seconds: int = 300
+
     # Coding workspaces (Coder editor + Forgejo monorepo auto-clone). clone
     # host/scheme are what a *workspace container* uses to reach Forgejo (not the
     # admin API URL); docker_network is the network the workspace must join to

libs/naas-abi/naas_abi/apps/nexus/apps/api/app/services/graph/adapters/primary/graph__primary_adapter__FastAPI.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import datetime
 import io
 
 from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile
@@ -11,6 +12,7 @@
     get_current_user_required,
     require_workspace_access,
 )
+from naas_abi.apps.nexus.apps.api.app.core.config import settings
 from naas_abi.apps.nexus.apps.api.app.services.graph.adapters.primary.graph__primary_adapter__dependencies import (  # noqa: E501
     get_graph_service,
 )
@@ -83,16 +85,77 @@
     GraphQueryTripleStoreAdapter,
     resolve_fts_backend,
 )
-from naas_abi.apps.nexus.apps.api.app.services.graph.query.service import GraphQueryService
+from naas_abi.apps.nexus.apps.api.app.services.graph.query.service import (
+    CountCache,
+    GraphQueryService,
+)
 from naas_abi.apps.nexus.apps.api.app.services.graph.service import (
     NEXUS_GRAPH_URI,
     SCHEMA_GRAPH_URI,
     GraphService,
     _detect_rdf_format,
 )
+from naas_abi_core.services.cache.CacheFactory import CacheFactory
+from naas_abi_core.services.cache.CachePort import CacheExpiredError, CacheNotFoundError
+from naas_abi_core.services.cache.CacheService import CacheService
 
 router = APIRouter(dependencies=[Depends(get_current_user_required)])
 
+# Composer query result cache (page rows + count + column discovery). Prefer the engine's
+# multi-tier CacheService (hot Redis + cold FS/object-storage) resolved per request via
+# `_resolve_query_cache()`; if the engine has no cache configured we fall back to a local
+# FS cache. The TTL bounds staleness and the Composer's "always refresh" tick (`force_refresh`)
+# bypasses it. TTL is Settings-driven (`graph_query_cache_ttl_seconds`, env
+# GRAPH_QUERY_CACHE_TTL_SECONDS); 0 disables caching entirely.
+_FALLBACK_QUERY_CACHE = CacheFactory.CacheFS_find_storage(subpath="nexus/graph-query")
+_QUERY_CACHE_TTL_SECONDS = settings.graph_query_cache_ttl_seconds
+
+
+class _QueryResultCache(CountCache):
+    """Adapts a naas-abi-core CacheService to the query service's ``fetch``/``store`` port.
+
+    Reads are a hot→cold read-through (``get``). Writes go to the cold tier (durable) AND, when
+    a hot tier exists, to hot (fast subsequent reads) — the CacheService's default ``set_json``
+    is cold-only, so we populate hot explicitly. A cache failure never breaks a query.
+    """
+
+    def __init__(self, cache: CacheService, ttl: datetime.timedelta) -> None:
+        self._cache = cache
+        self._ttl = ttl
+
+    def fetch(self, key: str) -> dict | None:
+        try:
+            return self._cache.get(key, ttl=self._ttl)  # hot → cold
+        except (CacheNotFoundError, CacheExpiredError):
+            return None
+        except Exception:  # noqa: BLE001 - cache is best-effort, degrade to a live query
+            return None
+
+    def store(self, key: str, value: dict) -> None:
+        try:
+            self._cache.set_json(key, value)  # cold tier (durable)
+            if self._cache.hot_available():
+                self._cache.hot.set_json(key, value)  # hot tier (fast reads)
+        except Exception:  # noqa: BLE001
+            pass
+
+
+def _resolve_query_cache() -> _QueryResultCache | None:
+    """The cache to back a query service. ``None`` disables caching (TTL ≤ 0). Prefers the
+    engine's configured (multi-tier) CacheService; falls back to a local FS cache."""
+    if _QUERY_CACHE_TTL_SECONDS <= 0:
+        return None
+    ttl = datetime.timedelta(seconds=_QUERY_CACHE_TTL_SECONDS)
+    try:
+        from naas_abi import ABIModule
+
+        services = ABIModule.get_instance().engine.services
+        if services.cache_available():
+            return _QueryResultCache(services.cache, ttl)
+    except Exception:  # noqa: BLE001 - engine/cache not available → fall back to local FS
+        pass
+    return _QueryResultCache(_FALLBACK_QUERY_CACHE, ttl)
+
 
 class GraphFastAPIPrimaryAdapter:
     def __init__(self) -> None:
@@ -1149,7 +1212,13 @@ async def _owned_graphs(workspace_id: str) -> set[str]:
 
     # schema/nexus are global system graphs every workspace may read.
     system_graphs = {str(SCHEMA_GRAPH_URI), str(NEXUS_GRAPH_URI)}
-    return GraphQueryService(store, owned_graphs=_owned_graphs, system_graphs=system_graphs)
+    # Engine's multi-tier cache (hot Redis + cold FS/object-storage) when available, else a
+    # local FS fallback; None when TTL <= 0 (caching disabled → the service's no-op caches).
+    cache = _resolve_query_cache()
+    return GraphQueryService(
+        store, owned_graphs=_owned_graphs, system_graphs=system_graphs,
+        count_cache=cache, page_cache=cache, columns_cache=cache,
+    )
 
 
 @router.post("/query")
@@ -1169,6 +1238,7 @@ async def graph_query(
             limit=payload.limit,
             include_sparql=payload.include_sparql,
             force_count_refresh=payload.force_count_refresh,
+            force_refresh=payload.force_refresh,
         )
     except GraphAccessError as exc:  # workspace does not own a requested graph
         raise HTTPException(status_code=403, detail=str(exc)) from exc

libs/naas-abi/naas_abi/apps/nexus/apps/api/app/services/graph/query/adapters/primary/graph_query__primary_adapter__schemas.py

@@ -244,6 +244,8 @@ class GraphQueryRequest(BaseModel):
     limit: int | None = Field(default=None, ge=1, le=5000)
     include_sparql: bool = False
     force_count_refresh: bool = False
+    # "Always refresh" tick in the Composer: bypass the result cache and re-run the SPARQL.
+    force_refresh: bool = False
 
 
 # ── Response ──────────────────────────────────────────────────────────────────