The query service now uses async_execute_paged (indexed path) and
async_scan (scan path) instead of async_execute. Tests were mocking
the old function, causing them to hang indefinitely.
- Pass auth token to schema discovery and descriptor generation in
tg-load-structured-data, fixing 401 errors with IAM enabled
- Fix row query pagination: replace single-page async_execute with
async_scan that streams pages and applies filters without
materialising the full result set (OOM on large datasets)
- Add missing filter operators (not, startsWith, endsWith, not_in)
to row query post-filter matching
- Fall back to scan path when an indexed field is queried with an
empty string value, since empty index values are not stored
- Revert top-level indexes array support — the current table schema
overwrites rows with duplicate index values, so only primary_key
fields are safe to index until the schema is redesigned
resolve_cassandra_config did not accept replication_factor as a kwarg,
so cassandra_replication_factor from YAML params was silently ignored
by all 6 callers. Add the kwarg and pass it from every caller.
Same fix for Qdrant: 3 writers now pass qdrant_replication_factor and
qdrant_shard_number from params.
Add tests covering the params path for both helpers.
- Add centralised qdrant_config.py helper with env-var fallback for
QDRANT_URL, QDRANT_API_KEY, QDRANT_REPLICATION_FACTOR, QDRANT_SHARD_NUMBER
- Update all 6 Qdrant processors to use the helper; writers pass
replication_factor and shard_number to create_collection
- Fix hardcoded Cassandra replication_factor=1 in cassandra_kg.py,
write.py, and sparql_cassandra.py to respect CASSANDRA_REPLICATION_FACTOR
- Upgrade Cassandra TLS from deprecated PROTOCOL_TLSv1_2 to
ssl.create_default_context() across all connectors
The librarian now reads OBJECT_STORE_ENDPOINT, OBJECT_STORE_ACCESS_KEY,
OBJECT_STORE_SECRET_KEY, OBJECT_STORE_REGION, and OBJECT_STORE_USE_SSL
from the environment when not set via params. This lets K8s Secrets
supply credentials without them appearing in launch.yaml.
Implements all three changes from the knowledge-core-completeness tech spec:
1. Named graph field preserved through Cassandra storage (7-element tuple),
enabling provenance triples to retain their graph URIs on round-trip.
2. Provenance triples already arrive on triples-input — no routing change
needed; Change 1 was sufficient.
3. Source material (library documents) streamed alongside triples and
embeddings during core download/upload. The knowledge manager fetches
the document hierarchy from the librarian on download and recreates it
on upload, preserving the full provenance chain across instances.
The mux unconditionally called auth.authorise() for every operation,
passing capability sentinels like AUTHENTICATED ("__authenticated__")
to the IAM regime. Since no role grants "__authenticated__", the regime
denied the request — breaking whoami (and any future AUTHENTICATED-only
operation) over the WebSocket path while the HTTP endpoints worked fine.
Match the guard pattern used by iam_endpoint.py and registry_endpoint.py:
only call authorise() for real capability strings, not sentinels.
Bulk clients (sync and async) were not forwarding the workspace parameter,
causing all bulk operations to hit the default workspace regardless of the
Api instance's workspace setting. Also fixes the gateway socket endpoint to
pass query parameters (including workspace) to the dispatcher, and prevents
the auth handshake from overwriting an explicitly set workspace.
Updates knowledge table store tests for paged query interface.
- Paginate heavy Cassandra reads (triples, graph/document embeddings)
using synchronous session.execute() in run_in_executor with fetch_size
paging, preventing materialization hang on large result sets
- Fix document stream endpoint to use workspace-scoped librarian queues
- Add decoder error handling for PDF/OCR/unstructured processors
- Add WebSocket mux guards for missing auth fields
- Add null check in librarian document streaming
- Rewrite get_document_content CLI to stream via librarian
- Add Poppler dependency to unstructured container
The tests were patching
trustgraph.embeddings.hf.hf.HuggingFaceEmbeddings - a module-level
attribute that doesn't exist because HuggingFaceEmbeddings is
imported locally inside _load_model. Changed all 8 occurrences to
patch langchain_huggingface.HuggingFaceEmbeddings, which is the
actual import source the code uses at runtime.
The auth-ok response includes the token's bound workspace, and
AsyncSocketClient was unconditionally adopting it — clobbering any
workspace the caller explicitly requested via the constructor.
Several CLI commands silently routed requests to the default workspace
regardless of the -w flag: show-flows, show-flow-blueprints,
show-parameter-types, set-prompt --system, and load-structured-data.
The workspace was sent in the inner request body but not on the
WebSocket envelope or API client constructor, so the gateway always
dispatched to the default workspace queue.
Add a new `list-my-workspaces` operation so non-admin users can
discover which workspaces they have access to. For OSS IAM, regular
users see their home workspace; admins see all workspaces.
Also add the full IAM service to both OpenAPI and AsyncAPI specs —
it was previously undocumented despite being a first-class service
on both HTTP and WebSocket interfaces.
Added click to the pip install line. Looks like huggingface-hub 1.13.0
added a CLI dependency on click but didn't declare it as a hard
requirement (or it's only required for the CLI entrypoint). Needed to
unblock the build.
Replaces the URL-based PDF downloads in tg-load-sample-documents with
seven curated, locally bundled documents covering diverse topics (recipes,
Belgian beer, trade routes, corporate scandals, pets, fortifications,
Bronze Age collapse). Documents are packaged as data files within
trustgraph-cli and loaded from metadata.json, removing the dependency
on external URLs and the doc-cache mechanism.
Replace removed `user` parameter with `workspace` support following
the tenancy axis change in #840. Adds -w/--workspace flag and
$TRUSTGRAPH_WORKSPACE env var.
Replace hallucinated relative imports with correct absolute imports
across the ontology query package, and fix OntologyMatcher reference
to match the actual class name OntologyMatcherForQueries. Simplify
test to use standard imports instead of importlib hack.
Cosmetic, but simpler imports provides undeterministic imports in a dev
environment, and also means we're properly testing linkage
Convert the SPARQL algebra evaluator from eager list-based evaluation to
lazy async generators so results stream incrementally. This lets Slice
terminate early (via generator cleanup) and avoids materialising full
result sets for streamable operators like Project, Filter, Union, and
Extend. Blocking operators (Join, LeftJoin, OrderBy, Group) materialise
at their boundary then yield.
Add bind join optimization for Join nodes where one side is small
(VALUES/ToMultiSet): instead of materialising both sides independently
and hash-joining, iterate the small side's bindings and evaluate the
large side with those bindings pre-seeded. This turns wildcard BGP
queries into selective ones — e.g. VALUES ?x { <uri> } joined with a
BGP now queries the triple store with ?x bound rather than fetching
all triples.
Add TriplesClient.query_gen() async generator that wraps the existing
streaming callback API via an asyncio.Queue bridge, yielding individual
Triple objects as batches arrive.
Add streaming request path in the SPARQL query service that batches
solutions from the live async generator and sends them as they fill.
Fix FILTER IN/NOT IN: rdflib represents these as RelationalExpression
nodes with op="IN", not as Builtin_IN — handle both representations.
Fix Builtin_IN/Builtin_NOTIN dispatch ordering so the specific handlers
are checked before the generic Builtin_ prefix match.
Fix VALUES handling for rdflib's two representations: positional
(var/value) and dict-based (res).
The Slice evaluator was propagating the SPARQL LIMIT value as the
inner limit for child evaluations, starving LeftJoin (OPTIONAL) and
other operators of results. The safety limit parameter should flow
through unchanged; LIMIT/OFFSET are applied only at the Slice node.
Add 30+ SPARQL 1.1 built-in functions and the MINUS algebra operator to the
custom SPARQL query backend.
String functions:
- SUBSTR (2-arg and 3-arg forms), STRBEFORE, STRAFTER
- REPLACE (regex with flags), ENCODE_FOR_URI
Numeric functions:
- FLOOR, CEIL, ROUND, ABS
Date/time accessors:
- YEAR, MONTH, DAY, HOURS, MINUTES, SECONDS
- NOW, TZ
Hash functions:
- MD5, SHA1, SHA256, SHA512
Term constructors:
- IRI/URI, BNODE, UUID, STRUUID
Other functions:
- LANGMATCHES, RAND
- EXISTS / NOT EXISTS (with async pre-evaluation to bridge the
sync expression evaluator and async algebra evaluator)
Algebra:
- MINUS set-difference operator
- HAVING already works via rdflib's Filter mapping (verified)
Fix SPARQL ORDER handling
Includes 653 lines of new unit tests covering all added functionality
across expressions, solutions, and algebra layers.
service.py:
- Constructor takes **config (same pattern as api-gateway) instead
of individual args
- Creates IamAuth and calls await self.auth.start() before the
message loop
- Passes auth to both ConfigReceiver and MessageDispatcher
- Uses add_pubsub_args / add_logging_args instead of hand-rolled
Pulsar args
- Passes timeout through
dispatcher.py:
- Accepts auth and timeout parameters
- Passes both to DispatcherManager — fixes the missing auth argument
that would have crashed on startup
The remote end's requests now go through the same IAM authentication
path as api-gateway. Token validation, workspace resolution, and
permissions all work identically regardless of which direction
initiated the connection.
Fixed tests — the test now passes auth and timeout to MessageDispatcher
and verifies they're forwarded to DispatcherManager.
Update rev gateway dispatcher to align with IAM. A "token" parameter
must be passed with each message.
Fix websocket relay to align with rev-gateway changes, conforms to
the api-gateway protocol.
consumer.py called unsubscribe() on every flow stop, deleting the
server-side subscription cursor. On restart, initial_position='latest'
skipped any messages published during the gap — causing intermittent
data loss (e.g. graph embeddings silently never reaching Qdrant).
Replace unsubscribe() with close() so the cursor survives restarts.
Move subscription cleanup to where it belongs: the Pulsar backend's
delete_topic(), called by the flow controller on deliberate flow
deletion. This was previously a no-op TODO.
Revert consumer receive timeout from 100ms back to the original
2000ms. The 100ms change was based on a misunderstanding — receive()
is a blocking call that returns immediately when a message arrives,
so the timeout only affects how quickly a consumer checks the shutdown
flag during idle periods. 100ms generated ~200 WARN lines/sec from
the C++ client with no latency benefit.
Also set the Pulsar C++ client logger to Error level so residual
timeout warnings from the subscriber (250ms) don't produce noise.
Update poll timeout test to match reverted 2000ms value
IAM auto-bootstrap could get permanently stuck in a half-done state:
_seed_tables wrote the workspace first, so any_workspace_exists()
returned true on restart even when user/key/signing-key creation had
failed. Remove workspace creation from _seed_tables (WorkspaceInit
handles it) and use any_signing_key_exists() as the completion check
since the signing key is the last thing written.
Run pre-service initialisers (PulsarTopology) in start() before
opening pub/sub connections, breaking the chicken-and-egg where the
bootstrapper needed Pulsar namespaces that it was responsible for
creating. Guard against empty cluster list when broker isn't ready.
Split the query once and check the parts list before indexing,
preventing an IndexError if the LLM returns an empty or
whitespace-only string.
Fixes#870.
Adds `no-auth-svc`, a lightweight IAM service that permits all access
unconditionally — no database, no bootstrap, no signing keys. Deploy
it in place of `iam-svc` for development, demos, and single-user
setups where authentication overhead is unwanted.
The gateway no longer hard-codes a 401 on missing credentials.
Instead it asks the IAM regime via a new `authenticate-anonymous`
operation whether token-free access is allowed. This keeps the
gateway regime-agnostic: `iam-svc` rejects anonymous auth (preserving
existing security), while `no-auth-svc` grants it with a configurable
default user and workspace.
Includes a tech spec (docs/tech-specs/no-auth-regime.md) and tests
that pin the safety boundary — malformed tokens never fall through
to the anonymous path, and a contract test ensures the full iam-svc
always rejects `authenticate-anonymous`.
* CLI auth migration, document embeddings core lifecycle (#913)
Migrate get_kg_core and put_kg_core CLI tools to use Api/SocketClient
with first-frame auth (fixes broken raw websocket path). Fix wire
format field names (root/vector). Remove ~600 lines of dead raw
websocket code from invoke_graph_rag.py.
Add document embeddings core lifecycle to the knowledge service:
list/get/put/delete/load operations across schema, translator,
Cassandra table store, knowledge manager, gateway registry, REST API,
socket client, and CLI (tg-get-de-core, tg-put-de-core).
Fix delete_kg_core to also clean up document embeddings rows.
* Remove spurious workspace parameter from SPARQL algebra evaluator (#915)
Fix threading of workspace paramater:
- The SPARQL algebra evaluator was threading a workspace parameter
through every function and passing it to TriplesClient.query(),
which doesn't accept it. Workspace isolation is handled by pub/sub
topic routing — the TriplesClient is already scoped to a
workspace-specific flow, same as GraphRAG. Passing workspace
explicitly was both incorrect and unnecessary.
Update tests:
- tests/unit/test_query/test_sparql_algebra.py (new) — Tests
_query_pattern, _eval_bgp, and evaluate() with various algebra
nodes. Key tests assert workspace is never in tc.query() kwargs,
plus correctness tests for BGP, JOIN, UNION, SLICE, DISTINCT, and
edge cases.
- tests/unit/test_retrieval/test_graph_rag.py — Added
test_triples_query_never_passes_workspace (checks query()) and
test_follow_edges_never_passes_workspace (checks query_stream()).
* Make all Cassandra and Qdrant I/O async-safe with proper concurrency controls (#916)
Cassandra triples services were using syncronous EntityCentricKnowledgeGraph
methods from async contexts, and connection state was managed with
threading.local which is wrong for asyncio coroutines sharing a single
thread. Qdrant services had no async wrapping at all, blocking the event
loop on every network call. Rows services had unprotected shared state
mutations across concurrent coroutines.
- Add async methods to EntityCentricKnowledgeGraph (async_insert,
async_get_s/p/o/sp/po/os/spo/all, async_collection_exists,
async_create_collection, async_delete_collection) using the existing
cassandra_async.async_execute bridge
- Rewrite triples write + query services: replace threading.local with
asyncio.Lock + dict cache for per-workspace connections, use async
ECKG methods for all data operations, keep asyncio.to_thread only for
one-time blocking ECKG construction
- Wrap all Qdrant calls in asyncio.to_thread across all 6 services
(doc/graph/row embeddings write + query), add asyncio.Lock + set cache
for collection existence checks
- Add asyncio.Lock to rows write + query services to protect shared
state (schemas, sessions, config caches) from concurrent mutation
- Update all affected tests to match new async patterns
* Fixed error only returning a page of results (#921)
The root cause: async_execute only materialises the first result
page (by design — it says so in its docstring). The streaming query
set fetch_size=20 and expected to iterate all results, but only got
the first 20 rows back.
The fix uses
asyncio.to_thread(lambda: list(tg.session.execute(...)))
which lets the sync driver iterate
all pages in a worker thread — exactly what the pre-async code did.
* Optional test warning suppression (#923)
* Fix test collection module errors & silence upstream Pytest warnings (#823)
* chore: add virtual environment and .env directories to gitignore
* test: filter upstream DeprecationWarning and UserWarning messages
* fix(namespace): remove empty __init__.py files to fix PEP 420 implicit namespace routing for trustgraph sub-packages
* Revert __init__.py deletions
* Add .ini changes but commented out, will be useful at times
---------
Co-authored-by: Salil M <d2kyt@protonmail.com>
* fix(openai): fail fast on unrecoverable RateLimitError codes (#901) (#904) (#925)
Co-authored-by: Sahil Yadav <sahilyadav.sy2004@gmail.com>
* Ensure retry exception is properly raised (#926)
* fix: library API get/update document round-trip bugs (#893) (#928)
Fix 5 cascading bugs in the Library API wrapper that prevented
the get_documents → update_document round-trip from working:
- Tolerate missing title field in document metadata (use .get())
- Use attribute access on Triple objects instead of subscript
- Serialize datetime to int seconds for JSON compatibility
- Handle empty server response on successful update
- Send both id and document-id keys in update request
Added library API tests
* Fix ontology selector defaults, add bypass mode, enforce domain/range (#929)
- Align similarity_threshold default to 0.3 everywhere (class signature
had stale 0.7). Fix matching contradiction in tech-spec.
- Add bypass_selector_below parameter (default 5) to skip vector
similarity selection when ontology element count is small enough.
- Enforce domain/range constraints in TripleConverter for object
properties and datatype properties, with subclass hierarchy support.
Properties with no declared domain/range pass through unchanged.
- Add unit tests for domain/range validation, subclass acceptance,
polymorphic pass-through, and selector bypass.
Fixes#908, #920
* Close producers on flow stop to prevent stale non-persistent topics (#930)
Flow.stop() only stopped consumers, leaving response producers
connected to non-persistent Pulsar topics. After flow restart, the
orphaned producers held stale broker routing state, causing response
messages to never reach new consumers — manifesting as 120s timeouts
on document-embeddings and similar RPC paths.
Fix: Flow.stop() now explicitly stops all producers. Producer.stop()
closes the underlying Pulsar producer connection rather than just
setting a flag.
Fixes#906
* fix(gateway): propagate --timeout flag to per-service dispatchers (#931)
The api-gateway accepts a --timeout flag (default 600s) but the value
was not propagated into DispatcherManager, which hard-coded
timeout=120 for every per-service dispatcher (graph-rag, document-rag,
text-completion, embeddings, librarian, etc.).
This meant any synchronous request taking more than 120 seconds would
always return a Timeout error at the 120s mark, regardless of the
--timeout value set on the gateway.
Changes:
- Add timeout parameter to DispatcherManager.__init__ (default: 120
for backward compatibility)
- Store self.timeout in DispatcherManager
- Replace both hardcoded timeout=120 with self.timeout in
invoke_global_service and invoke_flow_service
- Pass self.timeout from Api to DispatcherManager in service.py
- Document the timeout parameter in the docstring
Fixes#894
---------
Co-authored-by: Salil M <d2kyt@protonmail.com>
Co-authored-by: Sahil Yadav <sahilyadav.sy2004@gmail.com>
Co-authored-by: Mister Lobster <jlaportebot@gmail.com>
The api-gateway accepts a --timeout flag (default 600s) but the value
was not propagated into DispatcherManager, which hard-coded
timeout=120 for every per-service dispatcher (graph-rag, document-rag,
text-completion, embeddings, librarian, etc.).
This meant any synchronous request taking more than 120 seconds would
always return a Timeout error at the 120s mark, regardless of the
--timeout value set on the gateway.
Changes:
- Add timeout parameter to DispatcherManager.__init__ (default: 120
for backward compatibility)
- Store self.timeout in DispatcherManager
- Replace both hardcoded timeout=120 with self.timeout in
invoke_global_service and invoke_flow_service
- Pass self.timeout from Api to DispatcherManager in service.py
- Document the timeout parameter in the docstring
Fixes#894
