diff --git a/query_service/PROVENANCE_MODEL.md b/query_service/PROVENANCE_MODEL.md
new file mode 100644
index 0000000..7dd3d60
--- /dev/null
+++ b/query_service/PROVENANCE_MODEL.md
@@ -0,0 +1,198 @@
+# BrainKB Provenance Model (PROV-O in Oxigraph)
+
+Status: implemented on branch `improve-ingestion-query-service`.
+
+BrainKB stores knowledge in Oxigraph (a triplestore), so provenance is tracked
+**natively as W3C PROV-O triples in the graph database** and queried with SPARQL —
+not in a relational side-table. Postgres continues to hold job *execution state*
+(`jobs`, `job_results`, `job_processing_log`); Oxigraph is the provenance
+*source of truth*.
+
+This replaces the previous approach, which embedded PROV triples directly into
+each uploaded file's domain data via `attach_provenance()`. That polluted domain
+graphs, generated per-file provenance with random UUIDs unlinked to the job, and
+forced a full parse+re-serialize of every file (a memory/latency bottleneck).
+
+## Where provenance lives
+
+All provenance is written to one dedicated named graph:
+
+```
+https://brainkb.org/provenance/
+```
+
+Domain data uploaded during ingestion is **no longer modified** — files land in
+their target named graph exactly as provided.
+
+## Vocabulary
+
+| Prefix | IRI |
+|-----------|------------------------------------------------|
+| `prov` | `http://www.w3.org/ns/prov#` |
+| `dcterms` | `http://purl.org/dc/terms/` |
+| `xsd` | `http://www.w3.org/2001/XMLSchema#` |
+| `brainkb` | `https://brainkb.org/vocab/` (custom terms) |
+
+Instance IRIs are minted under `https://brainkb.org/prov/`:
+
+- Activity: `…/prov/activity/{job_id}`
+- Ingested bundle (entity): `…/prov/bundle/{job_id}`
+- Per-file entity: `…/prov/file/{job_id}/{urlencoded filename}`
+- Agent: `…/prov/agent/{urlencoded id}`; system agent: `…/prov/agent/system`
+
+## Tracked activities
+
+Data-mutation actions become a `prov:Activity` in the provenance graph.
+Named-graph registration is **not** duplicated here — it is recorded on the
+registry graph (see §2).
+
+### 1. Ingestion — `brainkb:IngestionActivity` (agent: user)
+
+Written when a job reaches a terminal state (`done` / `error` / partial).
+
+```turtle
+GRAPH {
+ <…/prov/activity/{job_id}>
+ a prov:Activity, brainkb:IngestionActivity ;
+ prov:startedAtTime "…"^^xsd:dateTime ;
+ prov:endedAtTime "…"^^xsd:dateTime ;
+ prov:wasAssociatedWith <…/prov/agent/{user}> ;
+ brainkb:targetGraph <{named_graph_iri}> ;
+ brainkb:jobStatus "done" ;
+ brainkb:totalFiles 20 ;
+ brainkb:successCount 19 ;
+ brainkb:failCount 1 .
+
+ <…/prov/bundle/{job_id}>
+ a prov:Entity ;
+ prov:wasGeneratedBy <…/prov/activity/{job_id}> ;
+ prov:wasAttributedTo <…/prov/agent/{user}> ;
+ prov:generatedAtTime "…"^^xsd:dateTime ;
+ dcterms:isPartOf <{named_graph_iri}> .
+
+ <…/prov/file/{job_id}/data_009.ttl>
+ a prov:Entity ;
+ prov:wasGeneratedBy <…/prov/activity/{job_id}> ;
+ dcterms:isPartOf <…/prov/bundle/{job_id}> ;
+ brainkb:fileName "data_009.ttl" ;
+ brainkb:uploadStatus "success" ;
+ brainkb:httpStatus 204 ;
+ brainkb:sizeBytes 41943040 .
+
+ <…/prov/agent/{user}> a prov:Agent, prov:Person .
+}
+```
+
+### 2. Named-graph registration (recorded in the registry graph — no duplication)
+
+Registration is **not** written as a separate activity in the provenance graph,
+because the named-graph **registry** graph
+(`https://brainkb.org/metadata/named-graph`) already records each registration as
+a PROV entity. We simply attribute it to the registering user on that same entry,
+so registration facts live in exactly one place:
+
+```turtle
+GRAPH {
+ <{named_graph_url}>
+ a prov:Entity ;
+ prov:generatedAtTime "…"^^xsd:dateTime ;
+ dcterms:description "…" ;
+ prov:wasAttributedTo <…/prov/agent/{user}> .
+}
+```
+
+`GET /api/query/registered-named-graphs` returns `registered_by` alongside the
+description and timestamp.
+
+### 3. Crash recovery — `brainkb:RecoveryActivity` (agent: system)
+
+Written when `recover_stuck_jobs()` marks a stuck job as `error`.
+
+```turtle
+<…/prov/activity/rec-{job_id}-{ts}>
+ a prov:Activity, brainkb:RecoveryActivity ;
+ prov:startedAtTime "…"^^xsd:dateTime ;
+ prov:wasAssociatedWith <…/prov/agent/system> ;
+ prov:used <…/prov/activity/{job_id}> ;
+ dcterms:description "{cause}" .
+<…/prov/agent/system> a prov:Agent, prov:SoftwareAgent .
+```
+
+## Writing
+
+Provenance triples are serialized to Turtle and appended to the provenance graph
+via Oxigraph's Graph Store HTTP protocol (`POST {endpoint}/store?graph=…`, which
+*merges* rather than replaces). Writes are best-effort: a provenance failure is
+logged but never fails the underlying job/registration.
+
+## Incremental change tracking (triple-level deltas)
+
+Provenance above records *who/what/when* at the activity level. To also track
+*which triples changed*, each ingestion job stages its data in a **per-job delta
+graph** before it reaches the target:
+
+```
+https://brainkb.org/provenance/delta/{job_id}
+```
+
+Flow (when `TRACK_TRIPLE_DELTAS=true`, the default):
+
+1. Each uploaded file is written to the job's delta graph (not the target).
+2. When the job finalizes, the delta graph is merged into the target graph
+ server-side via SPARQL `ADD SILENT TO ` (a set union, so
+ re-ingesting identical triples is idempotent — no duplicates).
+3. The delta graph is **preserved** as the immutable record of exactly what that
+ job added, and a PROV-O `brainkb:IngestionDelta` entity is written:
+
+```turtle
+<…/prov/delta/{job_id}>
+ a prov:Entity, brainkb:IngestionDelta ;
+ prov:wasGeneratedBy <…/prov/activity/{job_id}> ;
+ prov:wasDerivedFrom <{target_graph}> ;
+ brainkb:changeType "addition" ;
+ brainkb:targetGraph <{target_graph}> ;
+ brainkb:deltaGraph ;
+ brainkb:addedTripleCount 1234 ;
+ prov:generatedAtTime "…"^^xsd:dateTime ;
+ dcterms:isPartOf <…/prov/bundle/{job_id}> .
+```
+
+Set `TRACK_TRIPLE_DELTAS=false` to upload directly to the target graph (no delta
+graphs). Trade-off: delta graphs persist, so this roughly doubles stored triples
+for ingested data — the cost of full triple-level history and diffing.
+
+Current scope is **additions** (ingestion is append-only). Removals/updates would
+add a `removed` delta graph per activity; that is a future extension.
+
+## Retrieval (JSON-LD)
+
+Read endpoints return a PROV-O bundle as `application/ld+json` via SPARQL
+`CONSTRUCT` against the provenance graph:
+
+- `GET /api/provenance/job?job_id=…&user_id=…` — provenance for one job
+ (access-controlled with `verify_user_access`).
+- `GET /api/provenance/named-graph?iri=…` — all ingestion activity that targeted
+ a given named graph. (Registration attribution is on the registry graph; see
+ `GET /api/query/registered-named-graphs`.)
+
+Delta / change endpoints:
+
+- `GET /api/provenance/delta?job_id=…&user_id=…` — the exact triples a job added
+ (its delta graph) as JSON-LD (access-controlled).
+- `GET /api/provenance/delta/history?iri=…` — the ordered change history of a
+ named graph: one entry per delta (job, added triple count, timestamp, status),
+ newest first.
+- `GET /api/provenance/delta/compare?job_id_a=…&job_id_b=…&user_id=…` — compare
+ the triples added by two jobs; returns counts (A-only / B-only / shared) and the
+ differing triples as JSON-LD (access-controlled).
+
+Because everything is in Oxigraph, arbitrary provenance questions can also be
+asked directly over SPARQL, e.g. "all graphs ingested by agent X since T".
+
+## Backward compatibility
+
+- The `skip_provenance` query parameter on the ingestion endpoints is retained
+ but no longer controls domain-data embedding (which is removed). Graph-level
+ PROV-O tracking always happens.
+- `attach_provenance()` / `process_file_with_provenance()` remain in the codebase
+ but are no longer on the ingestion hot path.
diff --git a/query_service/RBAC_MODEL.md b/query_service/RBAC_MODEL.md
new file mode 100644
index 0000000..d9b1be7
--- /dev/null
+++ b/query_service/RBAC_MODEL.md
@@ -0,0 +1,120 @@
+# BrainKB Authorization (RBAC)
+
+Status: implemented on branch `improve-ingestion-query-service`.
+
+**JWT = authentication / API access. Roles = authorization (what you may do).**
+The query_service authenticates via JWT (scopes gate *API access* only) and then
+authorizes actions from the user's **roles**, mapped to **capabilities**, plus
+per-resource **space membership**.
+
+## Where roles come from
+
+Roles live in the Django-owned RBAC tables and join to a JWT user **by email**:
+
+```
+Web_jwtuser.email == Web_user_profile.email
+Web_user_profile.id -> Web_user_role (is_active, not expired) -> role name
+```
+
+query_service only **reads** roles (see `core/rbac.py`). It never assigns roles —
+creating/removing Admins is **role assignment**, owned by the usermanagement/Django
+side. This keeps the two systems consistent and means the KG API cannot be used to
+escalate privileges.
+
+## Role hierarchy
+
+```
+SuperAdmin >= Admin > write roles > read roles > (no role)
+```
+
+- **SuperAdmin** — ultimate authority, **bootstrapped at deployment**. Can do
+ everything Admin can; the SuperAdmin-vs-Admin difference (managing admins) is
+ role assignment, handled outside query_service.
+- **Admin** — all KG capabilities, incl. granting delegatable capabilities.
+- **write roles** — Curator, Lab Member, Submitter, Annotator, Mapper,
+ Knowledge Contributor: create their own private spaces + ingest.
+- **read roles** — Reviewer, Validator, Moderator, etc.: read member content.
+- **no role** — a JWT user not linked to a profile/role gets **public content
+ only** (read), nothing else.
+
+## Capabilities
+
+| Capability | Granted by | Gates |
+|---|---|---|
+| `create_private_space` | write roles, admins | create an individual/private space |
+| `create_team_space` | admins (or delegated) | create a team space |
+| `manage_team_space` | admins (or delegated) | manage a team space's members/visibility/graphs |
+| `ingest` | write roles, admins | ingest into a graph (also needs space owner/editor) |
+| `recover` | write roles, admins | recover stuck/errored jobs |
+| `read_private` | any role | read non-public content you're a member of |
+| `sparql_admin` | admins only | run arbitrary SPARQL |
+| `grant` | admins only | grant/revoke delegatable capabilities |
+
+**Delegated upgrades:** Admin/SuperAdmin can grant a specific user extra
+capabilities via `POST /api/admin/capabilities/grant` — e.g. let a Curator or Lab
+Member create/manage **team** spaces. Only **delegatable** caps may be granted
+(everything except `grant` and `sparql_admin`), so the grant endpoint can't turn a
+non-admin into an admin.
+
+## Enforcement points (layered)
+
+For each action: **JWT scope** (API access) → **capability** (role) → **space
+membership** (resource).
+
+| Action | Capability required | + resource check |
+|---|---|---|
+| Create individual/private space | `create_private_space` | — |
+| Create team space | `create_team_space` | — |
+| Manage space (members/visibility/graphs) | owner, or `manage_team_space`/admin | space ownership |
+| Ingest (`/insert/*`) | `ingest` | space owner/editor of the target graph |
+| Recover jobs | `recover` | own jobs |
+| Arbitrary SPARQL (`/query/sparql/`) | `sparql_admin` | — |
+| Read space / data | `read_private` for private; public = anyone (anon) | membership for private |
+
+## Space types
+
+- **individual** — a personal/private workspace; any write-capable user can create
+ one (`create_private_space`).
+- **team** — a shared workspace; only Admin/SuperAdmin (or a user granted
+ `create_team_space`) can create/manage it.
+
+## Admin capability endpoints
+
+- `GET /api/admin/capabilities?member=` — a user's roles, effective
+ capabilities, and grants (admin only).
+- `POST /api/admin/capabilities/grant` `{member, capability}` — delegate a cap.
+- `POST /api/admin/capabilities/revoke` `{member, capability}`.
+
+## Fine-grained per-space access rules
+
+On top of capabilities + owner/editor/viewer membership, a space can carry
+**access rules** that restrict a specific action to specific subjects — e.g. "in
+this space, only Admins may write", "only these Lab Members may read", "let this
+member manage".
+
+- Rule = `(action, subject_type, subject_value)`:
+ - `action` ∈ `read` | `write` | `manage`
+ - `subject_type` ∈ `global_role` (e.g. `Admin`, `Lab Member`) | `member` (an
+ email) | `space_role` (`viewer`|`editor`|`owner`, matched as ">=")
+- Semantics per action:
+ - **Owner** of the space and **global Admin/SuperAdmin** always pass (no lockout).
+ - `read`/`write`: if **no** rules exist for the action, the normal
+ capability/membership/visibility gates apply; if rules **do** exist, the caller
+ must also **match at least one**.
+ - `manage`: owner/admin/`manage_team_space` by default; a `manage` rule can
+ additionally **grant** management to a matched subject.
+- Rules layer *on top of* the global gates — e.g. ingest still needs the `ingest`
+ capability and space owner/editor membership; a write rule then narrows *which*
+ of those writers may actually ingest here.
+
+Endpoints (space manager only, except GET which is member/manager):
+
+- `GET /api/spaces/{slug}/access-rules`
+- `POST /api/spaces/{slug}/access-rules` `{action, subject_type, subject_value}`
+- `DELETE /api/spaces/{slug}/access-rules/{rule_id}`
+
+## Notes
+
+- JWT scopes (`read`/`write`/`admin`) remain as an API-access layer for defense in
+ depth; the authoritative "who can do what" is the role/capability + per-space
+ rule layer above.
diff --git a/query_service/README.md b/query_service/README.md
index 6a1dc22..d6224c9 100644
--- a/query_service/README.md
+++ b/query_service/README.md
@@ -1,14 +1,101 @@
# Query Service
-This service provides the endpoints and the functionalities necessary for querying (and updating) the knowledge graphs from the graph database.
-## Features Implemented
-- [x] Logging
-- [ ] Endpoints to query the data from the graph database
+FastAPI service for querying **and** ingesting BrainKB knowledge graphs in the
+graph database (Oxigraph), with W3C PROV-O provenance and team-owned
+private/public spaces.
+## Features
+- [x] Structured logging (with correlation IDs)
+- [x] JWT auth with per-endpoint scopes (`read` / `write` / `admin`)
+- [x] SPARQL query endpoints (registered graphs, taxonomy, arbitrary query — admin only)
+- [x] Bulk RDF ingestion into named graphs (file + raw), run as background jobs
+- [x] Job tracking: live status, processing history, progress, crash recovery
+- [x] Native **PROV-O provenance** in Oxigraph (ingestion / recovery activities)
+- [x] **Triple-level delta tracking** — per-job delta graphs + query/compare endpoints
+- [x] **Spaces** — owner-controlled, private/public containers of named graphs
+- [x] **Search** — hybrid Postgres-locator + Oxigraph-data, access-filtered by space
+## Auth & scopes
+Tokens are issued by the JWT/token manager and validated here. Scope policy:
+- **GET (reads)** → `read`
+- **Mutations** (ingest, register/attach graph, recover, create/modify space) → `write`
+- **Arbitrary SPARQL** (`/query/sparql/`) → `admin`
+- **Public-space reads** → no token required (anonymous), see Spaces below
+- `/register`, `/token` → public
+
+Users may only act on their own `user_id` (enforced), and job-scoped endpoints are
+owner-only.
+
+## Endpoints (prefix `/api`)
+
+### Query
+- `GET /query/registered-named-graphs` — registry/catalog of graphs (visibility-filtered)
+- `GET /query/taxonomy` — taxonomy view
+- `GET /query/sparql/` — arbitrary SPARQL (**admin**)
+
+### Ingestion & jobs
+- `POST /insert/raw/knowledge-graph-triples` — ingest raw triples (background job)
+- `POST /insert/files/knowledge-graph-triples` — ingest uploaded RDF files
+- `POST /register-named-graph` — register a named graph
+- `GET /insert/jobs`, `GET /insert/user/jobs/detail` — job listing / detail
+- `GET /insert/jobs/check-recoverable`, `POST /insert/jobs/recover` — crash recovery
+
+Ingestion is **submit-and-forget**: the request saves data to disk, creates a
+`pending` job, and returns immediately with a `job_id`; processing runs in the
+background (concurrent file uploads + batched DB writes) and the client polls job
+status. A per-worker **resource-safety cap** (`MAX_CONCURRENT_INGEST_JOBS`, default
+3) bounds how many jobs process at once so a burst of submissions can't exhaust
+memory / the DB pool / Oxigraph and crash the process — excess jobs simply wait as
+`pending` until a slot frees (backpressure, no queue rework). Search indexing is
+then queued separately in the background.
+
+### Provenance (PROV-O, JSON-LD)
+- `GET /provenance/job` — full bundle for one job
+- `GET /provenance/named-graph` — ingestion/activity history of a graph
+- `GET /provenance/delta` — exact triples a job added
+- `GET /provenance/delta/history` — a graph's change history
+- `GET /provenance/delta/compare` — diff two jobs' deltas
+
+See [PROVENANCE_MODEL.md](PROVENANCE_MODEL.md).
+
+### Spaces (private/public)
+- `POST /spaces`, `GET /spaces`, `GET /spaces/{slug}`
+- `PATCH /spaces/{slug}/visibility` — flip private/public (owner)
+- `POST/DELETE /spaces/{slug}/members[/{member}]` — membership (owner)
+- `POST /spaces/{slug}/graphs` — register + bind a graph to a space (owner/editor)
+- `GET /spaces/{slug}/data` — read space RDF (public = anonymous)
+
+**public** = readable by anyone, including unauthenticated clients; **private** =
+members only; **write/ingest** = space owner/editor only. See
+[SPACES_MODEL.md](SPACES_MODEL.md).
+
+### Search
+- `GET /search?q=…[&space={slug}][&limit&offset]` — full-text search, access-filtered.
+- `POST /search/reindex` (**admin**) — queue a background backfill/rebuild of the index.
+- `GET /search/index-tasks[?task_id=…]` — background indexing task status.
+
+Hybrid design: **Postgres** holds a full-text **locator index** (`graph_search_index`:
+subject + text + named graph + owning space), populated at ingest. A search runs in
+Postgres (fast, filtered by space visibility/membership), then the matched subjects'
+triples are fetched from **Oxigraph** (the source of truth). Anonymous → public
+spaces only; authenticated → public + own/member spaces (+ legacy). Pass `space` to
+scope to one workspace, omit for a full search. Private data is never returned to
+non-members — the filter is enforced in the locator query.
+
+**Indexing is asynchronous.** It never blocks ingestion: ingest enqueues an indexing
+task on an in-process async queue (durable `index_tasks` table, background consumer,
+atomic cross-worker claim, restart recovery) and the job completes immediately. The
+`/search/reindex` backfill uses the same queue. Poll `/search/index-tasks` for status.
+
+## Architecture notes
+
+- **Postgres** holds identity/teams/enforcement (JWT users, jobs, spaces/members/graphs).
+- **Oxigraph** holds all knowledge-graph data, PROV-O provenance, per-job delta
+ graphs, and a mirror of each space manifest — the graph database is the source of
+ truth for graph data and provenance.
### Acknowledgements
Special thanks to the authors of the resources below who helped with some best practices.
@@ -17,4 +104,4 @@ Special thanks to the authors of the resources below who helped with some best p
- FastAPI official documentation
### License
-[MIT](https://github.com/git/git-scm.com/blob/main/MIT-LICENSE.txt)
\ No newline at end of file
+[MIT](https://github.com/git/git-scm.com/blob/main/MIT-LICENSE.txt)
diff --git a/query_service/SPACES_MODEL.md b/query_service/SPACES_MODEL.md
new file mode 100644
index 0000000..b57434a
--- /dev/null
+++ b/query_service/SPACES_MODEL.md
@@ -0,0 +1,95 @@
+# BrainKB Spaces (private/public, team-owned)
+
+Status: implemented on branch `improve-ingestion-query-service`.
+
+Spaces let a user or team create their own **owner-controlled container** of named
+graphs, keep it **private**, or publish it **publicly** for anyone — including
+unauthenticated clients — to read. This supports a decentralized model where each
+space is a sovereign, IRI-addressable pod (`https://brainkb.org/space/{slug}`).
+
+## Storage split (confirmed architecture)
+
+- **Postgres** — identity/teams/enforcement only: JWT users, and the space tables
+ (`spaces`, `space_members`, `space_graphs`). This is the source of truth for
+ authorization and is what every request checks (fast, joinable with the JWT user).
+- **Oxigraph (graph DB)** — all knowledge-graph data AND provenance AND a
+ best-effort **RDF mirror** of each space manifest (in the spaces metadata graph
+ `https://brainkb.org/metadata/spaces/`), so spaces are portable/queryable via SPARQL.
+
+The RDF mirror is best-effort: a mirror failure never fails the enforcing Postgres
+write.
+
+## Model
+
+- `spaces(space_id, slug, name, description, owner, visibility, …)` —
+ `visibility ∈ {private, public}`.
+- `space_members(space_id, member, role)` — `role ∈ {owner, editor, viewer}`;
+ `member` is the user email (= the PROV agent id).
+- `space_graphs(space_id, named_graph_iri)` — a named graph belongs to exactly one
+ space.
+
+Roles: **owner** manages members/visibility/graphs; **editor** may ingest;
+**viewer** may read a private space; **public** grants read to everyone.
+
+## Authorization (enforced on every request)
+
+`spaces.authorize(named_graph_iri, member, need)`:
+
+| Space state | read | write (ingest) |
+|-------------|------|----------------|
+| public | **anyone (even anonymous)** | owner/editor only |
+| private | owner/editor/viewer | owner/editor |
+| *unmapped (legacy graph)* | falls through to endpoint scope | falls through to endpoint scope |
+
+Backward-compat: graphs never attached to a space (e.g. pre-existing graphs) are
+"unmapped" and keep their previous scope-only behavior; only space-mapped graphs
+are governed by space ACL.
+
+**Public = anonymous**: public reads require no token. Read endpoints that serve
+space data use an optional-auth dependency (`get_current_user_optional`) — a token
+is used if present, but its absence is not an error for public spaces.
+
+## Endpoints (`/api`)
+
+| Method | Path | Auth | Purpose |
+|--------|------|------|---------|
+| POST | `/spaces` | write | Create a space (caller = owner) |
+| GET | `/spaces` | optional | List visible spaces (public + own; anon → public only) |
+| GET | `/spaces/{slug}` | optional | Space manifest (public to anyone, private to members) |
+| PATCH | `/spaces/{slug}/visibility` | owner | Flip private/public |
+| POST | `/spaces/{slug}/members` | owner | Add/update a member (role) |
+| DELETE | `/spaces/{slug}/members/{member}` | owner | Remove a member |
+| POST | `/spaces/{slug}/graphs` | owner/editor | Register + bind a named graph to the space |
+| GET | `/spaces/{slug}/data` | optional | Read the space's RDF (JSON-LD); public = anonymous |
+
+Ingestion (`/insert/{raw,files}/knowledge-graph-triples`) now also checks space
+write-authorization for the target graph (in addition to the `write` scope and the
+user-identity check).
+
+## RDF manifest (mirror)
+
+```turtle
+GRAPH {
+
+ a brainkb:Space ;
+ brainkb:slug "{slug}" ; schema:name "…" ; dcterms:description "…" ;
+ brainkb:visibility "public" ;
+ brainkb:owner ;
+ brainkb:editor <…/agent/{editor}> ;
+ brainkb:viewer <…/agent/{viewer}> ;
+ brainkb:containsGraph <{named_graph_iri}> .
+}
+```
+
+## Notes
+
+- `registered-named-graphs` is **visibility-filtered**: graphs in a private space
+ the caller isn't a member of are omitted, so private graph existence is not
+ leaked. Public-space and legacy (unmapped) graphs remain listed. The endpoint
+ requires authentication (read scope); anonymous discovery of public spaces is via
+ `GET /api/spaces`.
+- Job-scoped provenance endpoints are **intentionally owner-restricted** (by
+ `user_id` = authenticated identity) and stay that way — job provenance is not made
+ public even for public spaces. Ingestion is likewise restricted to activated JWT
+ users with valid credentials (and space owner/editor membership); it is never
+ anonymous. Only *reads* of public spaces are anonymous.
diff --git a/query_service/VERIFICATION_QUERIES.md b/query_service/VERIFICATION_QUERIES.md
new file mode 100644
index 0000000..82edc7b
--- /dev/null
+++ b/query_service/VERIFICATION_QUERIES.md
@@ -0,0 +1,208 @@
+# BrainKB — Verification Queries
+
+SPARQL (Oxigraph) and SQL (Postgres) queries to verify data written by the
+BrainKB skill / MCP: ingested graphs, provenance, deltas, spaces, and the search
+index. Replace `my-lab` / `JOB_ID` with your values.
+
+## Fixed graph IRIs
+
+| Purpose | Named graph |
+|---|---|
+| Graph registry (catalog) | `https://brainkb.org/metadata/named-graph` |
+| Spaces manifest | `https://brainkb.org/metadata/spaces/` |
+| Provenance | `https://brainkb.org/provenance/` |
+| Per-job delta | `https://brainkb.org/provenance/delta/{job_id}` |
+| Your data | e.g. `https://brainkb.org/graph/my-lab/` |
+
+## Prefixes
+
+```sparql
+PREFIX prov:
+PREFIX dcterms:
+PREFIX schema:
+PREFIX brainkb:
+PREFIX rdfs:
+```
+
+---
+
+## SPARQL (Oxigraph)
+
+### 1. All named graphs + triple counts (what exists)
+
+```sparql
+SELECT ?g (COUNT(*) AS ?triples) WHERE { GRAPH ?g { ?s ?p ?o } }
+GROUP BY ?g ORDER BY DESC(?triples)
+```
+
+### 2. Your ingested data
+
+```sparql
+SELECT ?s ?p ?o WHERE { GRAPH { ?s ?p ?o } } LIMIT 200
+```
+
+### 3. Registry — which graphs are registered + by whom
+
+```sparql
+PREFIX prov:
+PREFIX dcterms:
+SELECT ?graph ?description ?registered_at ?registered_by WHERE {
+ GRAPH {
+ ?graph dcterms:description ?description ; prov:generatedAtTime ?registered_at .
+ OPTIONAL { ?graph prov:wasAttributedTo ?registered_by }
+ }
+}
+```
+
+### 4. Spaces manifest — visibility, owner, contained graphs
+
+```sparql
+PREFIX schema:
+PREFIX brainkb:
+SELECT ?space ?name ?visibility ?owner
+ (GROUP_CONCAT(DISTINCT STR(?graph); SEPARATOR=", ") AS ?graphs) WHERE {
+ GRAPH {
+ ?space a brainkb:Space ; schema:name ?name ;
+ brainkb:visibility ?visibility ; brainkb:owner ?owner .
+ OPTIONAL { ?space brainkb:containsGraph ?graph }
+ }
+} GROUP BY ?space ?name ?visibility ?owner
+```
+
+Members (owner/editor/viewer):
+
+```sparql
+PREFIX brainkb:
+SELECT ?space ?role ?agent WHERE {
+ GRAPH {
+ VALUES ?role { brainkb:owner brainkb:editor brainkb:viewer }
+ ?space ?role ?agent .
+ }
+}
+```
+
+### 5. Provenance — ingestion activities (who/when/status)
+
+```sparql
+PREFIX prov:
+PREFIX brainkb:
+SELECT ?activity ?agent ?targetGraph ?status ?start ?end ?success ?fail WHERE {
+ GRAPH {
+ ?activity a brainkb:IngestionActivity ;
+ prov:wasAssociatedWith ?agent ;
+ brainkb:targetGraph ?targetGraph ;
+ brainkb:jobStatus ?status ;
+ prov:startedAtTime ?start .
+ OPTIONAL { ?activity prov:endedAtTime ?end }
+ OPTIONAL { ?activity brainkb:successCount ?success }
+ OPTIONAL { ?activity brainkb:failCount ?fail }
+ }
+} ORDER BY DESC(?start)
+```
+
+### 6. Change history + deltas for a graph
+
+```sparql
+PREFIX prov:
+PREFIX brainkb:
+SELECT ?delta ?deltaGraph ?added ?time WHERE {
+ GRAPH {
+ ?delta a brainkb:IngestionDelta ;
+ brainkb:targetGraph ;
+ brainkb:deltaGraph ?deltaGraph ;
+ brainkb:addedTripleCount ?added ;
+ prov:generatedAtTime ?time .
+ }
+} ORDER BY DESC(?time)
+```
+
+The exact triples one job added:
+
+```sparql
+SELECT ?s ?p ?o WHERE { GRAPH { ?s ?p ?o } }
+```
+
+### 7. Per-file results for a job
+
+```sparql
+PREFIX prov:
+PREFIX brainkb:
+SELECT ?name ?status ?http ?size WHERE {
+ GRAPH {
+ ?file prov:wasGeneratedBy ;
+ brainkb:fileName ?name ; brainkb:uploadStatus ?status .
+ OPTIONAL { ?file brainkb:httpStatus ?http }
+ OPTIONAL { ?file brainkb:sizeBytes ?size }
+ }
+}
+```
+
+### 8. Find a term across everything (what search matched)
+
+```sparql
+PREFIX rdfs:
+SELECT ?g ?s ?label WHERE {
+ GRAPH ?g { ?s rdfs:label ?label . FILTER(CONTAINS(LCASE(STR(?label)), "purkinje")) }
+}
+```
+
+---
+
+## How to run the SPARQL
+
+### Via the API (needs `admin` scope)
+
+```bash
+Q='SELECT ?g (COUNT(*) AS ?n) WHERE { GRAPH ?g {?s ?p ?o} } GROUP BY ?g'
+curl -s "http://localhost:8010/api/query/sparql/?sparql_query=$(python3 -c "import urllib.parse,sys;print(urllib.parse.quote(sys.argv[1]))" "$Q")" \
+ -H "Authorization: Bearer $TOKEN"
+```
+
+Get `$TOKEN`:
+
+```bash
+TOKEN=$(curl -s -X POST http://localhost:8010/api/token -H 'Content-Type: application/json' \
+ -d '{"email":"you@example.com","password":"***"}' \
+ | python3 -c 'import sys,json;print(json.load(sys.stdin)["access_token"])')
+```
+
+### Directly against Oxigraph (nginx proxy on :7878, HTTP Basic auth)
+
+```bash
+curl -s "http://localhost:7878/query" \
+ --data-urlencode 'query=SELECT ?g (COUNT(*) AS ?n) WHERE { GRAPH ?g {?s ?p ?o} } GROUP BY ?g' \
+ -u admin:"$OXIGRAPH_PASSWORD" -H 'Accept: application/sparql-results+json'
+```
+
+(`OXIGRAPH_USER` / `OXIGRAPH_PASSWORD` are in `.env`.)
+
+---
+
+## SQL (Postgres — not in Oxigraph)
+
+Jobs, space ACL, the search locator index, and indexing tasks live in Postgres.
+
+```sql
+-- spaces & membership & graph bindings
+SELECT slug, name, visibility, owner FROM spaces;
+SELECT space_id, member, role FROM space_members;
+SELECT space_id, named_graph_iri FROM space_graphs;
+
+-- ingest jobs
+SELECT job_id, status, total_files, success_count, fail_count, start_time, end_time
+FROM jobs ORDER BY start_time DESC LIMIT 10;
+
+-- search locator index (subject text per graph/space)
+SELECT named_graph_iri, subject, left(text, 60) AS text FROM graph_search_index;
+
+-- background indexing tasks
+SELECT task_id, kind, target, status, subjects_indexed, graphs_done, graphs_total
+FROM index_tasks ORDER BY created_at DESC LIMIT 10;
+```
+
+Run against the Docker Postgres:
+
+```bash
+docker exec -e PGPASSWORD="$JWT_POSTGRES_DATABASE_PASSWORD" brainkb-postgres \
+ psql -U postgres -d brainkb -c "SELECT slug, visibility, owner FROM spaces;"
+```
diff --git a/query_service/core/indexing.py b/query_service/core/indexing.py
new file mode 100644
index 0000000..5928921
--- /dev/null
+++ b/query_service/core/indexing.py
@@ -0,0 +1,207 @@
+# -*- coding: utf-8 -*-
+# -----------------------------------------------------------------------------
+# DISCLAIMER: This software is provided "as is" without any warranty,
+# express or implied, including but not limited to the warranties of
+# merchantability, fitness for a particular purpose, and non-infringement.
+# -----------------------------------------------------------------------------
+
+# @Author : Tek Raj Chhetri
+# @Email : tekraj@mit.edu
+# @File : indexing.py
+
+"""
+Async search-indexing task queue.
+
+Search indexing (building the Postgres locator rows from Oxigraph) can be slow for
+large graphs, so it must not run inline with ingestion. This module provides a
+lightweight in-process asyncio queue with a single background consumer, backed by a
+durable `index_tasks` table for observability and cross-restart recovery.
+
+Ingest enqueues an 'ingest' task (fast) and returns immediately; a 'backfill' task
+reindexes every graph. Tasks are claimed atomically in the DB so multiple gunicorn
+workers never process the same task twice.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+import uuid
+from typing import Any, Dict, List, Optional
+
+from core.database import get_db_connection
+from core.search import index_graph_subjects, _sparql_select
+
+logger = logging.getLogger(__name__)
+
+# Graphs that are infrastructure, not user data — never indexed for search.
+_SKIP_GRAPHS = {
+ "https://brainkb.org/metadata/named-graph",
+ "https://brainkb.org/metadata/spaces/",
+ "https://brainkb.org/provenance/",
+}
+
+_queue: Optional[asyncio.Queue] = None
+_consumer: Optional[asyncio.Task] = None
+
+
+def _get_queue() -> asyncio.Queue:
+ global _queue
+ if _queue is None:
+ _queue = asyncio.Queue()
+ return _queue
+
+
+async def enqueue_ingest(named_graph_iri: str, delta_graph: Optional[str] = None) -> str:
+ """Queue an indexing task for a single graph (called by ingest). Non-blocking."""
+ task_id = uuid.uuid4().hex
+ async with get_db_connection() as conn:
+ await conn.execute(
+ """
+ INSERT INTO index_tasks (task_id, kind, target, delta_graph, status, created_at)
+ VALUES ($1, 'ingest', $2, $3, 'queued', $4)
+ """,
+ task_id, named_graph_iri, delta_graph, time.time(),
+ )
+ _get_queue().put_nowait({"task_id": task_id, "kind": "ingest",
+ "target": named_graph_iri, "delta_graph": delta_graph})
+ return task_id
+
+
+async def enqueue_backfill() -> str:
+ """Queue a full reindex of every user graph. Non-blocking."""
+ task_id = uuid.uuid4().hex
+ async with get_db_connection() as conn:
+ await conn.execute(
+ """
+ INSERT INTO index_tasks (task_id, kind, target, status, created_at)
+ VALUES ($1, 'backfill', 'ALL', 'queued', $2)
+ """,
+ task_id, time.time(),
+ )
+ _get_queue().put_nowait({"task_id": task_id, "kind": "backfill", "target": "ALL"})
+ return task_id
+
+
+async def _claim(task_id: str) -> bool:
+ """Atomically move a task queued->running. Returns False if already claimed
+ (by another worker) or not claimable — prevents double processing."""
+ async with get_db_connection() as conn:
+ row = await conn.fetchrow(
+ """
+ UPDATE index_tasks SET status = 'running', started_at = $2
+ WHERE task_id = $1 AND status = 'queued'
+ RETURNING task_id
+ """,
+ task_id, time.time(),
+ )
+ return row is not None
+
+
+async def _finish(task_id: str, status: str, subjects: int = 0, message: str = "",
+ graphs_total: Optional[int] = None, graphs_done: int = 0) -> None:
+ async with get_db_connection() as conn:
+ await conn.execute(
+ """
+ UPDATE index_tasks
+ SET status = $2, subjects_indexed = $3, message = $4,
+ graphs_total = COALESCE($5, graphs_total), graphs_done = $6, ended_at = $7
+ WHERE task_id = $1
+ """,
+ task_id, status, subjects, message[:500] if message else None,
+ graphs_total, graphs_done, time.time(),
+ )
+
+
+async def _all_user_graphs() -> List[str]:
+ rows = await _sparql_select("SELECT DISTINCT ?g WHERE { GRAPH ?g { ?s ?p ?o } }")
+ graphs = []
+ for b in rows:
+ g = b.get("g", {}).get("value")
+ if g and g not in _SKIP_GRAPHS and "/provenance/delta/" not in g:
+ graphs.append(g)
+ return graphs
+
+
+async def _run(task: Dict[str, Any]) -> None:
+ task_id = task["task_id"]
+ if not await _claim(task_id):
+ return # another worker already handling it, or not queued
+ try:
+ if task["kind"] == "ingest":
+ n = await index_graph_subjects(task["target"], task.get("delta_graph"))
+ await _finish(task_id, "done", subjects=n, message=f"indexed {n} subject(s)")
+ elif task["kind"] == "backfill":
+ graphs = await _all_user_graphs()
+ total, done, subs = len(graphs), 0, 0
+ # record the total up front (status stays 'running')
+ async with get_db_connection() as conn:
+ await conn.execute("UPDATE index_tasks SET graphs_total=$2 WHERE task_id=$1", task_id, total)
+ for g in graphs:
+ subs += await index_graph_subjects(g)
+ done += 1
+ async with get_db_connection() as conn:
+ await conn.execute(
+ "UPDATE index_tasks SET graphs_done=$2, subjects_indexed=$3 WHERE task_id=$1",
+ task_id, done, subs,
+ )
+ await _finish(task_id, "done", subjects=subs, graphs_total=total, graphs_done=done,
+ message=f"reindexed {done}/{total} graph(s), {subs} subject(s)")
+ except Exception as e:
+ logger.error(f"[indexing] task {task_id} failed: {e}", exc_info=True)
+ await _finish(task_id, "error", message=str(e))
+
+
+async def _consume() -> None:
+ q = _get_queue()
+ while True:
+ task = await q.get()
+ try:
+ await _run(task)
+ except Exception as e:
+ logger.error(f"[indexing] consumer error: {e}", exc_info=True)
+ finally:
+ q.task_done()
+
+
+async def start_worker() -> None:
+ """Start the background consumer and recover tasks from a previous run.
+ Called once on application startup."""
+ global _consumer
+ # Recovery: any task left 'running' by a crashed process is re-queued.
+ try:
+ async with get_db_connection() as conn:
+ await conn.execute("UPDATE index_tasks SET status = 'queued' WHERE status = 'running'")
+ pending = await conn.fetch(
+ "SELECT task_id, kind, target, delta_graph FROM index_tasks WHERE status = 'queued'"
+ )
+ except Exception as e:
+ logger.warning(f"[indexing] recovery query failed: {e}")
+ pending = []
+
+ if _consumer is None or _consumer.done():
+ _consumer = asyncio.create_task(_consume())
+ logger.info("[indexing] background consumer started")
+
+ for p in pending:
+ _get_queue().put_nowait({
+ "task_id": p["task_id"], "kind": p["kind"],
+ "target": p["target"], "delta_graph": p["delta_graph"],
+ })
+ if pending:
+ logger.info(f"[indexing] re-queued {len(pending)} pending task(s) from previous session")
+
+
+async def get_task(task_id: str) -> Optional[Dict[str, Any]]:
+ async with get_db_connection() as conn:
+ row = await conn.fetchrow("SELECT * FROM index_tasks WHERE task_id = $1", task_id)
+ return dict(row) if row else None
+
+
+async def list_tasks(limit: int = 50) -> List[Dict[str, Any]]:
+ async with get_db_connection() as conn:
+ rows = await conn.fetch(
+ "SELECT * FROM index_tasks ORDER BY created_at DESC LIMIT $1", limit
+ )
+ return [dict(r) for r in rows]
diff --git a/query_service/core/main.py b/query_service/core/main.py
index 1978718..1dc8896 100644
--- a/query_service/core/main.py
+++ b/query_service/core/main.py
@@ -12,6 +12,8 @@
from core.routers.query import router as query_router
from core.routers.rapid_release import router as rapid_release
from core.routers.insert import router as insert_router
+from core.routers.spaces import router as spaces_router
+from core.routers.search import router as search_router
from core.configuration import load_environment
from core.database import init_db_pool
from core.graph_database_connection_manager import initialize_metadata_graph
@@ -50,6 +52,8 @@
app.include_router(jwt_router, prefix="/api")
app.include_router(query_router, prefix="/api")
app.include_router(insert_router,prefix="/api")
+app.include_router(spaces_router, prefix="/api", tags=["Spaces"])
+app.include_router(search_router, prefix="/api", tags=["Search"])
# rapid-release
app.include_router(rapid_release, prefix="/api/rapid-release", tags=["Rapid release"])
@@ -153,6 +157,155 @@ async def startup_event():
except Exception:
pass # Indexes may already exist
logger.info("Job tracking tables initialized")
+
+ # Spaces: owner-controlled containers of named graphs with
+ # private/public visibility and team membership (see spaces.py).
+ await conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS spaces (
+ space_id TEXT PRIMARY KEY,
+ slug TEXT UNIQUE NOT NULL,
+ name TEXT NOT NULL,
+ description TEXT,
+ owner TEXT NOT NULL,
+ visibility TEXT NOT NULL DEFAULT 'private',
+ created_at DOUBLE PRECISION,
+ updated_at DOUBLE PRECISION
+ )
+ """
+ )
+ await conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS space_members (
+ id SERIAL PRIMARY KEY,
+ space_id TEXT NOT NULL REFERENCES spaces(space_id) ON DELETE CASCADE,
+ member TEXT NOT NULL,
+ role TEXT NOT NULL,
+ added_at DOUBLE PRECISION,
+ UNIQUE (space_id, member)
+ )
+ """
+ )
+ await conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS space_graphs (
+ id SERIAL PRIMARY KEY,
+ space_id TEXT NOT NULL REFERENCES spaces(space_id) ON DELETE CASCADE,
+ named_graph_iri TEXT NOT NULL UNIQUE,
+ added_at DOUBLE PRECISION
+ )
+ """
+ )
+ try:
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_space_members_space ON space_members(space_id)")
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_space_members_member ON space_members(member)")
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_space_graphs_space ON space_graphs(space_id)")
+ except Exception:
+ pass
+ # Space type: 'individual' (personal) or 'team' (created by
+ # Admin/SuperAdmin or a user granted create_team_space).
+ try:
+ await conn.execute("ALTER TABLE spaces ADD COLUMN IF NOT EXISTS space_type TEXT NOT NULL DEFAULT 'individual'")
+ except Exception:
+ pass
+ logger.info("Spaces tables initialized")
+
+ # RBAC: capabilities granted directly to a user (delegated
+ # upgrades by Admin/SuperAdmin), on top of role-derived caps.
+ await conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS user_capability_grants (
+ id SERIAL PRIMARY KEY,
+ member TEXT NOT NULL,
+ capability TEXT NOT NULL,
+ granted_by TEXT,
+ created_at DOUBLE PRECISION,
+ UNIQUE (member, capability)
+ )
+ """
+ )
+ try:
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_capability_grants_member ON user_capability_grants(member)")
+ except Exception:
+ pass
+ logger.info("RBAC capability-grants table initialized")
+
+ # Fine-grained per-space access rules: restrict a space action
+ # (read/write/manage) to a global role, a space role, or specific
+ # members. Owner + global Admin/SuperAdmin always bypass.
+ await conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS space_access_rules (
+ id SERIAL PRIMARY KEY,
+ space_id TEXT NOT NULL REFERENCES spaces(space_id) ON DELETE CASCADE,
+ action TEXT NOT NULL,
+ subject_type TEXT NOT NULL,
+ subject_value TEXT NOT NULL,
+ created_by TEXT,
+ created_at DOUBLE PRECISION,
+ UNIQUE (space_id, action, subject_type, subject_value)
+ )
+ """
+ )
+ try:
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_space_access_rules_space ON space_access_rules(space_id, action)")
+ except Exception:
+ pass
+ logger.info("Space access-rules table initialized")
+
+ # Search locator index (hybrid search): Postgres full-text index that
+ # locates subjects/subgraphs (carrying graph + workspace/space), then the
+ # actual triples are fetched from Oxigraph. Access is filtered by space
+ # visibility/membership at query time.
+ await conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS graph_search_index (
+ id BIGSERIAL PRIMARY KEY,
+ named_graph_iri TEXT NOT NULL,
+ space_id TEXT,
+ subject TEXT NOT NULL,
+ text TEXT NOT NULL,
+ tsv tsvector,
+ updated_at DOUBLE PRECISION,
+ UNIQUE (named_graph_iri, subject)
+ )
+ """
+ )
+ try:
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_gsi_tsv ON graph_search_index USING GIN(tsv)")
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_gsi_graph ON graph_search_index(named_graph_iri)")
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_gsi_space ON graph_search_index(space_id)")
+ except Exception:
+ pass
+ logger.info("Search index table initialized")
+
+ # Async indexing task queue: search indexing runs in the
+ # background (not inline with ingest) so large graphs don't
+ # block jobs. Task status is durable here for observability
+ # and cross-restart recovery.
+ await conn.execute(
+ """
+ CREATE TABLE IF NOT EXISTS index_tasks (
+ task_id TEXT PRIMARY KEY,
+ kind TEXT NOT NULL,
+ target TEXT,
+ delta_graph TEXT,
+ status TEXT NOT NULL,
+ subjects_indexed INTEGER DEFAULT 0,
+ graphs_total INTEGER,
+ graphs_done INTEGER DEFAULT 0,
+ message TEXT,
+ created_at DOUBLE PRECISION,
+ started_at DOUBLE PRECISION,
+ ended_at DOUBLE PRECISION
+ )
+ """
+ )
+ try:
+ await conn.execute("CREATE INDEX IF NOT EXISTS idx_index_tasks_status ON index_tasks(status)")
+ except Exception:
+ pass
+ logger.info("Index task table initialized")
break # Success, exit retry loop
finally:
await pool.release(conn)
@@ -207,6 +360,14 @@ async def startup_event():
except Exception as e:
logger.warning(f"Failed to recover stuck jobs: {str(e)}. Continuing anyway...")
+ # Start the background search-indexing consumer and recover any pending tasks
+ logger.info("Starting background search-indexing worker...")
+ try:
+ from core.indexing import start_worker
+ await start_worker()
+ except Exception as e:
+ logger.warning(f"Failed to start indexing worker: {str(e)}. Continuing anyway...")
+
logger.info("FastAPI startup completed successfully")
diff --git a/query_service/core/provenance.py b/query_service/core/provenance.py
new file mode 100644
index 0000000..b8528bd
--- /dev/null
+++ b/query_service/core/provenance.py
@@ -0,0 +1,479 @@
+# -*- coding: utf-8 -*-
+# -----------------------------------------------------------------------------
+# DISCLAIMER: This software is provided "as is" without any warranty,
+# express or implied, including but not limited to the warranties of
+# merchantability, fitness for a particular purpose, and non-infringement.
+#
+# In no event shall the authors or copyright holders be liable for any
+# claim, damages, or other liability, whether in an action of contract,
+# tort, or otherwise, arising from, out of, or in connection with the
+# software or the use or other dealings in the software.
+# -----------------------------------------------------------------------------
+
+# @Author : Tek Raj Chhetri
+# @Email : tekraj@mit.edu
+# @Web : https://tekrajchhetri.com/
+# @File : provenance.py
+# @Software: PyCharm
+
+"""
+Native PROV-O provenance tracking in Oxigraph.
+
+BrainKB stores knowledge in a triplestore, so provenance is tracked as W3C
+PROV-O triples written to a dedicated provenance named graph and queried with
+SPARQL. See PROVENANCE_MODEL.md for the full design.
+
+This module is intentionally self-contained and best-effort: helpers here build
+and persist provenance, but callers must ensure a provenance failure never fails
+the underlying operation (wrap calls in try/except).
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import logging
+from typing import Any, Dict, List, Optional
+from urllib.parse import quote
+
+import httpx
+from rdflib import Graph, Literal, Namespace, URIRef, RDF, XSD, DCTERMS
+
+from core.shared import get_oxigraph_endpoint, get_oxigraph_auth
+from core.graph_database_connection_manager import _get_endpoint
+
+logger = logging.getLogger(__name__)
+
+# Dedicated named graph holding all provenance.
+PROVENANCE_GRAPH = "https://brainkb.org/provenance/"
+
+# Per-job delta graphs live under this base. Each ingestion job stages its
+# triples in its own delta graph, giving an exact, queryable record of what that
+# job added (triple-level change tracking) before/after merging into the target.
+PROVENANCE_DELTA_BASE = "https://brainkb.org/provenance/delta/"
+
+# Namespaces
+PROV = Namespace("http://www.w3.org/ns/prov#")
+BRAINKB = Namespace("https://brainkb.org/vocab/") # custom vocabulary
+PROV_BASE = Namespace("https://brainkb.org/prov/") # instance IRIs
+
+
+def _now_iso() -> str:
+ """Current UTC time as an ISO-8601 string."""
+ return datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+
+def _new_graph() -> Graph:
+ g = Graph()
+ g.bind("prov", PROV)
+ g.bind("brainkb", BRAINKB)
+ g.bind("dcterms", DCTERMS)
+ return g
+
+
+def agent_ref(agent_id: str) -> URIRef:
+ """Mint the agent IRI for a user/system identifier."""
+ return URIRef(PROV_BASE[f"agent/{quote(str(agent_id), safe='')}"])
+
+
+def activity_ref(job_id: str) -> URIRef:
+ return URIRef(PROV_BASE[f"activity/{quote(str(job_id), safe='')}"])
+
+
+def delta_graph_for(job_id: str) -> str:
+ """The per-job delta graph IRI (holds exactly the triples this job added)."""
+ return f"{PROVENANCE_DELTA_BASE}{quote(str(job_id), safe='')}"
+
+
+# ---------------------------------------------------------------------------
+# Builders — each returns an rdflib Graph of PROV-O triples
+# ---------------------------------------------------------------------------
+
+def build_ingestion_provenance(
+ *,
+ job_id: str,
+ agent_id: str,
+ named_graph_iri: str,
+ started_at: str,
+ ended_at: str,
+ status: str,
+ total_files: int,
+ success_count: int,
+ fail_count: int,
+ results: Optional[List[Dict[str, Any]]] = None,
+ agent_type: str = "user",
+ delta_graph: Optional[str] = None,
+ added_triple_count: Optional[int] = None,
+) -> Graph:
+ """Build the PROV-O bundle for a completed (terminal) ingestion job.
+
+ When ``delta_graph`` is given, a ``brainkb:IngestionDelta`` entity is added
+ describing the incremental change: which named graph it derives from, the
+ delta graph holding the exact added triples, and (optionally) the count.
+ """
+ g = _new_graph()
+
+ activity = activity_ref(job_id)
+ bundle = URIRef(PROV_BASE[f"bundle/{quote(str(job_id), safe='')}"])
+ agent = agent_ref(agent_id)
+
+ # Agent
+ g.add((agent, RDF.type, PROV.Agent))
+ g.add((agent, RDF.type, PROV.SoftwareAgent if agent_type == "system" else PROV.Person))
+
+ # Activity
+ g.add((activity, RDF.type, PROV.Activity))
+ g.add((activity, RDF.type, BRAINKB.IngestionActivity))
+ g.add((activity, PROV.startedAtTime, Literal(started_at, datatype=XSD.dateTime)))
+ g.add((activity, PROV.endedAtTime, Literal(ended_at, datatype=XSD.dateTime)))
+ g.add((activity, PROV.wasAssociatedWith, agent))
+ g.add((activity, BRAINKB.targetGraph, URIRef(named_graph_iri)))
+ g.add((activity, BRAINKB.jobStatus, Literal(status)))
+ g.add((activity, BRAINKB.totalFiles, Literal(int(total_files), datatype=XSD.integer)))
+ g.add((activity, BRAINKB.successCount, Literal(int(success_count), datatype=XSD.integer)))
+ g.add((activity, BRAINKB.failCount, Literal(int(fail_count), datatype=XSD.integer)))
+
+ # Ingested bundle entity
+ g.add((bundle, RDF.type, PROV.Entity))
+ g.add((bundle, PROV.wasGeneratedBy, activity))
+ g.add((bundle, PROV.wasAttributedTo, agent))
+ g.add((bundle, PROV.generatedAtTime, Literal(ended_at, datatype=XSD.dateTime)))
+ g.add((bundle, DCTERMS.isPartOf, URIRef(named_graph_iri)))
+
+ # Per-file entities
+ for r in results or []:
+ fname = str(r.get("file", "unknown"))
+ file_entity = URIRef(PROV_BASE[f"file/{quote(str(job_id), safe='')}/{quote(fname, safe='')}"])
+ g.add((file_entity, RDF.type, PROV.Entity))
+ g.add((file_entity, PROV.wasGeneratedBy, activity))
+ g.add((file_entity, DCTERMS.isPartOf, bundle))
+ g.add((file_entity, BRAINKB.fileName, Literal(fname)))
+ g.add((file_entity, BRAINKB.uploadStatus, Literal("success" if r.get("success") else "failed")))
+ if r.get("http_status") is not None:
+ g.add((file_entity, BRAINKB.httpStatus, Literal(int(r["http_status"]), datatype=XSD.integer)))
+ if r.get("size_bytes") is not None:
+ g.add((file_entity, BRAINKB.sizeBytes, Literal(int(r["size_bytes"]), datatype=XSD.integer)))
+
+ # Incremental-change (delta) entity: an isolated, queryable record of exactly
+ # the triples this job contributed to the target graph.
+ if delta_graph:
+ delta = URIRef(PROV_BASE[f"delta/{quote(str(job_id), safe='')}"])
+ g.add((delta, RDF.type, PROV.Entity))
+ g.add((delta, RDF.type, BRAINKB.IngestionDelta))
+ g.add((delta, PROV.wasGeneratedBy, activity))
+ # The change derives from (is applied to) the target named graph
+ g.add((delta, PROV.wasDerivedFrom, URIRef(named_graph_iri)))
+ g.add((delta, BRAINKB.changeType, Literal("addition")))
+ g.add((delta, BRAINKB.targetGraph, URIRef(named_graph_iri)))
+ g.add((delta, BRAINKB.deltaGraph, URIRef(delta_graph)))
+ g.add((delta, DCTERMS.isPartOf, bundle))
+ g.add((delta, PROV.generatedAtTime, Literal(ended_at, datatype=XSD.dateTime)))
+ if added_triple_count is not None:
+ g.add((delta, BRAINKB.addedTripleCount, Literal(int(added_triple_count), datatype=XSD.integer)))
+
+ return g
+
+
+def build_recovery_provenance(
+ *,
+ job_id: str,
+ at: Optional[str] = None,
+ cause: str = "",
+) -> Graph:
+ """Build PROV-O for an automated crash-recovery action (agent: system)."""
+ g = _new_graph()
+ at = at or _now_iso()
+ ts = at.replace(":", "").replace("-", "").replace(".", "")
+ activity = URIRef(PROV_BASE[f"activity/rec-{quote(str(job_id), safe='')}-{quote(ts, safe='')}"])
+ system_agent = agent_ref("system")
+
+ g.add((system_agent, RDF.type, PROV.Agent))
+ g.add((system_agent, RDF.type, PROV.SoftwareAgent))
+ g.add((activity, RDF.type, PROV.Activity))
+ g.add((activity, RDF.type, BRAINKB.RecoveryActivity))
+ g.add((activity, PROV.startedAtTime, Literal(at, datatype=XSD.dateTime)))
+ g.add((activity, PROV.wasAssociatedWith, system_agent))
+ # Link the recovery activity to the ingestion activity it acted upon
+ g.add((activity, PROV.used, activity_ref(job_id)))
+ if cause:
+ g.add((activity, DCTERMS.description, Literal(cause)))
+ return g
+
+
+# ---------------------------------------------------------------------------
+# Persistence + retrieval
+# ---------------------------------------------------------------------------
+
+async def write_provenance(graph: Graph) -> bool:
+ """
+ Append PROV-O triples to the provenance named graph via Oxigraph's Graph
+ Store HTTP protocol (POST merges into the graph). Best-effort: returns True
+ on success, False on failure (never raises).
+ """
+ if graph is None or len(graph) == 0:
+ return True
+ try:
+ payload = graph.serialize(format="turtle")
+ endpoint = get_oxigraph_endpoint() # .../store
+ url = f"{endpoint}?graph={quote(PROVENANCE_GRAPH, safe='')}"
+ auth = get_oxigraph_auth()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=15.0)) as client:
+ resp = await client.post(
+ url,
+ content=payload.encode("utf-8"),
+ headers={"Content-Type": "text/turtle"},
+ auth=auth,
+ )
+ if resp.status_code in (200, 201, 204):
+ return True
+ logger.warning(
+ "[provenance] Failed to write provenance (HTTP %s): %s",
+ resp.status_code, (resp.text or "")[:500],
+ )
+ return False
+ except Exception as e:
+ logger.warning(f"[provenance] Error writing provenance: {e}", exc_info=True)
+ return False
+
+
+async def merge_delta_into_target(delta_graph: str, target_graph: str) -> bool:
+ """
+ Copy all triples from a per-job delta graph into the target named graph via
+ SPARQL Update ADD (server-side; the delta graph is preserved as the change
+ record). Best-effort: returns True on success, False otherwise.
+ """
+ try:
+ update = f"ADD SILENT <{delta_graph}> TO <{target_graph}>"
+ endpoint = _get_endpoint("post") # .../update for OXIGRAPH
+ auth = get_oxigraph_auth()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(600.0, connect=15.0)) as client:
+ resp = await client.post(endpoint, data={"update": update}, auth=auth)
+ if resp.status_code in (200, 204):
+ return True
+ logger.warning(
+ "[provenance] Delta merge failed (HTTP %s): %s",
+ resp.status_code, (resp.text or "")[:500],
+ )
+ return False
+ except Exception as e:
+ logger.warning(f"[provenance] Error merging delta graph: {e}", exc_info=True)
+ return False
+
+
+async def count_graph_triples(graph_iri: str) -> Optional[int]:
+ """Count triples in a named graph. Returns None on error."""
+ try:
+ query = f"SELECT (COUNT(*) AS ?n) WHERE {{ GRAPH <{graph_iri}> {{ ?s ?p ?o }} }}"
+ endpoint = _get_endpoint("get") # .../query for OXIGRAPH
+ auth = get_oxigraph_auth()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(30.0, connect=10.0)) as client:
+ resp = await client.post(
+ endpoint,
+ data={"query": query},
+ headers={"Accept": "application/sparql-results+json"},
+ auth=auth,
+ )
+ if resp.status_code != 200:
+ return None
+ bindings = resp.json().get("results", {}).get("bindings", [])
+ return int(bindings[0]["n"]["value"]) if bindings else 0
+ except Exception as e:
+ logger.warning(f"[provenance] Error counting triples in {graph_iri}: {e}", exc_info=True)
+ return None
+
+
+async def query_provenance_jsonld(construct_query: str) -> Optional[str]:
+ """
+ Execute a SPARQL CONSTRUCT against Oxigraph and return JSON-LD text.
+ Returns None on error.
+
+ Oxigraph does not serialize CONSTRUCT results as JSON-LD (it offers
+ Turtle / N-Triples / N-Quads / RDF-XML), so we request Turtle and convert to
+ JSON-LD locally with rdflib. This keeps the JSON-LD API contract independent
+ of the triplestore's supported output formats.
+ """
+ try:
+ endpoint = _get_endpoint("get") # .../query for OXIGRAPH
+ auth = get_oxigraph_auth()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(30.0, connect=10.0)) as client:
+ resp = await client.post(
+ endpoint,
+ data={"query": construct_query},
+ headers={"Accept": "text/turtle"},
+ auth=auth,
+ )
+ if resp.status_code != 200:
+ logger.warning(
+ "[provenance] CONSTRUCT query failed (HTTP %s): %s",
+ resp.status_code, (resp.text or "")[:500],
+ )
+ return None
+ g = Graph()
+ g.parse(data=resp.text, format="turtle")
+ g.bind("prov", PROV)
+ g.bind("brainkb", BRAINKB)
+ g.bind("dcterms", DCTERMS)
+ return g.serialize(format="json-ld", auto_compact=True)
+ except Exception as e:
+ logger.warning(f"[provenance] Error querying provenance: {e}", exc_info=True)
+ return None
+
+
+def construct_for_job(job_id: str) -> str:
+ """SPARQL CONSTRUCT for all provenance about a single job (activity, bundle,
+ per-file entities, and the connected agent)."""
+ activity = str(activity_ref(job_id))
+ bundle = str(URIRef(PROV_BASE[f"bundle/{quote(str(job_id), safe='')}"]))
+ delta = str(URIRef(PROV_BASE[f"delta/{quote(str(job_id), safe='')}"]))
+ file_prefix = f"{str(PROV_BASE)}file/{quote(str(job_id), safe='')}/"
+ return f"""
+ CONSTRUCT {{ ?s ?p ?o }}
+ WHERE {{
+ GRAPH <{PROVENANCE_GRAPH}> {{
+ {{ <{activity}> ?p ?o . BIND(<{activity}> AS ?s) }}
+ UNION
+ {{ <{bundle}> ?p ?o . BIND(<{bundle}> AS ?s) }}
+ UNION
+ {{ <{delta}> ?p ?o . BIND(<{delta}> AS ?s) }}
+ UNION
+ {{ ?s ?p ?o . FILTER(STRSTARTS(STR(?s), "{file_prefix}")) }}
+ UNION
+ {{ <{activity}> (<{PROV.wasAssociatedWith}>|<{PROV.used}>) ?s . ?s ?p ?o }}
+ UNION
+ {{ <{bundle}> <{PROV.wasAttributedTo}> ?s . ?s ?p ?o }}
+ UNION
+ {{ ?s <{PROV.used}> <{activity}> . ?s ?p ?o }}
+ UNION
+ {{ ?rec <{PROV.used}> <{activity}> ; <{PROV.wasAssociatedWith}> ?s . ?s ?p ?o }}
+ }}
+ }}
+ """
+
+
+def construct_delta_content(job_id: str) -> str:
+ """SPARQL CONSTRUCT returning the exact triples a job added (its delta graph)."""
+ dg = delta_graph_for(job_id)
+ return f"CONSTRUCT {{ ?s ?p ?o }} WHERE {{ GRAPH <{dg}> {{ ?s ?p ?o }} }}"
+
+
+async def _fetch_construct_graph(construct_query: str) -> Optional[Graph]:
+ """Run a CONSTRUCT and return the result parsed into an rdflib Graph."""
+ try:
+ endpoint = _get_endpoint("get")
+ auth = get_oxigraph_auth()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=10.0)) as client:
+ resp = await client.post(
+ endpoint,
+ data={"query": construct_query},
+ headers={"Accept": "text/turtle"},
+ auth=auth,
+ )
+ if resp.status_code != 200:
+ logger.warning(
+ "[provenance] CONSTRUCT fetch failed (HTTP %s): %s",
+ resp.status_code, (resp.text or "")[:500],
+ )
+ return None
+ g = Graph()
+ g.parse(data=resp.text, format="turtle")
+ return g
+ except Exception as e:
+ logger.warning(f"[provenance] Error fetching graph: {e}", exc_info=True)
+ return None
+
+
+async def delta_history_for_graph(named_graph_iri: str) -> Optional[List[Dict[str, Any]]]:
+ """
+ Return the change history for a named graph: one entry per ingestion delta
+ (job, added triple count, timestamp, status), newest first.
+ """
+ query = f"""
+ PREFIX prov: <{str(PROV)}>
+ PREFIX brainkb: <{str(BRAINKB)}>
+ SELECT ?delta ?activity ?count ?time ?status ?deltaGraph
+ WHERE {{
+ GRAPH <{PROVENANCE_GRAPH}> {{
+ ?delta a brainkb:IngestionDelta ;
+ brainkb:targetGraph <{named_graph_iri}> ;
+ prov:wasGeneratedBy ?activity .
+ OPTIONAL {{ ?delta brainkb:addedTripleCount ?count }}
+ OPTIONAL {{ ?delta prov:generatedAtTime ?time }}
+ OPTIONAL {{ ?delta brainkb:deltaGraph ?deltaGraph }}
+ OPTIONAL {{ ?activity brainkb:jobStatus ?status }}
+ }}
+ }}
+ ORDER BY DESC(?time)
+ """
+ try:
+ endpoint = _get_endpoint("get")
+ auth = get_oxigraph_auth()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(30.0, connect=10.0)) as client:
+ resp = await client.post(
+ endpoint,
+ data={"query": query},
+ headers={"Accept": "application/sparql-results+json"},
+ auth=auth,
+ )
+ if resp.status_code != 200:
+ return None
+ out = []
+ for b in resp.json().get("results", {}).get("bindings", []):
+ out.append({
+ "delta": b.get("delta", {}).get("value"),
+ "activity": b.get("activity", {}).get("value"),
+ "added_triple_count": int(b["count"]["value"]) if "count" in b else None,
+ "generated_at": b.get("time", {}).get("value"),
+ "status": b.get("status", {}).get("value"),
+ "delta_graph": b.get("deltaGraph", {}).get("value"),
+ })
+ return out
+ except Exception as e:
+ logger.warning(f"[provenance] Error fetching delta history: {e}", exc_info=True)
+ return None
+
+
+async def compare_deltas(job_id_a: str, job_id_b: str) -> Optional[Dict[str, Any]]:
+ """
+ Compare the triples added by two jobs. Returns counts and the differing
+ triples (A-only / B-only) as JSON-LD, plus the shared-triple count.
+ """
+ ga = await _fetch_construct_graph(construct_delta_content(job_id_a))
+ gb = await _fetch_construct_graph(construct_delta_content(job_id_b))
+ if ga is None or gb is None:
+ return None
+
+ only_a = ga - gb # in A, not in B
+ only_b = gb - ga # in B, not in A
+ common = ga & gb # in both
+
+ def _jsonld(graph: Graph):
+ graph.bind("prov", PROV)
+ graph.bind("brainkb", BRAINKB)
+ graph.bind("dcterms", DCTERMS)
+ return json.loads(graph.serialize(format="json-ld", auto_compact=True))
+
+ return {
+ "job_a": job_id_a,
+ "job_b": job_id_b,
+ "a_total": len(ga),
+ "b_total": len(gb),
+ "only_in_a_count": len(only_a),
+ "only_in_b_count": len(only_b),
+ "common_count": len(common),
+ "only_in_a": _jsonld(only_a),
+ "only_in_b": _jsonld(only_b),
+ }
+
+
+def construct_for_named_graph(named_graph_iri: str) -> str:
+ """SPARQL CONSTRUCT for all activity that targeted a given named graph."""
+ return f"""
+ CONSTRUCT {{ ?activity ?p ?o . ?ent ?ep ?eo }}
+ WHERE {{
+ GRAPH <{PROVENANCE_GRAPH}> {{
+ ?activity <{BRAINKB.targetGraph}> <{named_graph_iri}> ;
+ ?p ?o .
+ OPTIONAL {{ ?ent <{PROV.wasGeneratedBy}> ?activity ; ?ep ?eo . }}
+ }}
+ }}
+ """
diff --git a/query_service/core/rbac.py b/query_service/core/rbac.py
new file mode 100644
index 0000000..91480a3
--- /dev/null
+++ b/query_service/core/rbac.py
@@ -0,0 +1,177 @@
+# -*- coding: utf-8 -*-
+# -----------------------------------------------------------------------------
+# DISCLAIMER: This software is provided "as is" without any warranty,
+# express or implied, including but not limited to the warranties of
+# merchantability, fitness for a particular purpose, and non-infringement.
+# -----------------------------------------------------------------------------
+
+# @Author : Tek Raj Chhetri
+# @Email : tekraj@mit.edu
+# @File : rbac.py
+
+"""
+Role-based access control for BrainKB.
+
+Separation of concerns:
+ * JWT -> authentication + API access (who you are; can you call the API).
+ * Roles -> authorization (what you may DO): create spaces, ingest, admin, etc.
+
+Roles live in the Django-owned RBAC tables and are joined to a JWT user by email:
+ Web_jwtuser.email == Web_user_profile.email
+ Web_user_profile.id -> Web_user_role (active, non-expired) -> role name
+
+Roles map to capabilities via the policy below. Admins/SuperAdmins can also grant
+extra capabilities to individual users (delegated upgrades) via
+`user_capability_grants` — e.g. letting a Curator create/manage team spaces.
+
+A JWT user with NO active role is treated as having NO capabilities: read-only
+access to PUBLIC content, nothing else.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from typing import Optional, Set
+
+from core.database import get_db_connection
+
+logger = logging.getLogger(__name__)
+
+# ---- capabilities ----------------------------------------------------------
+CREATE_PRIVATE_SPACE = "create_private_space" # make your own individual/private space
+CREATE_TEAM_SPACE = "create_team_space" # create a team space (Admin/SuperAdmin or granted)
+MANAGE_TEAM_SPACE = "manage_team_space" # manage members/visibility/graphs of team spaces
+INGEST = "ingest" # ingest data (still needs per-space write membership)
+RECOVER = "recover" # recover stuck/errored jobs
+SPARQL_ADMIN = "sparql_admin" # run arbitrary SPARQL
+GRANT = "grant" # grant/revoke capabilities to other users
+READ_PRIVATE = "read_private" # read non-public content you're a member of
+
+ALL_CAPS = {
+ CREATE_PRIVATE_SPACE, CREATE_TEAM_SPACE, MANAGE_TEAM_SPACE, INGEST,
+ RECOVER, SPARQL_ADMIN, GRANT, READ_PRIVATE,
+}
+
+# Capabilities an admin may DELEGATE to another user via the grant endpoint.
+# Admin-intrinsic caps (grant, sparql_admin) are intentionally NOT delegatable:
+# they come only from an Admin/SuperAdmin role, so the grant endpoint cannot be
+# used to escalate a non-admin into an admin.
+GRANTABLE_CAPS = {
+ CREATE_PRIVATE_SPACE, CREATE_TEAM_SPACE, MANAGE_TEAM_SPACE, INGEST,
+ RECOVER, READ_PRIVATE,
+}
+
+# ---- role hierarchy / policy ----------------------------------------------
+# Hierarchy (highest first): SuperAdmin >= Admin > write roles > read roles > none.
+# SuperAdmin is the ultimate authority and is bootstrapped at deployment. Both
+# SuperAdmin and Admin get every KG capability here; the *difference* between them
+# (e.g. SuperAdmin creating/removing Admins) is ROLE ASSIGNMENT, which is owned by
+# the usermanagement/Django side — query_service never assigns roles, it only
+# reads them and grants delegatable KG capabilities. This keeps the two systems
+# consistent and prevents privilege escalation through the KG API.
+SUPERADMIN_ROLE = "SuperAdmin"
+ADMIN_ROLES = {"Admin", "SuperAdmin"}
+# Roles that confer "write" (may create their own private space + ingest).
+WRITE_ROLES = {
+ "Admin", "SuperAdmin", "Curator", "Lab Member", "Submitter",
+ "Annotator", "Mapper", "Knowledge Contributor",
+}
+
+
+def _caps_for_role(role: str) -> Set[str]:
+ if role in ADMIN_ROLES:
+ return set(ALL_CAPS) # admins can do everything
+ if role in WRITE_ROLES:
+ return {CREATE_PRIVATE_SPACE, INGEST, RECOVER, READ_PRIVATE}
+ # any other active role (Reviewer, Validator, Moderator, ...) = read member content
+ return {READ_PRIVATE}
+
+
+# ---------------------------------------------------------------------------
+# lookups
+# ---------------------------------------------------------------------------
+
+async def active_roles(email: Optional[str]) -> Set[str]:
+ """Active, non-expired role names for the JWT user's email (via profile)."""
+ if not email:
+ return set()
+ async with get_db_connection() as conn:
+ rows = await conn.fetch(
+ """
+ SELECT r.role
+ FROM "Web_user_role" r
+ JOIN "Web_user_profile" p ON p.id = r.profile_id
+ WHERE p.email = $1
+ AND r.is_active IS TRUE
+ AND (r.expires_at IS NULL OR r.expires_at > now())
+ """,
+ email,
+ )
+ return {row["role"] for row in rows}
+
+
+async def granted_capabilities(email: Optional[str]) -> Set[str]:
+ """Extra capabilities granted directly to this user (delegated upgrades)."""
+ if not email:
+ return set()
+ async with get_db_connection() as conn:
+ rows = await conn.fetch(
+ "SELECT capability FROM user_capability_grants WHERE member = $1",
+ email,
+ )
+ return {row["capability"] for row in rows if row["capability"] in ALL_CAPS}
+
+
+async def capabilities(email: Optional[str]) -> Set[str]:
+ """Effective capabilities = role-derived caps ∪ delegated grants."""
+ caps: Set[str] = set()
+ roles = await active_roles(email)
+ for r in roles:
+ caps |= _caps_for_role(r)
+ if roles: # only users with at least one role can be granted extras
+ caps |= await granted_capabilities(email)
+ return caps
+
+
+async def has_capability(email: Optional[str], cap: str) -> bool:
+ return cap in await capabilities(email)
+
+
+async def is_admin(email: Optional[str]) -> bool:
+ return bool(await active_roles(email) & ADMIN_ROLES)
+
+
+# ---------------------------------------------------------------------------
+# delegated grants (admin only — enforced at the endpoint)
+# ---------------------------------------------------------------------------
+
+async def grant_capability(member: str, capability: str, granted_by: str) -> None:
+ if capability not in GRANTABLE_CAPS:
+ raise ValueError(f"capability is not delegatable: {capability}")
+ async with get_db_connection() as conn:
+ await conn.execute(
+ """
+ INSERT INTO user_capability_grants (member, capability, granted_by, created_at)
+ VALUES ($1, $2, $3, $4)
+ ON CONFLICT (member, capability) DO NOTHING
+ """,
+ member, capability, granted_by, time.time(),
+ )
+
+
+async def revoke_capability(member: str, capability: str) -> None:
+ async with get_db_connection() as conn:
+ await conn.execute(
+ "DELETE FROM user_capability_grants WHERE member = $1 AND capability = $2",
+ member, capability,
+ )
+
+
+async def list_grants(member: str) -> list:
+ async with get_db_connection() as conn:
+ rows = await conn.fetch(
+ "SELECT capability, granted_by, created_at FROM user_capability_grants WHERE member = $1",
+ member,
+ )
+ return [dict(r) for r in rows]
diff --git a/query_service/core/routers/insert.py b/query_service/core/routers/insert.py
index d765c52..de27068 100644
--- a/query_service/core/routers/insert.py
+++ b/query_service/core/routers/insert.py
@@ -18,7 +18,7 @@
from fastapi import APIRouter, Request, HTTPException, status, UploadFile, File, Form, BackgroundTasks, Query, Body
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, Response
from core.graph_database_connection_manager import insert_data_gdb, insert_data_gdb_async
import logging
from core.pydantic_schema import InputKGTripleSchema, NamedGraphSchema
@@ -48,6 +48,24 @@
batch_insert_job_results,
)
from core.configuration import load_environment
+from core.spaces import authorize as authorize_space_access
+from core import spaces as _spaces
+from core import rbac
+from core.provenance import (
+ build_ingestion_provenance,
+ build_recovery_provenance,
+ agent_ref,
+ write_provenance,
+ query_provenance_jsonld,
+ construct_for_job,
+ construct_for_named_graph,
+ delta_graph_for,
+ merge_delta_into_target,
+ count_graph_triples,
+ construct_delta_content,
+ delta_history_for_graph,
+ compare_deltas,
+)
import datetime
import uuid
import asyncio
@@ -61,6 +79,14 @@
router = APIRouter()
logger = logging.getLogger(__name__)
+
+def _agent_email(user):
+ """Caller's identity string (email), used for space membership checks."""
+ try:
+ return user["email"]
+ except (KeyError, TypeError, IndexError):
+ return None
+
# Global dictionary to track running background tasks
# Maps job_id -> asyncio.Task for checking if job is actually running
_running_job_tasks: Dict[str, asyncio.Task] = {}
@@ -77,6 +103,28 @@
# Supported file extensions
SUPPORTED_EXTS = {"ttl", "turtle", "nt", "nq", "trig", "rdf", "owl", "jsonld", "json"}
+# Triple-level change tracking. When enabled, each job stages its triples in a
+# per-job delta graph (an exact, queryable record of what the job added) which is
+# then merged into the target graph. Costs extra storage (delta graphs persist);
+# set TRACK_TRIPLE_DELTAS=false to upload directly to the target instead.
+TRACK_TRIPLE_DELTAS = os.getenv("TRACK_TRIPLE_DELTAS", "true").strip().lower() in ("1", "true", "yes", "on")
+
+# Resource-safety cap: maximum ingest jobs PROCESSING concurrently per worker
+# process. Ingestion stays fire-and-forget (submit and forget) — this just bounds
+# how many jobs run at once so a burst of submissions can't exhaust memory, the DB
+# pool, or Oxigraph. Excess jobs return immediately and wait as 'pending' until a
+# slot frees (backpressure without a queue). Effective global cap ≈ this × workers.
+MAX_CONCURRENT_INGEST_JOBS = int(os.getenv("MAX_CONCURRENT_INGEST_JOBS", "3"))
+_ingest_semaphore: Optional[asyncio.Semaphore] = None
+
+
+def _get_ingest_semaphore() -> asyncio.Semaphore:
+ """Lazily create the per-worker ingest concurrency limiter (binds to the loop)."""
+ global _ingest_semaphore
+ if _ingest_semaphore is None:
+ _ingest_semaphore = asyncio.Semaphore(MAX_CONCURRENT_INGEST_JOBS)
+ return _ingest_semaphore
+
# Ensure job directory exists
os.makedirs(JOB_BASE_DIR, exist_ok=True)
@@ -279,29 +327,13 @@ async def upload_single_file_path(
total_files=total_files,
)
- # Process file (attach provenance if text-based RDF format)
- if job_id and not skip_provenance:
- from core.database import insert_processing_log
- status_msg = f"Attaching provenance to {filename}"
- await update_job_processing_state(
- job_id=job_id,
- current_stage="attaching_provenance",
- status_message=status_msg
- )
- # Log to history
- await insert_processing_log(
- job_id=job_id,
- file_name=filename,
- stage="attaching_provenance",
- status_message=status_msg,
- file_index=file_index,
- total_files=total_files,
- )
-
- processed_filepath, provenance_success = await process_file_with_provenance(
- filepath, user_id, ext, skip_provenance=skip_provenance
- )
-
+ # Provenance is tracked natively as PROV-O in Oxigraph's dedicated provenance
+ # graph (see core/provenance.py and PROVENANCE_MODEL.md), NOT embedded into the
+ # domain data. The uploaded file is therefore sent to Oxigraph unmodified.
+ # `skip_provenance` is retained on the API for backward compatibility but no
+ # longer controls domain-data embedding (which has been removed).
+ processed_filepath = filepath
+
# Update processing state: Uploading file
if job_id:
from core.database import insert_processing_log
@@ -440,23 +472,6 @@ async def upload_single_file_path(
if resp_text and len(resp_text) > max_len:
resp_text = resp_text[:max_len] + "... [truncated]"
- # Surface provenance-attachment failures. process_file_with_provenance returns
- # success=False when it was asked to attach provenance but rdflib parsing failed;
- # in that case the ORIGINAL (un-provenanced) file was uploaded. Previously this
- # was silently swallowed and the job still reported success, hiding a data-integrity
- # gap. We now flag it explicitly so the job result records it.
- provenance_requested = not skip_provenance and ext in [
- "ttl", "turtle", "nt", "nq", "jsonld", "json", "rdf", "owl"
- ]
- provenance_failed = provenance_requested and not provenance_success
- if provenance_failed:
- warning = (
- f"WARNING: provenance could not be attached to {filename} "
- f"(RDF parsing failed); the original file was uploaded WITHOUT provenance metadata. "
- )
- logger.warning(f"[upload_single_file_path] {warning.strip()}")
- resp_text = warning + (resp_text or "")
-
return {
"file": filename,
"ext": ext,
@@ -464,8 +479,6 @@ async def upload_single_file_path(
"elapsed_s": elapsed,
"http_status": resp.status_code,
"success": success,
- "provenance_attached": provenance_requested and provenance_success,
- "provenance_requested": provenance_requested,
"bps": bps,
"response_body": resp_text,
}
@@ -476,6 +489,27 @@ async def run_ingest_job(
max_concurrency: int,
user_id: str,
skip_provenance: bool = False,
+):
+ """Background ingest runner, gated by a per-worker concurrency limiter so a
+ burst of concurrent submissions cannot exhaust resources and crash the process.
+
+ While waiting for a slot the job stays 'pending' (accurate — it is queued). The
+ actual work runs in _run_ingest_job_body once a slot is acquired."""
+ sem = _get_ingest_semaphore()
+ if sem.locked():
+ logger.info(
+ f"[run_ingest_job] Job {job_id} is waiting for an ingest slot "
+ f"(max {MAX_CONCURRENT_INGEST_JOBS} concurrent/worker)"
+ )
+ async with sem:
+ await _run_ingest_job_body(job_id, max_concurrency, user_id, skip_provenance)
+
+
+async def _run_ingest_job_body(
+ job_id: str,
+ max_concurrency: int,
+ user_id: str,
+ skip_provenance: bool = False,
):
"""Background job runner for file ingestion.
Processes files (attaches provenance) and uploads them to Oxigraph.
@@ -488,7 +522,58 @@ async def run_ingest_job(
# This prevents jobs from running indefinitely
MAX_JOB_TIMEOUT = 2 * 60 * 60 # 2 hours
job_start_time = time.time()
-
+
+ delta_graph = delta_graph_for(job_id)
+
+ async def _write_ingestion_prov(status_label: str, results):
+ """Best-effort finalizer: merge the per-job delta graph into the target,
+ then record this job as a PROV-O IngestionActivity (+ IngestionDelta) in
+ Oxigraph. Never raises — a provenance failure must not fail the job."""
+ try:
+ details = await get_job_details(job_id)
+ named_graph = details.get("graph") if details else None
+ if not named_graph:
+ return
+ results = results or []
+ succ = sum(1 for r in results if r.get("success"))
+ fail = len(results) - succ
+
+ added_count = None
+ effective_delta_graph = None
+ if TRACK_TRIPLE_DELTAS:
+ # Count what the job staged, then merge the delta into the target graph.
+ added_count = await count_graph_triples(delta_graph)
+ if added_count and added_count > 0:
+ await merge_delta_into_target(delta_graph, named_graph)
+ effective_delta_graph = delta_graph
+
+ prov_graph = build_ingestion_provenance(
+ job_id=job_id,
+ agent_id=user_id,
+ named_graph_iri=named_graph,
+ started_at=datetime.datetime.fromtimestamp(job_start_time, datetime.timezone.utc).isoformat(),
+ ended_at=datetime.datetime.now(datetime.timezone.utc).isoformat(),
+ status=status_label,
+ total_files=len(results),
+ success_count=succ,
+ fail_count=fail,
+ results=results,
+ agent_type="user",
+ delta_graph=effective_delta_graph,
+ added_triple_count=added_count,
+ )
+ await write_provenance(prov_graph)
+
+ # Queue search indexing to run in the BACKGROUND (do not block the job
+ # on it — indexing a large graph can be slow). Best-effort enqueue.
+ try:
+ from core.indexing import enqueue_ingest
+ await enqueue_ingest(named_graph, effective_delta_graph)
+ except Exception as _se:
+ logger.warning(f"[run_ingest_job] Failed to queue search indexing for {job_id}: {_se}")
+ except Exception as _pe:
+ logger.warning(f"[run_ingest_job] Provenance write failed for {job_id}: {_pe}")
+
try:
# Mark job as running (start_time was already set when job was created)
from core.database import update_job_processing_state, insert_processing_log
@@ -513,7 +598,10 @@ async def run_ingest_job(
job_dir = job_details["job_dir"]
graph = job_details["graph"]
-
+ # When delta tracking is on, stage uploads in the per-job delta graph;
+ # _write_ingestion_prov merges it into the target graph at the end.
+ upload_graph = delta_graph if TRACK_TRIPLE_DELTAS else graph
+
# Collect files in job_dir (exclude .processed files)
file_infos: List[Dict[str, Any]] = []
for name in os.listdir(job_dir):
@@ -553,7 +641,7 @@ async def worker(fi: Dict[str, Any], index: int):
client,
filepath,
size,
- graph,
+ upload_graph,
user_id,
skip_provenance=skip_provenance,
job_id=job_id,
@@ -615,8 +703,13 @@ async def worker(fi: Dict[str, Any], index: int):
status_message=status_msg,
)
await update_job_status(job_id, "done", end_time=time.time())
+ # Record PROV-O ingestion provenance in Oxigraph (source of truth)
+ _succ = sum(1 for r in all_results if r.get("success"))
+ _fail = len(all_results) - _succ
+ _status_label = "done" if _fail == 0 else ("failed" if _succ == 0 else "partial")
+ await _write_ingestion_prov(_status_label, all_results)
logger.info(f"[run_ingest_job] Job {job_id} completed successfully")
-
+
except asyncio.TimeoutError as e:
# Mark job as errored due to timeout
from core.database import update_job_processing_state, insert_processing_log
@@ -632,6 +725,7 @@ async def worker(fi: Dict[str, Any], index: int):
status_message=status_msg,
)
await update_job_status(job_id, "error", end_time=time.time())
+ await _write_ingestion_prov("error", [])
logger.error(f"[run_ingest_job] Job {job_id} timed out: {e}", exc_info=True)
except Exception as e:
# Mark job as errored
@@ -648,6 +742,7 @@ async def worker(fi: Dict[str, Any], index: int):
status_message=status_msg,
)
await update_job_status(job_id, "error", end_time=time.time())
+ await _write_ingestion_prov("error", [])
logger.error(f"[run_ingest_job] Job {job_id} failed: {e}", exc_info=True)
finally:
# Ensure job status is always updated, even if something goes wrong
@@ -980,7 +1075,15 @@ async def recover_stuck_jobs(
f"The job was marked as 'error' to prevent it from running indefinitely."
),
)
-
+
+ # Record the automated recovery as a PROV-O RecoveryActivity (system agent)
+ try:
+ await write_provenance(
+ build_recovery_provenance(job_id=job_id, cause=cause)
+ )
+ except Exception as _pe:
+ logger.warning(f"[recover_stuck_jobs] Provenance write failed for {job_id}: {_pe}")
+
logger.info(f"[recover_stuck_jobs] Recovered {len(stuck_jobs)} stuck job(s) from server crash/restart")
return len(stuck_jobs)
else:
@@ -1077,12 +1180,37 @@ async def insert_knowledge_graph_triples(
},
status_code=400,
)
-
+
+ # Role-based authorization: ingesting requires a write-capable role. JWT
+ # scope is only API access — it does not by itself grant permission to ingest.
+ if not await rbac.has_capability(_agent_email(user), rbac.INGEST):
+ return JSONResponse(
+ {"error": "Not authorized to ingest: a write-capable role is required."},
+ status_code=403,
+ )
+ # If the graph belongs to a space, enforce space write-authorization
+ # (owner/editor). Unmapped legacy graphs fall through (scope check applies).
+ _graph_key = named_graph_iri if named_graph_iri.endswith("/") else named_graph_iri + "/"
+ _allowed, _reason = await authorize_space_access(_graph_key, _agent_email(user), "write")
+ if not _allowed:
+ return JSONResponse(
+ {"error": f"Not authorized to ingest into this graph: {_reason}", "named_graph_iri": named_graph_iri},
+ status_code=403,
+ )
+ # Fine-grained per-space write rules (if any) further restrict who can ingest.
+ _space_for_graph = await _spaces.get_space_for_graph(_graph_key)
+ if _space_for_graph and not await _spaces.space_action_permitted(_space_for_graph, "write", _agent_email(user)):
+ return JSONResponse(
+ {"error": "Not authorized to ingest into this graph: restricted by a space access rule.",
+ "named_graph_iri": named_graph_iri},
+ status_code=403,
+ )
+
job_id = uuid.uuid4().hex
-
+
# Get Oxigraph endpoint from configuration
endpoint = get_oxigraph_endpoint()
-
+
raw_bytes = data.encode("utf-8")
if len(raw_bytes) > MAX_RAW_SIZE_BYTES:
return JSONResponse(
@@ -1189,7 +1317,32 @@ async def insert_file_knowledge_graph_triples(
},
status_code=400,
)
-
+
+ # Role-based authorization: ingesting requires a write-capable role. JWT
+ # scope is only API access — it does not by itself grant permission to ingest.
+ if not await rbac.has_capability(_agent_email(user), rbac.INGEST):
+ return JSONResponse(
+ {"error": "Not authorized to ingest: a write-capable role is required."},
+ status_code=403,
+ )
+ # If the graph belongs to a space, enforce space write-authorization
+ # (owner/editor). Unmapped legacy graphs fall through (scope check applies).
+ _graph_key = named_graph_iri if named_graph_iri.endswith("/") else named_graph_iri + "/"
+ _allowed, _reason = await authorize_space_access(_graph_key, _agent_email(user), "write")
+ if not _allowed:
+ return JSONResponse(
+ {"error": f"Not authorized to ingest into this graph: {_reason}", "named_graph_iri": named_graph_iri},
+ status_code=403,
+ )
+ # Fine-grained per-space write rules (if any) further restrict who can ingest.
+ _space_for_graph = await _spaces.get_space_for_graph(_graph_key)
+ if _space_for_graph and not await _spaces.space_action_permitted(_space_for_graph, "write", _agent_email(user)):
+ return JSONResponse(
+ {"error": "Not authorized to ingest into this graph: restricted by a space access rule.",
+ "named_graph_iri": named_graph_iri},
+ status_code=403,
+ )
+
job_id = uuid.uuid4().hex # generate for job tracking
# Get Oxigraph endpoint from configuration
@@ -1336,7 +1489,8 @@ async def run_and_track_job():
}
@router.get("/insert/jobs",
- include_in_schema=True
+ include_in_schema=True,
+ dependencies=[Depends(require_scopes(["read"]))],
)
async def list_jobs(
user: Annotated[LoginUserIn, Depends(get_current_user)],
@@ -1362,7 +1516,8 @@ async def list_jobs(
@router.get("/insert/user/jobs/detail",
- include_in_schema=True
+ include_in_schema=True,
+ dependencies=[Depends(require_scopes(["read"]))],
)
async def get_job_detail(
user: Annotated[LoginUserIn, Depends(get_current_user)],
@@ -1515,7 +1670,8 @@ async def get_job_detail(
@router.get("/insert/jobs/check-recoverable",
- include_in_schema=True
+ include_in_schema=True,
+ dependencies=[Depends(require_scopes(["read"]))],
)
async def check_job_recoverable_endpoint(
user: Annotated[LoginUserIn, Depends(get_current_user)],
@@ -1554,7 +1710,8 @@ async def check_job_recoverable_endpoint(
@router.post("/insert/jobs/recover",
- include_in_schema=True
+ include_in_schema=True,
+ dependencies=[Depends(require_scopes(["write"]))],
)
async def recover_stuck_jobs_endpoint(
user: Annotated[LoginUserIn, Depends(get_current_user)],
@@ -1590,6 +1747,11 @@ async def recover_stuck_jobs_endpoint(
Returns the number of jobs recovered and details about recovered jobs.
"""
verify_user_access(user_id, user)
+ if not await rbac.has_capability(_agent_email(user), rbac.RECOVER):
+ return JSONResponse(
+ {"error": "Not authorized to recover jobs: a write-capable role is required."},
+ status_code=403,
+ )
try:
# For single job recovery, MUST check recoverability first (includes process check)
if job_id:
@@ -1714,7 +1876,9 @@ async def recover_stuck_jobs_endpoint(
)
-@router.post("/register-named-graph")
+@router.post("/register-named-graph",
+ dependencies=[Depends(require_scopes(["write"]))],
+ )
async def create_named_graph(
user: Annotated[LoginUserIn, Depends(get_current_user)],
request: NamedGraphSchema
@@ -1743,10 +1907,20 @@ async def create_named_graph(
"""
named_graph_exists = await fetch_data_gdb_async(query)
if not named_graph_exists.get("message", {}).get("boolean", False):
+ # Attribute the registration to the authenticated user (PROV-O). This is
+ # recorded on the registry entry itself (see named_graph_metadata) rather
+ # than duplicated as a separate activity in the provenance graph.
+ try:
+ agent_id = user["email"] or user["id"]
+ except (KeyError, TypeError, IndexError):
+ agent_id = "unknown"
+ agent_uri = str(agent_ref(str(agent_id)))
+
# Register the new named graph
response = await insert_data_gdb_async(named_graph_metadata(
named_graph_url=named_graph_url,
description=description,
+ agent_uri=agent_uri,
)
)
return response
@@ -1765,3 +1939,143 @@ async def create_named_graph(
detail=f"An error occurred processing the request {e}",
)
+
+@router.get(
+ "/provenance/job",
+ include_in_schema=True,
+ dependencies=[Depends(require_scopes(["read"]))],
+ summary="PROV-O provenance bundle for one ingestion job",
+ description=(
+ "Returns the **W3C PROV-O bundle** (JSON-LD) for a single ingestion job: "
+ "the `IngestionActivity` (with agent, start/end time, status, file counts), "
+ "the generated bundle entity, each per-file entity (upload status, HTTP "
+ "status, size), the `IngestionDelta` entity, and any recovery activity that "
+ "acted on the job. Access-controlled — the `user_id` must match the "
+ "authenticated caller."
+ ),
+)
+async def get_job_provenance(
+ user: Annotated[LoginUserIn, Depends(get_current_user)],
+ user_id: Annotated[str, Query(..., description="User identifier (must match the authenticated user)")],
+ job_id: Annotated[str, Query(..., description="Job identifier to fetch provenance for")],
+):
+ """
+ Return the full PROV-O provenance bundle (JSON-LD) for a single ingestion job,
+ reconstructed from the dedicated provenance graph in Oxigraph.
+
+ Scope: everything about ONE job (activity + bundle + files + delta + recovery).
+ For a whole graph's history use /provenance/named-graph.
+ """
+ verify_user_access(user_id, user)
+ jsonld = await query_provenance_jsonld(construct_for_job(job_id))
+ if jsonld is None:
+ return JSONResponse(
+ {"error": "Failed to retrieve provenance from the graph database"},
+ status_code=502,
+ )
+ return Response(content=jsonld, media_type="application/ld+json")
+
+
+@router.get(
+ "/provenance/named-graph",
+ include_in_schema=True,
+ dependencies=[Depends(require_scopes(["read"]))],
+ summary="PROV-O activity history for a named graph",
+ description=(
+ "Returns the **W3C PROV-O provenance** (JSON-LD) describing how a named "
+ "graph's data came to be: every ingestion activity that targeted it, with "
+ "the agent, start/end times, per-file entities, and job status.\n\n"
+ "Difference from `GET /api/query/registered-named-graphs`:\n"
+ "- **registered-named-graphs** = the *registry/catalog* — which graphs "
+ "exist and their registration metadata (one row per graph).\n"
+ "- **provenance/named-graph** = the *activity history* — what was ingested "
+ "into a given graph, when, and by whom (a PROV-O bundle, potentially many "
+ "activities). Reads the provenance graph `https://brainkb.org/provenance/`."
+ ),
+)
+async def get_named_graph_provenance(
+ user: Annotated[LoginUserIn, Depends(get_current_user)],
+ iri: Annotated[str, Query(..., description="Named graph IRI to fetch the ingestion/activity provenance for")],
+):
+ """
+ Return the PROV-O provenance (JSON-LD) for every ingestion activity that
+ targeted the given named graph.
+
+ This is the *history of data mutations* on the graph, distinct from the
+ registry catalog returned by /api/query/registered-named-graphs. Registration
+ attribution lives on the registry entry (see that endpoint's `registered_by`).
+ """
+ jsonld = await query_provenance_jsonld(construct_for_named_graph(iri))
+ if jsonld is None:
+ return JSONResponse(
+ {"error": "Failed to retrieve provenance from the graph database"},
+ status_code=502,
+ )
+ return Response(content=jsonld, media_type="application/ld+json")
+
+
+@router.get("/provenance/delta", include_in_schema=True,
+ dependencies=[Depends(require_scopes(["read"]))])
+async def get_job_delta(
+ user: Annotated[LoginUserIn, Depends(get_current_user)],
+ user_id: Annotated[str, Query(..., description="User identifier (must match the authenticated user)")],
+ job_id: Annotated[str, Query(..., description="Job identifier whose added triples to return")],
+):
+ """
+ GET /provenance/delta
+ Return the exact set of triples a job added (its delta graph) as JSON-LD.
+ This is the incremental change that job contributed to the target graph.
+ """
+ verify_user_access(user_id, user)
+ jsonld = await query_provenance_jsonld(construct_delta_content(job_id))
+ if jsonld is None:
+ return JSONResponse(
+ {"error": "Failed to retrieve delta from the graph database"},
+ status_code=502,
+ )
+ return Response(content=jsonld, media_type="application/ld+json")
+
+
+@router.get("/provenance/delta/history", include_in_schema=True,
+ dependencies=[Depends(require_scopes(["read"]))])
+async def get_delta_history(
+ user: Annotated[LoginUserIn, Depends(get_current_user)],
+ iri: Annotated[str, Query(..., description="Named graph IRI to list the change history for")],
+):
+ """
+ GET /provenance/delta/history
+ Return the ordered change history of a named graph: one entry per ingestion
+ delta (job, added triple count, timestamp, status), newest first.
+ """
+ history = await delta_history_for_graph(iri)
+ if history is None:
+ return JSONResponse(
+ {"error": "Failed to retrieve delta history from the graph database"},
+ status_code=502,
+ )
+ return {"named_graph_iri": iri, "changes": history, "total": len(history)}
+
+
+@router.get("/provenance/delta/compare", include_in_schema=True,
+ dependencies=[Depends(require_scopes(["read"]))])
+async def compare_job_deltas(
+ user: Annotated[LoginUserIn, Depends(get_current_user)],
+ user_id: Annotated[str, Query(..., description="User identifier (must match the authenticated user)")],
+ job_id_a: Annotated[str, Query(..., description="First job identifier")],
+ job_id_b: Annotated[str, Query(..., description="Second job identifier")],
+):
+ """
+ GET /provenance/delta/compare
+ Compare the triples added by two jobs. Returns counts (A-only, B-only,
+ shared) and the differing triples as JSON-LD, so users can see exactly how
+ two ingestion changes differ.
+ """
+ verify_user_access(user_id, user)
+ result = await compare_deltas(job_id_a, job_id_b)
+ if result is None:
+ return JSONResponse(
+ {"error": "Failed to compare deltas (one or both delta graphs unavailable)"},
+ status_code=502,
+ )
+ return result
+
diff --git a/query_service/core/routers/query.py b/query_service/core/routers/query.py
index 88c521d..907d0f7 100644
--- a/query_service/core/routers/query.py
+++ b/query_service/core/routers/query.py
@@ -16,13 +16,14 @@
# @File : query.py
# @Software: PyCharm
-from fastapi import APIRouter
+from fastapi import APIRouter, HTTPException
from core.graph_database_connection_manager import fetch_data_gdb_async, check_named_graph_exists
import logging
from typing import Annotated
from core.models.user import LoginUserIn
from core.security import get_current_user, require_scopes
from core.shared import taxonomy_postprocessing
+from core import rbac
from fastapi import Depends
from pydantic import BaseModel, root_validator
from typing import List
@@ -30,17 +31,42 @@
logger = logging.getLogger(__name__)
-@router.get("/query/registered-named-graphs")
-async def get_named_graphs():
+@router.get(
+ "/query/registered-named-graphs",
+ dependencies=[Depends(require_scopes(["read"]))],
+ summary="List registered named graphs (the registry/catalog)",
+ description=(
+ "Returns the **catalog of named graphs** that have been registered in "
+ "BrainKB — i.e. *which* graphs exist and may be ingested into. Each entry "
+ "carries its `description`, `registered_at` timestamp, and `registered_by` "
+ "(the user who registered it). Data is read from the registry graph "
+ "`https://brainkb.org/metadata/named-graph`.\n\n"
+ "Visibility-filtered: graphs that belong to a **private space** the caller "
+ "is not a member of are omitted (so private graph existence is not leaked). "
+ "Public-space graphs and legacy (unmapped) graphs are always listed.\n\n"
+ "This answers *\"what graphs are available?\"*. It is NOT a history of what "
+ "was ingested — for the ingestion/activity history of a specific graph, use "
+ "`GET /api/provenance/named-graph?iri=…`."
+ ),
+)
+async def get_named_graphs(user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ """List every registered named graph with its registration metadata,
+ filtered so private-space graphs the caller can't access are hidden.
+
+ Registry (catalog) view: one row per graph with description, when it was
+ registered, and by whom. Contrast with /api/provenance/named-graph, which
+ returns the PROV-O activity history (ingestions) that targeted a graph.
+ """
query_named_graph = """
PREFIX prov:
PREFIX dcterms:
- Select distinct ?graph ?description ?registered_at
+ Select distinct ?graph ?description ?registered_at ?registered_by
WHERE {
GRAPH {
?graph dcterms:description ?description;
prov:generatedAtTime ?registered_at.
- }
+ OPTIONAL { ?graph prov:wasAttributedTo ?registered_by. }
+ }
}
"""
response = await fetch_data_gdb_async(query_named_graph)
@@ -54,17 +80,45 @@ async def get_named_graphs():
response_graph[graphs_info["graph"]["value"]] = {
"graph": graphs_info["graph"]["value"],
"description": graphs_info["description"]["value"],
- "registered_at": graphs_info["registered_at"]["value"]
+ "registered_at": graphs_info["registered_at"]["value"],
+ "registered_by": graphs_info.get("registered_by", {}).get("value"),
}
+
+ # Hide graphs belonging to private spaces the caller is not a member of, so
+ # private graph existence is not leaked via the registry listing.
+ from core.spaces import hidden_graphs_for
+ try:
+ member = user["email"]
+ except (KeyError, TypeError, IndexError):
+ member = None
+ hidden = await hidden_graphs_for(member)
+ if hidden:
+ response_graph = {k: v for k, v in response_graph.items() if k not in hidden}
return response_graph
@router.get("/query/sparql/",
- dependencies=[Depends(require_scopes(["write","admin"]))],
+ dependencies=[Depends(require_scopes(["admin"]))],
+ summary="Run an arbitrary SPARQL query (admin only)",
+ description=(
+ "Executes a caller-supplied SPARQL query against the graph database. "
+ "This is a powerful, unrestricted capability, so it is gated to the "
+ "**admin** scope only — deliberately NOT the default 'read' scope that "
+ "the fixed-shape read endpoints (e.g. /query/taxonomy, /query/"
+ "registered-named-graphs) use."
+ ),
)
async def sparql_query(
user: Annotated[LoginUserIn, Depends(get_current_user)], sparql_query: str
):
+ # Authorization is role-based: arbitrary SPARQL requires the sparql_admin
+ # capability (Admin/SuperAdmin). The JWT scope only gates API access.
+ try:
+ email = user["email"]
+ except (KeyError, TypeError, IndexError):
+ email = None
+ if not await rbac.has_capability(email, rbac.SPARQL_ADMIN):
+ raise HTTPException(status_code=403, detail="Arbitrary SPARQL requires an Admin/SuperAdmin role.")
response = await fetch_data_gdb_async(sparql_query)
return response
diff --git a/query_service/core/routers/search.py b/query_service/core/routers/search.py
new file mode 100644
index 0000000..4525044
--- /dev/null
+++ b/query_service/core/routers/search.py
@@ -0,0 +1,90 @@
+# -*- coding: utf-8 -*-
+# -----------------------------------------------------------------------------
+# DISCLAIMER: This software is provided "as is" without any warranty,
+# express or implied, including but not limited to the warranties of
+# merchantability, fitness for a particular purpose, and non-infringement.
+# -----------------------------------------------------------------------------
+
+# @Author : Tek Raj Chhetri
+# @Email : tekraj@mit.edu
+# @File : search.py (router)
+
+"""Search endpoint — hybrid Postgres-locator + Oxigraph-data, access-filtered."""
+
+import logging
+from typing import Annotated, Optional
+
+from fastapi import APIRouter, Depends, Query
+
+from core.models.user import LoginUserIn
+from core.security import get_current_user, get_current_user_optional, require_scopes
+from core import search as se
+from core import indexing as ix
+
+router = APIRouter()
+logger = logging.getLogger(__name__)
+
+
+def _agent(user) -> Optional[str]:
+ if not user:
+ return None
+ try:
+ return user["email"]
+ except (KeyError, TypeError, IndexError):
+ return None
+
+
+@router.get(
+ "/search",
+ summary="Search knowledge graphs (access-filtered)",
+ description=(
+ "Full-text search over the knowledge graphs, honoring space visibility. "
+ "Postgres locates matching subjects (fast, filtered by workspace/visibility); "
+ "the matched triples are then fetched from Oxigraph.\n\n"
+ "- **Anonymous** (no token): searches **public** spaces only.\n"
+ "- **Authenticated**: public spaces + the caller's own/member (private) spaces "
+ "(and legacy/unmapped graphs).\n"
+ "- Pass `space` to scope the search to a single space you can access; omit it "
+ "for a full search across everything you may read.\n\n"
+ "Private data is never returned to non-members — the filter is enforced in the "
+ "locator query itself."
+ ),
+)
+async def search(
+ user: Annotated[Optional[object], Depends(get_current_user_optional)],
+ q: Annotated[str, Query(..., min_length=1, description="Search terms")],
+ space: Annotated[Optional[str], Query(description="Restrict to a single space slug")] = None,
+ limit: Annotated[int, Query(ge=1, le=100)] = 25,
+ offset: Annotated[int, Query(ge=0)] = 0,
+):
+ return await se.search(q=q, caller=_agent(user), space_slug=space, limit=limit, offset=offset)
+
+
+@router.post(
+ "/search/reindex",
+ dependencies=[Depends(require_scopes(["admin"]))],
+ summary="Backfill/rebuild the search index (admin) — runs in background",
+ description="Queues a background task that reindexes every user graph into the "
+ "Postgres locator index. Returns immediately with a task_id; poll "
+ "GET /search/index-tasks for progress.",
+)
+async def reindex(user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ task_id = await ix.enqueue_backfill()
+ return {"status": "queued", "task_id": task_id,
+ "message": "Backfill reindex queued; runs in the background."}
+
+
+@router.get(
+ "/search/index-tasks",
+ dependencies=[Depends(require_scopes(["read"]))],
+ summary="List background indexing tasks and their status",
+)
+async def index_tasks(
+ user: Annotated[LoginUserIn, Depends(get_current_user)],
+ task_id: Annotated[Optional[str], Query(description="Fetch a single task by id")] = None,
+ limit: Annotated[int, Query(ge=1, le=200)] = 50,
+):
+ if task_id:
+ t = await ix.get_task(task_id)
+ return t or {"error": "task not found"}
+ return {"tasks": await ix.list_tasks(limit)}
diff --git a/query_service/core/routers/spaces.py b/query_service/core/routers/spaces.py
new file mode 100644
index 0000000..3c54740
--- /dev/null
+++ b/query_service/core/routers/spaces.py
@@ -0,0 +1,357 @@
+# -*- coding: utf-8 -*-
+# -----------------------------------------------------------------------------
+# DISCLAIMER: This software is provided "as is" without any warranty,
+# express or implied, including but not limited to the warranties of
+# merchantability, fitness for a particular purpose, and non-infringement.
+# -----------------------------------------------------------------------------
+
+# @Author : Tek Raj Chhetri
+# @Email : tekraj@mit.edu
+# @File : spaces.py (router)
+
+"""REST endpoints for spaces — private/public containers of named graphs."""
+
+import logging
+import re
+from typing import Annotated, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from fastapi.responses import JSONResponse, Response
+from pydantic import BaseModel, HttpUrl
+
+from core.models.user import LoginUserIn
+from core.security import get_current_user, get_current_user_optional, require_scopes
+from core.shared import named_graph_metadata
+from core.provenance import agent_ref, query_provenance_jsonld
+from core.graph_database_connection_manager import insert_data_gdb_async, check_named_graph_exists
+from core import spaces as sp
+from core import rbac
+
+router = APIRouter()
+logger = logging.getLogger(__name__)
+
+_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{1,62}[a-z0-9]$")
+
+
+def _agent(user) -> str:
+ """Caller's identity string (email preferred), matching the PROV agent id."""
+ try:
+ return str(user["email"] or user["id"])
+ except (KeyError, TypeError, IndexError):
+ return "unknown"
+
+
+async def _can_manage(space: dict, email: str) -> bool:
+ """Who may manage a space (members/visibility/graphs): the space owner, an
+ Admin/SuperAdmin, a holder of manage_team_space (team spaces), or someone
+ matched by a per-space 'manage' access rule."""
+ if await sp.member_role(space["space_id"], email) == "owner":
+ return True
+ if await rbac.is_admin(email):
+ return True
+ if space.get("space_type") == "team" and await rbac.has_capability(email, rbac.MANAGE_TEAM_SPACE):
+ return True
+ if await sp.matches_access_rule(space["space_id"], "manage", email):
+ return True
+ return False
+
+
+class SpaceCreate(BaseModel):
+ slug: str
+ name: str
+ description: Optional[str] = None
+ visibility: str = "private"
+ space_type: str = "individual" # 'individual' | 'team'
+
+
+class MemberIn(BaseModel):
+ member: str
+ role: str = "viewer"
+
+
+class VisibilityIn(BaseModel):
+ visibility: str
+
+
+class SpaceGraphIn(BaseModel):
+ named_graph_url: HttpUrl
+ description: str = ""
+
+
+@router.post("/spaces", status_code=201,
+ dependencies=[Depends(require_scopes(["write"]))],
+ summary="Create a space",
+ description="Create an owner-controlled space (private by default). The "
+ "caller becomes its owner. Members can later be added and the "
+ "space flipped public for anonymous read access.")
+async def create_space(body: SpaceCreate, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ if not _SLUG_RE.match(body.slug):
+ raise HTTPException(400, "slug must be lowercase alphanumeric/hyphen, 3-64 chars")
+ if body.visibility not in ("private", "public"):
+ raise HTTPException(400, "visibility must be 'private' or 'public'")
+ if body.space_type not in ("individual", "team"):
+ raise HTTPException(400, "space_type must be 'individual' or 'team'")
+
+ # Role-based authorization: team spaces require create_team_space (Admin/
+ # SuperAdmin, or a user an admin has granted it); individual/private spaces
+ # require create_private_space (any write-capable role).
+ email = _agent(user)
+ if body.space_type == "team":
+ if not await rbac.has_capability(email, rbac.CREATE_TEAM_SPACE):
+ raise HTTPException(403, "not authorized to create a team space (needs Admin/SuperAdmin "
+ "or a granted create_team_space capability)")
+ else:
+ if not await rbac.has_capability(email, rbac.CREATE_PRIVATE_SPACE):
+ raise HTTPException(403, "not authorized to create a space (needs a write-capable role)")
+
+ if await sp.get_space(body.slug):
+ raise HTTPException(409, f"space '{body.slug}' already exists")
+ space = await sp.create_space(body.slug, body.name, body.description, email,
+ body.visibility, body.space_type)
+ await sp.mirror_space_to_rdf(space)
+ return space
+
+
+@router.get("/spaces",
+ summary="List visible spaces",
+ description="Lists spaces the caller may see: all public spaces plus any "
+ "the caller is a member of. Anonymous callers see only public "
+ "spaces (no token required).")
+async def list_spaces(user: Annotated[Optional[object], Depends(get_current_user_optional)]):
+ member = _agent(user) if user else None
+ return {"spaces": await sp.list_visible_spaces(member)}
+
+
+@router.get("/spaces/{slug}",
+ summary="Get a space",
+ description="Returns a space's manifest (members, graphs, visibility) if the "
+ "caller may see it — public to anyone, private to members only.")
+async def get_space(slug: str, user: Annotated[Optional[object], Depends(get_current_user_optional)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ member = _agent(user) if user else None
+ if space["visibility"] != "public":
+ # Private: needs membership AND a role that grants read_private
+ # (a JWT user with no role gets public content only).
+ role = await sp.member_role(space["space_id"], member)
+ if role is None or not await rbac.has_capability(member, rbac.READ_PRIVATE):
+ raise HTTPException(403, "private space — membership and a role are required")
+ # Fine-grained per-space read rules (if any) further restrict who can read.
+ if not await sp.space_action_permitted(space, "read", member):
+ raise HTTPException(403, "restricted by a space access rule (read)")
+ return space
+
+
+@router.patch("/spaces/{slug}/visibility",
+ dependencies=[Depends(require_scopes(["write"]))],
+ summary="Set space visibility (owner only)",
+ description="Flip a space between 'private' and 'public'. Public spaces are "
+ "readable by anyone, including unauthenticated clients.")
+async def set_visibility(slug: str, body: VisibilityIn, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ if not await _can_manage(space, _agent(user)):
+ raise HTTPException(403, "not authorized to change this space's visibility")
+ if body.visibility not in ("private", "public"):
+ raise HTTPException(400, "visibility must be 'private' or 'public'")
+ await sp.set_visibility(slug, body.visibility)
+ space = await sp.get_space(slug)
+ await sp.mirror_space_to_rdf(space)
+ return space
+
+
+@router.post("/spaces/{slug}/members",
+ dependencies=[Depends(require_scopes(["write"]))],
+ summary="Add or update a member (owner only)")
+async def add_member(slug: str, body: MemberIn, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ if not await _can_manage(space, _agent(user)):
+ raise HTTPException(403, "not authorized to manage members of this space")
+ if body.role not in sp.ROLES:
+ raise HTTPException(400, f"role must be one of {sp.ROLES}")
+ await sp.add_member(space["space_id"], body.member, body.role)
+ space = await sp.get_space(slug)
+ await sp.mirror_space_to_rdf(space)
+ return space
+
+
+@router.delete("/spaces/{slug}/members/{member}",
+ dependencies=[Depends(require_scopes(["write"]))],
+ summary="Remove a member (owner only)")
+async def remove_member(slug: str, member: str, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ if not await _can_manage(space, _agent(user)):
+ raise HTTPException(403, "not authorized to manage members of this space")
+ await sp.remove_member(space["space_id"], member)
+ space = await sp.get_space(slug)
+ await sp.mirror_space_to_rdf(space)
+ return space
+
+
+@router.post("/spaces/{slug}/graphs",
+ dependencies=[Depends(require_scopes(["write"]))],
+ summary="Register a named graph into a space (owner/editor)",
+ description="Registers a named graph and binds it to this space so that "
+ "ingestion and reads on that graph are governed by the space's "
+ "membership and visibility.")
+async def add_graph(slug: str, body: SpaceGraphIn, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ _role = await sp.member_role(space["space_id"], _agent(user))
+ if not (await _can_manage(space, _agent(user)) or _role in sp.WRITE_ROLES):
+ raise HTTPException(403, "only the space owner/editors (or a space manager) can add graphs")
+
+ named_graph_url = str(body.named_graph_url)
+ if not named_graph_url.endswith("/"):
+ named_graph_url += "/"
+
+ # Register in the graph registry if not already there (idempotent-ish).
+ if not await check_named_graph_exists(named_graph_url):
+ await insert_data_gdb_async(named_graph_metadata(
+ named_graph_url=named_graph_url,
+ description=body.description,
+ agent_uri=str(agent_ref(_agent(user))),
+ ))
+ await sp.attach_graph(space["space_id"], named_graph_url)
+ space = await sp.get_space(slug)
+ await sp.mirror_space_to_rdf(space)
+ return space
+
+
+@router.get("/spaces/{slug}/data",
+ summary="Read a space's data (public = anonymous)",
+ description="Returns the RDF (JSON-LD) across all named graphs in the space. "
+ "Public spaces are readable by anyone (no token); private spaces "
+ "require membership.")
+async def read_space_data(slug: str, user: Annotated[Optional[object], Depends(get_current_user_optional)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ member = _agent(user) if user else None
+ if space["visibility"] != "public":
+ # Private: membership AND a role that grants read_private.
+ if await sp.member_role(space["space_id"], member) is None \
+ or not await rbac.has_capability(member, rbac.READ_PRIVATE):
+ raise HTTPException(403, "private space — membership and a role are required")
+ # Fine-grained per-space read rules (if any) further restrict who can read.
+ if not await sp.space_action_permitted(space, "read", member):
+ raise HTTPException(403, "restricted by a space access rule (read)")
+ if not space["graphs"]:
+ return Response(content='{"@graph": []}', media_type="application/ld+json")
+ jsonld = await query_provenance_jsonld(await sp.construct_space_graphs(space))
+ if jsonld is None:
+ return JSONResponse({"error": "failed to read space data"}, status_code=502)
+ return Response(content=jsonld, media_type="application/ld+json")
+
+
+# --------------------------------------------------------------------------- #
+# Admin: delegated capability grants (Admin/SuperAdmin only)
+# --------------------------------------------------------------------------- #
+
+class GrantIn(BaseModel):
+ member: str
+ capability: str
+
+
+class AccessRuleIn(BaseModel):
+ action: str # read | write | manage
+ subject_type: str # global_role | member | space_role
+ subject_value: str # role name / email / viewer|editor|owner
+
+
+@router.get("/spaces/{slug}/access-rules",
+ summary="List a space's fine-grained access rules")
+async def list_access_rules(slug: str, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ email = _agent(user)
+ # visible to managers or members of the space
+ if not await _can_manage(space, email) and await sp.member_role(space["space_id"], email) is None:
+ raise HTTPException(403, "must be a member or manager of the space")
+ return {"slug": slug, "rules": await sp.list_access_rules(space["space_id"])}
+
+
+@router.post("/spaces/{slug}/access-rules",
+ dependencies=[Depends(require_scopes(["write"]))],
+ summary="Add a fine-grained access rule (space manager only)",
+ description="Restrict a space action to a subject. action: read|write|manage. "
+ "subject_type: global_role (e.g. 'Admin','Lab Member') | member "
+ "(an email) | space_role (viewer|editor|owner). When rules exist "
+ "for an action, only matching callers may perform it (owner and "
+ "Admin/SuperAdmin always bypass).")
+async def add_access_rule(slug: str, body: AccessRuleIn, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ if not await _can_manage(space, _agent(user)):
+ raise HTTPException(403, "not authorized to manage this space's access rules")
+ try:
+ await sp.add_access_rule(space["space_id"], body.action, body.subject_type,
+ body.subject_value, _agent(user))
+ except ValueError as e:
+ raise HTTPException(400, str(e))
+ return {"slug": slug, "rules": await sp.list_access_rules(space["space_id"])}
+
+
+@router.delete("/spaces/{slug}/access-rules/{rule_id}",
+ dependencies=[Depends(require_scopes(["write"]))],
+ summary="Delete a fine-grained access rule (space manager only)")
+async def delete_access_rule(slug: str, rule_id: int, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ space = await sp.get_space(slug)
+ if not space:
+ return JSONResponse({"error": "space not found"}, status_code=404)
+ if not await _can_manage(space, _agent(user)):
+ raise HTTPException(403, "not authorized to manage this space's access rules")
+ await sp.remove_access_rule(space["space_id"], rule_id)
+ return {"slug": slug, "rules": await sp.list_access_rules(space["space_id"])}
+
+
+@router.get("/admin/capabilities",
+ dependencies=[Depends(require_scopes(["admin"]))],
+ summary="List a user's effective capabilities (admin only)")
+async def get_capabilities(
+ user: Annotated[LoginUserIn, Depends(get_current_user)],
+ member: Annotated[str, Query(..., description="User email to inspect")],
+):
+ if not await rbac.is_admin(_agent(user)):
+ raise HTTPException(403, "Admin/SuperAdmin role required")
+ return {
+ "member": member,
+ "roles": sorted(await rbac.active_roles(member)),
+ "capabilities": sorted(await rbac.capabilities(member)),
+ "grants": await rbac.list_grants(member),
+ }
+
+
+@router.post("/admin/capabilities/grant",
+ dependencies=[Depends(require_scopes(["admin"]))],
+ summary="Grant a capability to a user (admin only)",
+ description="Delegated upgrade: e.g. grant 'create_team_space' or "
+ "'manage_team_space' to a Curator/Lab Member so they can create "
+ "and manage team spaces. Requires the caller to hold an "
+ "Admin/SuperAdmin role.")
+async def grant_capability(body: GrantIn, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ if not await rbac.is_admin(_agent(user)):
+ raise HTTPException(403, "Admin/SuperAdmin role required to grant capabilities")
+ if body.capability not in rbac.GRANTABLE_CAPS:
+ raise HTTPException(400, f"capability is not delegatable; valid: {sorted(rbac.GRANTABLE_CAPS)}")
+ await rbac.grant_capability(body.member, body.capability, _agent(user))
+ return {"status": "granted", "member": body.member, "capability": body.capability}
+
+
+@router.post("/admin/capabilities/revoke",
+ dependencies=[Depends(require_scopes(["admin"]))],
+ summary="Revoke a granted capability (admin only)")
+async def revoke_capability(body: GrantIn, user: Annotated[LoginUserIn, Depends(get_current_user)]):
+ if not await rbac.is_admin(_agent(user)):
+ raise HTTPException(403, "Admin/SuperAdmin role required to revoke capabilities")
+ await rbac.revoke_capability(body.member, body.capability)
+ return {"status": "revoked", "member": body.member, "capability": body.capability}
diff --git a/query_service/core/search.py b/query_service/core/search.py
new file mode 100644
index 0000000..adb4eb1
--- /dev/null
+++ b/query_service/core/search.py
@@ -0,0 +1,195 @@
+# -*- coding: utf-8 -*-
+# -----------------------------------------------------------------------------
+# DISCLAIMER: This software is provided "as is" without any warranty,
+# express or implied, including but not limited to the warranties of
+# merchantability, fitness for a particular purpose, and non-infringement.
+# -----------------------------------------------------------------------------
+
+# @Author : Tek Raj Chhetri
+# @Email : tekraj@mit.edu
+# @File : search.py
+
+"""
+Hybrid search: Postgres is a full-text **locator** index, Oxigraph is the source
+of truth for the data.
+
+At ingest time each subject's literal text is indexed into `graph_search_index`
+along with its named graph and owning space (workspace). A search query runs in
+Postgres (fast, access-filtered by space visibility/membership) to LOCATE the
+matching subjects, then the actual triples for that page of hits are fetched from
+Oxigraph.
+
+Access model (identical to spaces):
+ * anonymous -> public-space subjects only
+ * logged-in -> public + the caller's member spaces (+ legacy/unmapped, authed)
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Optional
+from urllib.parse import quote
+
+import httpx
+
+from core.database import get_db_connection
+from core.shared import get_oxigraph_auth
+from core.graph_database_connection_manager import _get_endpoint
+from core.provenance import query_provenance_jsonld
+from core.spaces import get_space_for_graph
+
+logger = logging.getLogger(__name__)
+
+
+async def _sparql_select(query: str) -> List[Dict[str, Any]]:
+ """Run a SPARQL SELECT and return its bindings (empty list on error)."""
+ try:
+ endpoint = _get_endpoint("get")
+ auth = get_oxigraph_auth()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(120.0, connect=15.0)) as client:
+ resp = await client.post(
+ endpoint,
+ data={"query": query},
+ headers={"Accept": "application/sparql-results+json"},
+ auth=auth,
+ )
+ if resp.status_code != 200:
+ logger.warning("[search] SELECT failed (HTTP %s): %s", resp.status_code, (resp.text or "")[:400])
+ return []
+ return resp.json().get("results", {}).get("bindings", [])
+ except Exception as e:
+ logger.warning(f"[search] SELECT error: {e}", exc_info=True)
+ return []
+
+
+async def index_graph_subjects(named_graph_iri: str, delta_graph: Optional[str] = None) -> int:
+ """
+ (Re)index the subjects of a named graph into the Postgres locator index.
+
+ Each subject's searchable text is the concatenation of its literal objects,
+ read from Oxigraph. When `delta_graph` is given, only subjects touched by that
+ job are (re)indexed (their full text is still read from the target graph).
+ Best-effort — returns the number of subjects indexed (0 on failure).
+ """
+ try:
+ if delta_graph:
+ q = f"""
+ SELECT ?s (GROUP_CONCAT(DISTINCT STR(?o); SEPARATOR=" ") AS ?text)
+ WHERE {{
+ GRAPH <{delta_graph}> {{ ?s ?dp ?do }}
+ GRAPH <{named_graph_iri}> {{ ?s ?p ?o FILTER(isLiteral(?o)) }}
+ }} GROUP BY ?s
+ """
+ else:
+ q = f"""
+ SELECT ?s (GROUP_CONCAT(DISTINCT STR(?o); SEPARATOR=" ") AS ?text)
+ WHERE {{ GRAPH <{named_graph_iri}> {{ ?s ?p ?o FILTER(isLiteral(?o)) }} }}
+ GROUP BY ?s
+ """
+ rows = await _sparql_select(q)
+ if not rows:
+ return 0
+
+ space = await get_space_for_graph(named_graph_iri)
+ space_id = space["space_id"] if space else None
+
+ import time
+ now = time.time()
+ records = []
+ for b in rows:
+ subj = b.get("s", {}).get("value")
+ text = b.get("text", {}).get("value", "")
+ if subj and text.strip():
+ records.append((named_graph_iri, space_id, subj, text, now))
+ if not records:
+ return 0
+
+ async with get_db_connection() as conn:
+ await conn.executemany(
+ """
+ INSERT INTO graph_search_index (named_graph_iri, space_id, subject, text, tsv, updated_at)
+ VALUES ($1, $2, $3, $4, to_tsvector('english', $4), $5)
+ ON CONFLICT (named_graph_iri, subject) DO UPDATE SET
+ space_id = EXCLUDED.space_id,
+ text = EXCLUDED.text,
+ tsv = EXCLUDED.tsv,
+ updated_at = EXCLUDED.updated_at
+ """,
+ records,
+ )
+ logger.info(f"[search] Indexed {len(records)} subject(s) for {named_graph_iri}")
+ return len(records)
+ except Exception as e:
+ logger.warning(f"[search] Failed to index {named_graph_iri}: {e}", exc_info=True)
+ return 0
+
+
+async def reindex_graph_space(named_graph_iri: str, space_id: str) -> None:
+ """Point a graph's existing index rows at a (new) space — e.g. when a graph is
+ attached to a space after it was already indexed."""
+ try:
+ async with get_db_connection() as conn:
+ await conn.execute(
+ "UPDATE graph_search_index SET space_id = $1 WHERE named_graph_iri = $2",
+ space_id, named_graph_iri,
+ )
+ except Exception as e:
+ logger.warning(f"[search] Failed to reassign space for {named_graph_iri}: {e}", exc_info=True)
+
+
+async def search(q: str, caller: Optional[str], space_slug: Optional[str] = None,
+ limit: int = 25, offset: int = 0) -> Dict[str, Any]:
+ """
+ Locate matching subjects in Postgres (access-filtered), then fetch their triples
+ from Oxigraph. `caller` is the user email, or None for anonymous.
+ """
+ # Build the access-filtered locator query. Access is authoritative from the
+ # spaces tables (joined live), so visibility flips need no reindex.
+ params: List[Any] = [q, caller]
+ where = [
+ "i.tsv @@ plainto_tsquery('english', $1)",
+ "(s.visibility = 'public' OR m.member IS NOT NULL OR (i.space_id IS NULL AND $2 IS NOT NULL))",
+ ]
+ idx = 3
+ if space_slug:
+ where.append(f"s.slug = ${idx}")
+ params.append(space_slug)
+ idx += 1
+ limit_i, offset_i = idx, idx + 1
+ params.extend([limit, offset])
+
+ sql = f"""
+ SELECT i.named_graph_iri, i.subject, i.text,
+ s.slug AS space_slug, s.visibility AS visibility,
+ ts_rank(i.tsv, plainto_tsquery('english', $1)) AS rank
+ FROM graph_search_index i
+ LEFT JOIN spaces s ON s.space_id = i.space_id
+ LEFT JOIN space_members m ON m.space_id = i.space_id AND m.member = $2
+ WHERE {' AND '.join(where)}
+ ORDER BY rank DESC, i.subject
+ LIMIT ${limit_i} OFFSET ${offset_i}
+ """
+ async with get_db_connection() as conn:
+ rows = await conn.fetch(sql, *params)
+
+ hits = []
+ for r in rows:
+ text = r["text"] or ""
+ hits.append({
+ "subject": r["subject"],
+ "named_graph_iri": r["named_graph_iri"],
+ "space": r["space_slug"],
+ "visibility": r["visibility"] or ("legacy" if r["named_graph_iri"] else None),
+ "snippet": (text[:240] + "…") if len(text) > 240 else text,
+ })
+
+ # Fetch the located subjects' triples from Oxigraph (the source of truth).
+ data = None
+ if hits:
+ construct = "CONSTRUCT { ?s ?p ?o } WHERE { " + " UNION ".join(
+ f'{{ BIND(<{h["subject"]}> AS ?s) GRAPH <{h["named_graph_iri"]}> {{ ?s ?p ?o }} }}'
+ for h in hits
+ ) + " }"
+ data = await query_provenance_jsonld(construct)
+
+ return {"query": q, "space": space_slug, "count": len(hits), "hits": hits, "data": data}
diff --git a/query_service/core/security.py b/query_service/core/security.py
index d6b7348..4e78f68 100644
--- a/query_service/core/security.py
+++ b/query_service/core/security.py
@@ -21,7 +21,7 @@
import asyncio
from typing import Annotated, List, Optional, Dict
-from fastapi import Depends, HTTPException, status, WebSocket
+from fastapi import Depends, HTTPException, status, WebSocket, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from fastapi.security import OAuth2PasswordBearer
from jose import ExpiredSignatureError, JWTError, jwt
@@ -107,6 +107,30 @@ async def get_current_user(
return user
+async def get_current_user_optional(request: Request):
+ """
+ Return the authenticated user if a valid Bearer token is present, else None.
+
+ Unlike get_current_user this NEVER raises on a missing/invalid token — it is
+ for endpoints that serve public resources anonymously but still want to know
+ the caller's identity when a token is supplied (e.g. public-space reads).
+ """
+ auth = request.headers.get("authorization", "")
+ if not auth[:7].lower() == "bearer ":
+ return None
+ token = auth[7:].strip()
+ if not token:
+ return None
+ try:
+ payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+ email = payload.get("sub")
+ if not email:
+ return None
+ return await get_user(email=email)
+ except (ExpiredSignatureError, JWTError, Exception):
+ return None
+
+
def verify_scopes(required_scopes: List[str], token: str) -> bool:
decoded_token = decode_jwt(token)
token_scopes = decoded_token.get("scopes", [])
diff --git a/query_service/core/shared.py b/query_service/core/shared.py
index 2e8f8ac..200cf66 100644
--- a/query_service/core/shared.py
+++ b/query_service/core/shared.py
@@ -353,7 +353,7 @@ def chunk_ttl_to_named_graphs(ttl_str: str, named_graph_uri: str = "https://brai
return chunks
-def named_graph_metadata(named_graph_url, description):
+def named_graph_metadata(named_graph_url, description, agent_uri=None):
"""
Generates metadata for a named graph using the PROV and DCTERMS ontologies.
@@ -386,6 +386,13 @@ def named_graph_metadata(named_graph_url, description):
g.add((prov_entity, RDF.type, PROV.Entity))
g.add((prov_entity,PROV.generatedAtTime, Literal(created_At, datatype=XSD.dateTime)))
g.add((prov_entity,DCTERMS.description, Literal(description, datatype=XSD.string)))
+ # Record who registered the graph directly on the registry entry (PROV-O).
+ # This keeps registration provenance in one place (the registry graph) rather
+ # than duplicating it as a separate activity in the provenance graph.
+ if agent_uri:
+ agent = URIRef(agent_uri)
+ g.add((agent, RDF.type, PROV.Agent))
+ g.add((prov_entity, PROV.wasAttributedTo, agent))
named_graph_metadata = convert_ttl_to_named_graph(
ttl_str=g.serialize(format='turtle'),
named_graph_uri="https://brainkb.org/metadata/named-graph"
diff --git a/query_service/core/spaces.py b/query_service/core/spaces.py
new file mode 100644
index 0000000..1daf23c
--- /dev/null
+++ b/query_service/core/spaces.py
@@ -0,0 +1,433 @@
+# -*- coding: utf-8 -*-
+# -----------------------------------------------------------------------------
+# DISCLAIMER: This software is provided "as is" without any warranty,
+# express or implied, including but not limited to the warranties of
+# merchantability, fitness for a particular purpose, and non-infringement.
+# -----------------------------------------------------------------------------
+
+# @Author : Tek Raj Chhetri
+# @Email : tekraj@mit.edu
+# @File : spaces.py
+
+"""
+Spaces: owner-controlled containers of named graphs with private/public
+visibility and team membership.
+
+A space is a sovereign, IRI-addressable container (think Solid-style pod) that a
+user or team owns. Named graphs belong to a space; ingestion into a graph is
+allowed only for the owning space's owner/editors, while reads are allowed to
+members always and to ANYONE (even anonymous) when the space is public.
+
+Storage is hybrid (see SPACES_MODEL.md):
+ * Postgres (spaces / space_members / space_graphs) is the enforcement source of
+ truth — fast per-request authorization.
+ * A best-effort RDF mirror in the spaces metadata graph makes space manifests
+ portable/decentralizable and queryable via SPARQL.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+import uuid
+from typing import Any, Dict, List, Optional, Tuple
+from urllib.parse import quote
+
+import httpx
+from rdflib import Graph, Literal, Namespace, URIRef, RDF
+from rdflib.namespace import DCTERMS
+
+from core.database import get_db_connection
+from core.shared import get_oxigraph_auth
+from core.graph_database_connection_manager import _get_endpoint
+from core.provenance import agent_ref, BRAINKB, PROV
+
+logger = logging.getLogger(__name__)
+
+SPACES_METADATA_GRAPH = "https://brainkb.org/metadata/spaces/"
+SPACE_BASE = "https://brainkb.org/space/"
+SCHEMA = Namespace("https://schema.org/")
+
+ROLES = ("owner", "editor", "viewer")
+READ_ROLES = ("owner", "editor", "viewer")
+WRITE_ROLES = ("owner", "editor")
+
+
+def space_iri(slug: str) -> str:
+ return f"{SPACE_BASE}{quote(str(slug), safe='')}"
+
+
+# ---------------------------------------------------------------------------
+# Postgres CRUD
+# ---------------------------------------------------------------------------
+
+async def create_space(slug: str, name: str, description: Optional[str], owner: str,
+ visibility: str = "private", space_type: str = "individual") -> Dict[str, Any]:
+ """Create a space and register the owner as a member with role 'owner'."""
+ if visibility not in ("private", "public"):
+ raise ValueError("visibility must be 'private' or 'public'")
+ if space_type not in ("individual", "team"):
+ raise ValueError("space_type must be 'individual' or 'team'")
+ space_id = uuid.uuid4().hex
+ now = time.time()
+ async with get_db_connection() as conn:
+ async with conn.transaction():
+ await conn.execute(
+ """
+ INSERT INTO spaces (space_id, slug, name, description, owner, visibility, space_type, created_at, updated_at)
+ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $8)
+ """,
+ space_id, slug, name, description, owner, visibility, space_type, now,
+ )
+ await conn.execute(
+ """
+ INSERT INTO space_members (space_id, member, role, added_at)
+ VALUES ($1, $2, 'owner', $3)
+ """,
+ space_id, owner, now,
+ )
+ return await get_space(slug)
+
+
+async def get_space(slug: str) -> Optional[Dict[str, Any]]:
+ async with get_db_connection() as conn:
+ row = await conn.fetchrow("SELECT * FROM spaces WHERE slug = $1", slug)
+ if not row:
+ return None
+ members = await conn.fetch(
+ "SELECT member, role FROM space_members WHERE space_id = $1 ORDER BY role, member",
+ row["space_id"],
+ )
+ graphs = await conn.fetch(
+ "SELECT named_graph_iri FROM space_graphs WHERE space_id = $1 ORDER BY named_graph_iri",
+ row["space_id"],
+ )
+ return {
+ "space_id": row["space_id"],
+ "slug": row["slug"],
+ "name": row["name"],
+ "description": row["description"],
+ "owner": row["owner"],
+ "visibility": row["visibility"],
+ "space_type": row.get("space_type", "individual"),
+ "iri": space_iri(row["slug"]),
+ "members": [{"member": m["member"], "role": m["role"]} for m in members],
+ "graphs": [g["named_graph_iri"] for g in graphs],
+ "created_at": row["created_at"],
+ "updated_at": row["updated_at"],
+ }
+
+
+async def get_space_for_graph(named_graph_iri: str) -> Optional[Dict[str, Any]]:
+ """Return the space that owns a named graph, or None if the graph is unmapped."""
+ async with get_db_connection() as conn:
+ row = await conn.fetchrow(
+ """
+ SELECT s.* FROM spaces s
+ JOIN space_graphs g ON g.space_id = s.space_id
+ WHERE g.named_graph_iri = $1
+ """,
+ named_graph_iri,
+ )
+ if not row:
+ return None
+ return {
+ "space_id": row["space_id"], "slug": row["slug"], "name": row["name"],
+ "owner": row["owner"], "visibility": row["visibility"],
+ "space_type": row.get("space_type", "individual"),
+ }
+
+
+async def hidden_graphs_for(member: Optional[str]) -> set:
+ """
+ Return the set of named-graph IRIs the caller must NOT see in listings: graphs
+ belonging to a PRIVATE space the caller is not a member of. Public-space graphs
+ and legacy (unmapped) graphs are never hidden. Anonymous callers (member=None)
+ have every private-space graph hidden.
+ """
+ async with get_db_connection() as conn:
+ rows = await conn.fetch(
+ """
+ SELECT g.named_graph_iri
+ FROM space_graphs g
+ JOIN spaces s ON s.space_id = g.space_id
+ WHERE s.visibility = 'private'
+ AND NOT EXISTS (
+ SELECT 1 FROM space_members m
+ WHERE m.space_id = s.space_id AND m.member = $1
+ )
+ """,
+ member,
+ )
+ return {r["named_graph_iri"] for r in rows}
+
+
+async def member_role(space_id: str, member: Optional[str]) -> Optional[str]:
+ if not member:
+ return None
+ async with get_db_connection() as conn:
+ return await conn.fetchval(
+ "SELECT role FROM space_members WHERE space_id = $1 AND member = $2",
+ space_id, member,
+ )
+
+
+async def list_visible_spaces(member: Optional[str]) -> List[Dict[str, Any]]:
+ """Spaces the caller may see: all public spaces plus any they are a member of.
+ Anonymous callers (member=None) see only public spaces."""
+ async with get_db_connection() as conn:
+ if member:
+ rows = await conn.fetch(
+ """
+ SELECT DISTINCT s.slug, s.name, s.description, s.owner, s.visibility, s.created_at
+ FROM spaces s
+ LEFT JOIN space_members m ON m.space_id = s.space_id AND m.member = $1
+ WHERE s.visibility = 'public' OR m.member IS NOT NULL
+ ORDER BY s.created_at DESC
+ """,
+ member,
+ )
+ else:
+ rows = await conn.fetch(
+ """
+ SELECT s.slug, s.name, s.description, s.owner, s.visibility, s.created_at
+ FROM spaces s WHERE s.visibility = 'public'
+ ORDER BY s.created_at DESC
+ """,
+ )
+ return [
+ {"slug": r["slug"], "name": r["name"], "description": r["description"],
+ "owner": r["owner"], "visibility": r["visibility"], "iri": space_iri(r["slug"]),
+ "created_at": r["created_at"]}
+ for r in rows
+ ]
+
+
+async def add_member(space_id: str, member: str, role: str) -> None:
+ if role not in ROLES:
+ raise ValueError(f"role must be one of {ROLES}")
+ async with get_db_connection() as conn:
+ await conn.execute(
+ """
+ INSERT INTO space_members (space_id, member, role, added_at)
+ VALUES ($1, $2, $3, $4)
+ ON CONFLICT (space_id, member) DO UPDATE SET role = EXCLUDED.role
+ """,
+ space_id, member, role, time.time(),
+ )
+
+
+async def remove_member(space_id: str, member: str) -> None:
+ async with get_db_connection() as conn:
+ await conn.execute(
+ "DELETE FROM space_members WHERE space_id = $1 AND member = $2 AND role <> 'owner'",
+ space_id, member,
+ )
+
+
+async def set_visibility(slug: str, visibility: str) -> None:
+ if visibility not in ("private", "public"):
+ raise ValueError("visibility must be 'private' or 'public'")
+ async with get_db_connection() as conn:
+ await conn.execute(
+ "UPDATE spaces SET visibility = $1, updated_at = $2 WHERE slug = $3",
+ visibility, time.time(), slug,
+ )
+
+
+async def attach_graph(space_id: str, named_graph_iri: str) -> None:
+ async with get_db_connection() as conn:
+ await conn.execute(
+ """
+ INSERT INTO space_graphs (space_id, named_graph_iri, added_at)
+ VALUES ($1, $2, $3)
+ ON CONFLICT (named_graph_iri) DO NOTHING
+ """,
+ space_id, named_graph_iri, time.time(),
+ )
+ # Point any already-indexed rows for this graph at the space so search
+ # access-filtering picks up the (new) workspace immediately. Done inline
+ # (not via core.search) to avoid an import cycle.
+ await conn.execute(
+ "UPDATE graph_search_index SET space_id = $1 WHERE named_graph_iri = $2",
+ space_id, named_graph_iri,
+ )
+
+
+# ---------------------------------------------------------------------------
+# Authorization (enforcement)
+# ---------------------------------------------------------------------------
+
+ACTIONS = ("read", "write", "manage")
+RULE_SUBJECTS = ("global_role", "member", "space_role")
+_SPACE_ROLE_RANK = {"viewer": 1, "editor": 2, "owner": 3}
+
+
+async def add_access_rule(space_id: str, action: str, subject_type: str,
+ subject_value: str, created_by: str) -> None:
+ if action not in ACTIONS:
+ raise ValueError(f"action must be one of {ACTIONS}")
+ if subject_type not in RULE_SUBJECTS:
+ raise ValueError(f"subject_type must be one of {RULE_SUBJECTS}")
+ if subject_type == "space_role" and subject_value not in _SPACE_ROLE_RANK:
+ raise ValueError("space_role subject_value must be viewer/editor/owner")
+ async with get_db_connection() as conn:
+ await conn.execute(
+ """
+ INSERT INTO space_access_rules (space_id, action, subject_type, subject_value, created_by, created_at)
+ VALUES ($1, $2, $3, $4, $5, $6)
+ ON CONFLICT (space_id, action, subject_type, subject_value) DO NOTHING
+ """,
+ space_id, action, subject_type, subject_value, created_by, time.time(),
+ )
+
+
+async def remove_access_rule(space_id: str, rule_id: int) -> None:
+ async with get_db_connection() as conn:
+ await conn.execute(
+ "DELETE FROM space_access_rules WHERE id = $1 AND space_id = $2", rule_id, space_id
+ )
+
+
+async def list_access_rules(space_id: str, action: Optional[str] = None) -> List[Dict[str, Any]]:
+ async with get_db_connection() as conn:
+ if action:
+ rows = await conn.fetch(
+ "SELECT id, action, subject_type, subject_value FROM space_access_rules WHERE space_id=$1 AND action=$2 ORDER BY id",
+ space_id, action)
+ else:
+ rows = await conn.fetch(
+ "SELECT id, action, subject_type, subject_value FROM space_access_rules WHERE space_id=$1 ORDER BY action, id",
+ space_id)
+ return [dict(r) for r in rows]
+
+
+async def matches_access_rule(space_id: str, action: str, email: Optional[str]) -> bool:
+ """True iff the caller matches at least one access rule for (space, action).
+ Pure rule match — no owner/admin bypass, no 'no rules' default."""
+ from core import rbac
+ rules = await list_access_rules(space_id, action)
+ if not rules:
+ return False
+ roles = await rbac.active_roles(email)
+ srole = await member_role(space_id, email)
+ srank = _SPACE_ROLE_RANK.get(srole or "", 0)
+ for r in rules:
+ st, sv = r["subject_type"], r["subject_value"]
+ if st == "global_role" and sv in roles:
+ return True
+ if st == "member" and email and sv == email:
+ return True
+ if st == "space_role" and srank >= _SPACE_ROLE_RANK.get(sv, 99):
+ return True
+ return False
+
+
+async def space_action_permitted(space: Dict[str, Any], action: str, email: Optional[str]) -> bool:
+ """
+ Fine-grained per-space check for read/write. Returns True if the caller may do
+ `action` in this space under the space's access rules.
+
+ - Owner and global Admin/SuperAdmin always pass (no lockout).
+ - No rules for the action -> True (the endpoint's normal capability / membership
+ / visibility gates still apply separately).
+ - Rules present -> caller must match at least one.
+ """
+ from core import rbac
+ space_id = space["space_id"]
+ if email and await member_role(space_id, email) == "owner":
+ return True
+ if await rbac.is_admin(email):
+ return True
+ if not await list_access_rules(space_id, action):
+ return True
+ return await matches_access_rule(space_id, action, email)
+
+
+async def authorize(named_graph_iri: str, member: Optional[str], need: str) -> Tuple[bool, str]:
+ """
+ Decide whether ``member`` (a user email, or None if anonymous) may read/write
+ the given named graph, based on the owning space's visibility and membership.
+
+ Backward-compat: a graph not mapped to any space is treated as legacy — access
+ falls through to whatever scope check the endpoint already applied (returns
+ allowed=True here). Only space-mapped graphs are governed by space ACL.
+ """
+ space = await get_space_for_graph(named_graph_iri)
+ if space is None:
+ return True, "legacy (unmapped) graph — governed by endpoint scope only"
+
+ if need == "read":
+ if space["visibility"] == "public":
+ return True, "public space"
+ role = await member_role(space["space_id"], member)
+ if role in READ_ROLES:
+ return True, f"member ({role})"
+ return False, "private space — membership required"
+
+ if need == "write":
+ role = await member_role(space["space_id"], member)
+ if role in WRITE_ROLES:
+ return True, f"member ({role})"
+ return False, "write requires owner/editor membership of the space"
+
+ return False, f"unknown access mode: {need}"
+
+
+# ---------------------------------------------------------------------------
+# RDF mirror (best-effort, for portability / decentralization)
+# ---------------------------------------------------------------------------
+
+def _space_manifest_graph(space: Dict[str, Any]) -> Graph:
+ g = Graph()
+ g.bind("brainkb", BRAINKB)
+ g.bind("prov", PROV)
+ g.bind("dcterms", DCTERMS)
+ g.bind("schema", SCHEMA)
+ s = URIRef(space_iri(space["slug"]))
+ g.add((s, RDF.type, BRAINKB.Space))
+ g.add((s, BRAINKB.slug, Literal(space["slug"])))
+ g.add((s, SCHEMA.name, Literal(space["name"])))
+ if space.get("description"):
+ g.add((s, DCTERMS.description, Literal(space["description"])))
+ g.add((s, BRAINKB.visibility, Literal(space["visibility"])))
+ g.add((s, BRAINKB.owner, agent_ref(space["owner"])))
+ for m in space.get("members", []):
+ pred = {"owner": BRAINKB.owner, "editor": BRAINKB.editor, "viewer": BRAINKB.viewer}[m["role"]]
+ g.add((s, pred, agent_ref(m["member"])))
+ for giri in space.get("graphs", []):
+ g.add((s, BRAINKB.containsGraph, URIRef(giri)))
+ return g
+
+
+async def mirror_space_to_rdf(space: Dict[str, Any]) -> bool:
+ """Upsert a space's manifest into the spaces metadata graph via SPARQL Update.
+ Best-effort — a mirror failure never fails the enforcing Postgres operation."""
+ try:
+ s = space_iri(space["slug"])
+ triples = _space_manifest_graph(space).serialize(format="nt")
+ update = (
+ f"DELETE {{ GRAPH <{SPACES_METADATA_GRAPH}> {{ <{s}> ?p ?o }} }} "
+ f"WHERE {{ GRAPH <{SPACES_METADATA_GRAPH}> {{ <{s}> ?p ?o }} }} ; "
+ f"INSERT DATA {{ GRAPH <{SPACES_METADATA_GRAPH}> {{ {triples} }} }}"
+ )
+ endpoint = _get_endpoint("post") # .../update
+ auth = get_oxigraph_auth()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, connect=15.0)) as client:
+ resp = await client.post(endpoint, data={"update": update}, auth=auth)
+ if resp.status_code in (200, 204):
+ return True
+ logger.warning("[spaces] RDF mirror failed (HTTP %s): %s", resp.status_code, (resp.text or "")[:400])
+ return False
+ except Exception as e:
+ logger.warning(f"[spaces] Error mirroring space to RDF: {e}", exc_info=True)
+ return False
+
+
+async def construct_space_graphs(space: Dict[str, Any]) -> str:
+ """SPARQL CONSTRUCT returning all triples across the space's named graphs."""
+ graphs = space.get("graphs", [])
+ if not graphs:
+ return "CONSTRUCT {} WHERE {}"
+ unions = " UNION ".join(f"{{ GRAPH <{g}> {{ ?s ?p ?o }} }}" for g in graphs)
+ return f"CONSTRUCT {{ ?s ?p ?o }} WHERE {{ {unions} }}"
diff --git a/readme.md b/readme.md
index 1ac77a9..481019a 100644
--- a/readme.md
+++ b/readme.md
@@ -47,7 +47,17 @@ Once started, services are accessible at:
- **API Token Manager (Django)**: `http://localhost:8000/`
- Once you register JWT user you need to activate it using token manager. You can also assign permission.
- **Query Service (FastAPI)**: `http://localhost:8010/`
- - Now supports ingestion than just querying.
+ - Supports querying **and** ingestion of the knowledge graphs.
+ - Native W3C PROV-O provenance in the graph database, with triple-level delta
+ tracking (per-job delta graphs + query/compare endpoints).
+ - **Spaces**: team-owned, private/public containers of named graphs — keep data
+ private to members or publish it publicly (anonymous read). Per-endpoint JWT
+ scopes (`read`/`write`/`admin`).
+ - **Search**: hybrid full-text search — Postgres locator index (aware of
+ workspace + visibility) finds subjects, data is fetched from Oxigraph. Results
+ are access-filtered (anonymous sees public only).
+ - See `query_service/README.md`, `query_service/PROVENANCE_MODEL.md`, and
+ `query_service/SPACES_MODEL.md` for details.
- **ML Service (FastAPI)**: `http://localhost:8007/`
- Integrates StructSense (multi-agent NER + structured-resource extraction).
- Hosts **SynthScholar** at `/api/synth-scholar/*` — PRISMA-guided literature
diff --git a/usermanagement_service/core/routers/admin.py b/usermanagement_service/core/routers/admin.py
index 9f81ddb..300d54e 100644
--- a/usermanagement_service/core/routers/admin.py
+++ b/usermanagement_service/core/routers/admin.py
@@ -17,10 +17,11 @@
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import select, func
+from pydantic import BaseModel
from core.database import (
user_db_manager, user_profile_repo, user_role_repo, user_activity_repo,
available_role_repo, permission_repo, role_permission_repo, page_access_repo,
- oauth_identity_repo,
+ oauth_identity_repo, jwt_user_repo,
)
from core.models.database_models import (
AvailableRole as AvailableRoleModel,
@@ -495,6 +496,35 @@ async def admin_delete_openrouter_key(_admin: Annotated[dict, Depends(require_ad
# which re-reads `is_banned` per request. To delete a user entirely use
# DELETE /api/admin/users/{profile_id}.
+class _EmailIn(BaseModel):
+ email: str
+
+
+@router.post("/users/activate")
+async def activate_user(body: _EmailIn, _admin: Annotated[dict, Depends(require_admin)]):
+ """Activate a user's account (set the JWT user `is_active=True`) by email.
+ Needed e.g. after a password self-registration, or to re-enable an account."""
+ async with user_db_manager.get_async_session() as session:
+ u = await jwt_user_repo.get_by_email_any_status(session, body.email)
+ if not u:
+ raise HTTPException(status_code=404, detail="user not found")
+ changed = await jwt_user_repo.activate_user(session, u.id)
+ await session.commit()
+ return {"email": body.email, "is_active": True, "changed": bool(changed)}
+
+
+@router.post("/users/deactivate")
+async def deactivate_user(body: _EmailIn, _admin: Annotated[dict, Depends(require_admin)]):
+ """Deactivate a user's account (set the JWT user `is_active=False`) by email."""
+ async with user_db_manager.get_async_session() as session:
+ u = await jwt_user_repo.get_by_email_any_status(session, body.email)
+ if not u:
+ raise HTTPException(status_code=404, detail="user not found")
+ changed = await jwt_user_repo.deactivate_user(session, u.id)
+ await session.commit()
+ return {"email": body.email, "is_active": False, "changed": bool(changed)}
+
+
@router.post("/users/{profile_id}/ban")
async def ban_user(
profile_id: int,