Use instance for Knowledge Provider ownership

README.md

@@ -518,7 +518,7 @@ Translation data is surfaced through the page model and review endpoints.
 - `GET /documents/{workflow_id}/qdrant`
 - `GET /documents/{workflow_id}/qdrant/chunks`
 - `POST /documents/{workflow_id}/reingest`
-- `POST /search` — only answers from documents whose validity period covers today; `valid_on` asks about another day and `include_expired` drops the filter. Optional `group` (e.g. `mh`) limits results to one group's documents; omitted means every group, and the response's `effective_config.group` echoes what was applied (see `docs/openapi-search.yaml`)
+- `POST /search` — only answers from documents whose validity period covers today; `valid_on` asks about another day and `include_expired` drops the filter. Optional `instance` (e.g. `mh`) limits results to one instance's documents; omitted means every instance, and the response's `effective_config.instance` echoes what was applied (see `docs/openapi-search.yaml`)
 - `GET /indexes/summary`
 - `GET /indexes/{index_name}/settings`
 - `GET /indexes/{index_name}/stats`

docs/ADR/0006-groups-and-catalogs-model.md

@@ -5,7 +5,7 @@ Context: the first Postgres schema modelled a tenant as a free-text `documents.i
 Decision:
 - A **group** is a provider: one `groups` row per Keycloak tenant node (`/states/MH`, `/global`). Role leaves are not mirrored; permissions still come from the token. `users.group_id` is a single nullable FK for now (first tenant in the token, sorted, with a warning when there are several); a join table can replace it if multi-group membership is needed.
 - A **catalog** is `(group, type)`: `catalogs.group_id` plus `catalogs.type`, unique together. `type` holds the network schema name. The kind-to-schema map lives in `pipeline/network_constants.py` and is a placeholder: `scheme` is `openagrinet:AgricultureResource`, `advisory` is `openagrinet:KnowledgeAdvisory`, `video` is `openagrinet:VideoResource`. Code branches through that map, never on raw `openagrinet:*` strings. Catalog Beckn information (`publish_directives`, `visible_networks`, `criteria`, `network`) sits in `catalogs.metadata`.
-- A **document** belongs to one catalog and is identified there by `code`: `documents.catalog_id` NOT NULL, `UNIQUE (catalog_id, code)`, with `name` and `aliases` on the document. The API fields `code`, `name` and `aliases` are those columns. `documents.instance`, `document_kind` and `is_demo` are gone; the API returns the derived `group` (the group name) and `kind` (from the type map). The `/catalog/v1/*` snapshot keys, the Qdrant payload keys and Temporal arguments keep their earlier names (`scheme_code`, `document_kind`, `instance`, ...) because other services and in-flight workflows depend on them.
+- A **document** belongs to one catalog and is identified there by `code`: `documents.catalog_id` NOT NULL, `UNIQUE (catalog_id, code)`, with `name` and `aliases` on the document. The API fields `code`, `name` and `aliases` are those columns. `documents.instance`, `document_kind` and `is_demo` are gone; the API returns the derived `instance` (the group name) and `kind` (from the type map). The `/catalog/v1/*` snapshot keys, the Qdrant payload keys and Temporal arguments keep their earlier names (`scheme_code`, `document_kind`, `instance`, ...) because other services and in-flight workflows depend on them.
 - Re-uploading an existing code updates the same document (new workflow, `source` replaced, pages and chunks cleared, `reindex_required` set) rather than adding a second one.
 - Catalogs are created on upload for the tenant and kind. The upload therefore takes the document type up front.
 - Bootstrap schemes (the code-seeded entries with no documents) are dropped. `/catalog/v1/*` is served from the `catalog_status` view, one row per document, and a scheme appears once a document exists.

docs/openapi-search.yaml

@@ -52,11 +52,11 @@ paths:
                   top_k: 10
                   hybrid_alpha: 0.6
                   rerank_mode: "none"
-              one_group:
-                summary: Only one group's documents (e.g. a state's DSS)
+              one_instance:
+                summary: Only one instance's documents (e.g. a state's DSS)
                 value:
                   query: "cotton pest advisory"
-                  group: "mh"
+                  instance: "mh"
               one_provider:
                 summary: Only the chunks one network provider publishes
                 value:
@@ -70,7 +70,7 @@ paths:
               schema:
                 $ref: '#/components/schemas/SearchResponse'
         '400':
-          description: Missing/empty query, malformed valid_on, non-string group, non-string provider_id, or the underlying vector-store search failed
+          description: Missing/empty query, malformed valid_on, non-string instance, non-string provider_id, or the underlying vector-store search failed
           content:
             application/json:
               schema:
@@ -263,21 +263,21 @@ components:
             what search would have returned, or will return, on that date. A
             malformed value returns 400. Ignored when include_expired is true.
           default: "(today)"
-        group:
+        instance:
           type: string
           nullable: true
           description: |
-            Limit results to chunks tagged with this group code (for example
+            Limit results to chunks tagged with this instance code (for example
             `mh` for Maharashtra, `bv` for Bharat Vistaar). Case-insensitive
             and trimmed: `MH` and ` mh ` both mean `mh`. One value only —
-            a caller belongs to one group; fallback to another group is the
-            caller's second request. Omitted, null or empty means no group
-            filter: chunks from every group are searched as one pool. An
-            unknown group is not an error:
+            a caller belongs to one instance; fallback to another instance is the
+            caller's second request. Omitted, null or empty means no instance
+            filter: chunks from every instance are searched as one pool. An
+            unknown instance is not an error:
             the response is a success with zero hits. A non-string value
             returns 400. Applies in every search mode. A filter, not an
             access check: this endpoint applies no tenant scoping, so any
-            search-capable caller may ask for any group.
+            search-capable caller may ask for any instance.
           example: "mh"
         provider_id:
           type: string
@@ -289,7 +289,7 @@ components:
             empty means no provider filter; an unknown id returns zero hits;
             a non-string value returns 400. Only advisory and scheme chunks
             carry a provider id, so a provider filter leaves other kinds out.
-            Combines with `group` and validity. Any other type returns 400.
+            Combines with `instance` and validity. Any other type returns 400.
             A filter, not an access check.
           example: "mh.provider.oan.dev"
 
@@ -424,10 +424,10 @@ components:
           type: boolean
           description: True if the query-expansion profile actually rewrote the query.
         rerank_mode: { type: string }
-        group:
+        instance:
           type: string
           nullable: true
-          description: The normalised group filter that was applied (e.g. `mh`), or null when the search spanned every group.
+          description: The normalised instance filter that was applied (e.g. `mh`), or null when the search spanned every instance.
         provider_id:
           type: string
           nullable: true

pipeline/api/api.py

@@ -418,13 +418,15 @@ def _i(key: str, default: int = 0) -> int:
             "ready_for_ingestion": _i("ready_for_ingestion_documents"),
             "failed": _i("failed_documents"),
         },
-        "by_group": by_group or [],
+        "by_instance": [
+            {**row, "instance": row["group"]} for row in (by_group or [])
+        ],
     }
 
 
-def _resolve_create_group(user: AuthUser, requested: Optional[str] = None) -> str:
-    """Pick the document's group from Keycloak-allowed states / portal."""
-    return resolve_create_group(user, requested)
+def _resolve_upload_instance(user: AuthUser, instance: str) -> str:
+    """Resolve the upload owner using the existing Keycloak tenancy rules."""
+    return resolve_create_group(user, instance)
 
 
 def _document_title(filename: str) -> str:
@@ -558,7 +560,7 @@ def _document_summary_from_row(doc: dict, current_job: Optional[dict] = None) ->
         source_manifest_name=doc.get("source_manifest_name"),
         source_file_fingerprint=doc.get("source_file_fingerprint"),
         authoritative=bool(doc.get("source_manifest_name")),
-        group=normalize_group(doc.get("group")),
+        instance=normalize_group(doc.get("group")),
         stage=DocumentStage(doc["stage"]),
         page_count=doc.get("page_count") or 0,
         chunk_count=doc.get("chunk_count") or 0,
@@ -804,7 +806,7 @@ async def auth_me(user: CurrentUser):
         "state_roles": dict(user.state_roles or {}),
         "is_super_admin": bool(user.is_superadmin),
         # Portal group code for BV / platform docs (not a state).
-        "portal_group": PORTAL_GROUP,
+        "portal_instance": PORTAL_GROUP,
         "auth_disabled": user.token_disabled_mode,
     }
 
@@ -945,13 +947,17 @@ async def list_runs(
     Restricted users only see jobs whose document ``group`` is in their
     allowed states. Super-admin / bypass users see all tenants.
     """
-    return db.list_runs(
+    runs = db.list_runs(
         limit=limit,
         offset=offset,
         status=status,
         groups=_group_scope_for_user(user),
     )
 
+    for run in runs:
+        run["instance"] = run.pop("group")
+    return runs
+
 
 @app.get("/runs/{job_id}")
 async def get_run(job_id: str, user: RequireSearch):
@@ -1010,7 +1016,7 @@ def _build_document_detail(doc: dict) -> DocumentDetail:
         source_manifest_name=doc.get("source_manifest_name"),
         source_file_fingerprint=doc.get("source_file_fingerprint"),
         authoritative=bool(doc.get("source_manifest_name")),
-        group=normalize_group(doc.get("group")),
+        instance=normalize_group(doc.get("group")),
         filepath=doc["filepath"],
         stage=DocumentStage(doc["stage"]),
         page_count=doc.get("page_count", 0),

pipeline/api/documents.py

@@ -36,7 +36,7 @@ async def start_document_workflow(
     min_tokens: int = 100,
     index_name: str = "documents-index",
     stop_after_ocr: bool = False,
-    group: str = "",
+    instance: str = "",
     kind: str = Query(..., description="scheme or advisory"),
     code: str = "",
 ):
@@ -54,7 +54,7 @@ async def start_document_workflow(
     Rate limited to 10 requests/minute per IP.
     Requires permission: upload (no-op while AUTH_DISABLED=true).
     """
-    create_group = _api._resolve_create_group(user, group)
+    create_group = _api._resolve_upload_instance(user, instance)
     # Validate file path to prevent path traversal attacks
     filepath = _api.validate_file_path(data.filepath)
     source_filename = _api.get_filename_from_path(filepath)
@@ -85,7 +85,7 @@ async def start_document_workflow(
                     source_filename=source_filename,
                     source_file_fingerprint=source_file_fingerprint,
                     authoritative=bool(existing_doc.get("source_manifest_name")) if existing_doc else False,
-                    group=normalize_group(existing_doc.get("group")),
+                    instance=normalize_group(existing_doc.get("group")),
                     stage=DocumentStage(state.get("stage", "registered")),
                     page_count=state.get("page_count", 0),
                     chunk_count=state.get("chunk_count", 0),
@@ -182,7 +182,7 @@ async def start_document_workflow(
         new_value=source_filename,
         metadata={
             "source": "register_endpoint",
-            "group": create_group,
+            "instance": create_group,
             "kind": doc_kind,
             "code": doc_code,
             "replaced_existing": bool(replace_doc),
@@ -198,7 +198,7 @@ async def start_document_workflow(
         source_filename=source_filename,
         source_file_fingerprint=source_file_fingerprint,
         authoritative=False,
-        group=create_group,
+        instance=create_group,
         stage=DocumentStage.REGISTERED,
         page_count=0,
         chunk_count=0,
@@ -220,11 +220,11 @@ async def start_batch_workflows(
     chunk_overlap: int = 128,
     min_tokens: int = 100,
     stop_after_ocr: bool = False,
-    group: str = "",
+    instance: str = "",
     kind: str = Query(..., description="scheme or advisory"),
 ):
     """Start workflows for all supported documents in a directory."""
-    create_group = _api._resolve_create_group(user, group)
+    create_group = _api._resolve_upload_instance(user, instance)
     directory = Path(data.directory)
     if not directory.exists():
         raise HTTPException(404, f"Directory not found: {data.directory}")
@@ -245,7 +245,7 @@ async def start_batch_workflows(
                 chunk_overlap=chunk_overlap,
                 min_tokens=min_tokens,
                 stop_after_ocr=stop_after_ocr,
-                group=create_group,
+                instance=create_group,
                 kind=kind,
                 code="",
             )

pipeline/api/search.py

@@ -170,10 +170,10 @@ def _rerank_hits(query: str, hits: list[dict], rerank_mode: str) -> list[dict]:
 
 
 
-def _requested_group(payload: dict) -> str | None:
-    raw = payload.get("group")
+def _requested_instance(payload: dict) -> str | None:
+    raw = payload.get("instance")
     if raw is not None and not isinstance(raw, str):
-        raise HTTPException(400, "group must be a string")
+        raise HTTPException(400, "instance must be a string")
     return (raw or "").strip().lower() or None
 
 
@@ -188,9 +188,9 @@ def _requested_provider_id(payload: dict) -> str | None:
 async def run_search(payload: dict, user: RequireSearch):
     """Search the vector index.
 
-    `group` is a filter, not an access check: this endpoint applies no tenant
+    `instance` is a filter, not an access check: this endpoint applies no tenant
     scoping to its results (unlike `GET /chunks/search`), so any
-    search-capable caller may ask for any group.
+    search-capable caller may ask for any instance.
     """
     from ..vector_store import get_default_index_name, get_vector_store
 
@@ -247,7 +247,7 @@ async def run_search(payload: dict, user: RequireSearch):
             valid_on = document_validity.parse_day(valid_on, "valid_on")
         except document_validity.DocumentValidityError as exc:
             raise HTTPException(400, str(exc)) from None
-    group = _requested_group(payload)
+    instance = _requested_instance(payload)
     provider_id = _requested_provider_id(payload)
     expanded_query = _expand_query(query, query_expansion_profile)
     # Qdrant embeddings apply E5 prefixes internally; don't pre-prefix here.
@@ -270,7 +270,7 @@ async def run_search(payload: dict, user: RequireSearch):
             ef_search=ef_search,
             apply_validity=not include_expired,
             valid_on=valid_on,
-            instance=group,
+            instance=instance,
             provider_id=provider_id,
         )
     except Exception as error:
@@ -320,7 +320,7 @@ async def run_search(payload: dict, user: RequireSearch):
             "rerank_mode": rerank_mode,
             "apply_validity": not include_expired,
             "valid_on": result.get("valid_on"),
-            "group": group,
+            "instance": instance,
             "provider_id": provider_id,
             "filter_string": None,
         },

pipeline/api/upload.py

@@ -31,7 +31,7 @@ async def upload_and_process(
     min_tokens: int = 100,
     index_name: str = "documents-index",
     stop_after_ocr: bool = False,
-    group: str = "",
+    instance: str = "",
     kind: str = Query(..., description="scheme or advisory"),
     code: str = "",
     display_name: Optional[str] = Form(None),
@@ -44,7 +44,7 @@ async def upload_and_process(
     Rate limited to 10 requests/minute per IP.
     Requires permission: upload (no-op while AUTH_DISABLED=true).
     """
-    create_group = _api._resolve_create_group(user, group)
+    create_group = _api._resolve_upload_instance(user, instance)
     doc_kind, doc_code, replace_doc = _api._resolve_document_identity(
         user, create_group, kind, code, file.filename or ""
     )
@@ -104,7 +104,7 @@ async def upload_and_process(
                     source_filename=file.filename,
                     source_file_fingerprint=file_hash,
                     authoritative=bool(existing_doc.get("source_manifest_name")) if existing_doc else False,
-                    group=normalize_group(existing_doc.get("group")),
+                    instance=normalize_group(existing_doc.get("group")),
                     stage=DocumentStage(state.get("stage", "registered")),
                     page_count=state.get("page_count", 0),
                     chunk_count=state.get("chunk_count", 0),
@@ -211,7 +211,7 @@ async def upload_and_process(
         new_value=file.filename,
         metadata={
             "source": "upload_endpoint",
-            "group": create_group,
+            "instance": create_group,
             "kind": doc_kind,
             "code": doc_code,
             "replaced_existing": bool(replace_doc),
@@ -237,7 +237,7 @@ async def upload_and_process(
         source_filename=file.filename,
         source_file_fingerprint=file_hash,
         authoritative=False,
-        group=create_group,
+        instance=create_group,
         stage=DocumentStage.REGISTERED,
         page_count=0,
         chunk_count=0,

pipeline/auth/keycloak_admin.py

@@ -506,7 +506,7 @@ def provision_user(
         "roles": [r.get("name") for r in (roles or []) if isinstance(r, dict)],
         "app_url": ui_url,
         "login_url": login_url,
-        "portal_group": PORTAL_GROUP,
+        "portal_instance": PORTAL_GROUP,
         "share_message": "\n".join(share_lines),
     }
 
@@ -517,7 +517,7 @@ def list_access_options() -> dict[str, Any]:
     return {
         "keycloak_admin_configured": cfg.configured,
         "realm": cfg.realm,
-        "portal_group": PORTAL_GROUP,
+        "portal_instance": PORTAL_GROUP,
         "access_types": [
             {
                 "id": "super_admin",

pipeline/db.py

@@ -2612,7 +2612,7 @@ def _audit_row_to_legacy(row: dict) -> dict:
         "actor_roles": ",".join(roles) if roles else None,
         "filename": row["filename"],
         "name": row["document_name"],
-        "group": row["doc_group"] or row["audit_group"],
+        "instance": row["doc_group"] or row["audit_group"],
         "uploaded_by_username": row["uploaded_by_username"],
         "uploaded_by_email": row["uploaded_by_email"],
     }

pipeline/models.py

@@ -218,7 +218,7 @@ class DocumentSummary(BaseModel):
     source_manifest_name: Optional[str] = None
     source_file_fingerprint: Optional[str] = None
     authoritative: bool = False
-    group: str = "default"  # tenant code (groups.name) of the document's catalog
+    instance: str = "default"  # tenant code of the document's catalog
     stage: DocumentStage
     page_count: int
     chunk_count: int
@@ -352,8 +352,8 @@ class DocumentGraph(BaseModel):
     runtime: dict[str, Any]
 
 
-class GroupDocumentCount(BaseModel):
-    group: str
+class InstanceDocumentCount(BaseModel):
+    instance: str
     count: int
     success: int = 0
     failed: int = 0
@@ -368,7 +368,7 @@ class DocumentCohortsResponse(BaseModel):
     review_queue: int
     failed_documents: int
     by_stage: dict[str, int]
-    by_group: list[GroupDocumentCount] = []
+    by_instance: list[InstanceDocumentCount] = []
     needs_reindex: int
     running_jobs: int = 0
 
@@ -445,7 +445,7 @@ class AuditLogEntry(BaseModel):
     # strips them from the response and the UI can only show a raw doc_id hash.
     filename: Optional[str] = None
     name: Optional[str] = None
-    group: Optional[str] = None
+    instance: Optional[str] = None
     uploaded_by_username: Optional[str] = None
     uploaded_by_email: Optional[str] = None