feat: publish group documents under the group's own network provider [#190]

.env.example

@@ -223,6 +223,11 @@ DISCOVERY_SERVICE_ENDPOINT=
 NETWORK_SENDER_ID=
 # Publicly reachable base URL of this provider, sent as each catalog's bppUri.
 NETWORK_SENDER_URI=
+NETWORK_PROVIDER_ID_TEMPLATE={kind}.{instance}.provider.oan.dev
+NETWORK_ADVISORY_PROVIDER_NAME=Advisories from documents
+NETWORK_SCHEMES_PROVIDER_NAME=Schemes from documents
+NETWORK_GROUP_PROVIDER_NAME_TEMPLATE={group_name} {provider_name}
+PROVIDER_CONFIG_OVERRIDE_PATH=
 
 
 # --- Keycloak container (only when running Keycloak inside docker-compose) ---

AGENTS.md

@@ -61,7 +61,8 @@ Core modules (all under `pipeline/`):
 - `db.py` — SQLite access layer; SQLite is the **canonical** metadata/review-state store (documents, pages, chunks, jobs, artifacts, audit log), not just a cache
 - `document_repository.py` — `DocumentRepository`, a domain layer over `db.py` for document reads; `db.py` stays one-function-per-query with no domain knowledge
 - `network_constants.py` — fixed values sent on the network (catalog/resource ids, topics, languages, Beckn version). The file v2 edits when the AI layer starts deriving them
-- `catalog_builder.py` — pure builder mapping a document's knowledge kind (`advisory`/`scheme`) to the single `OnDemand` Beckn catalog announced for that kind; returns `None` for any other kind. No env, no I/O
+- `catalog_builder.py` — pure builder mapping a document's knowledge kind (`advisory`/`scheme`) to the single `OnDemand` Beckn catalog announced for that kind; takes the resolved provider and prefixes catalog/resource ids for a Group; returns `None` for any other kind. No env, no I/O
+- `network_provider.py` — pure provider resolution for Publish to Network: the provider id (overrides, then `{kind}.{instance}` template) and name for a Knowledge Kind and Group. `network_group.py` builds its `ProviderNaming` from env and the overrides file; `catalog_builder.py` takes the resolved provider
 - `document_validity.py` — pure module owning Document Validity: the period a document's chunks answer searches in, the upload-day start with no end by default, and the clock injection point. Shared by `api.py` (stamping the default at upload), `scheme_catalog.py` (validating a reviewer's edit), `activities.py` (stamping chunk payloads) and `vector_store/qdrant_store.py` (filtering at search time). `db.py` stores the dates it is handed and defaults nothing
 - `discovery_publish_service.py` — `DiscoveryPublishService`: owns the Publish to Network env vars and the HTTP call. Deliberately Temporal-free
 - `models.py` — Pydantic models, including `DocumentStage` enum and `PIPELINE_STAGES` (the stepper-UI stage list)

CONTEXT.md

@@ -9,7 +9,7 @@ The operator action that approves a document out of `ready_for_ingestion` and le
 _Avoid_: "publish" alone — ambiguous with Publish to Network.
 
 **Publish to Network**:
-The pipeline stage that runs only after a document's PROD promotion succeeds, and POSTs a Beckn-shaped `publish` envelope to an external Discovery Service. Skipped entirely when PROD is disabled (`DISABLE_PROD_SETTING=true`) — a document never reaches the network without first promoting to PROD. The envelope it sends is kind-level (see Knowledge Kind), not document-level: it does not make any single document searchable, only re-announces that this provider serves that kind at all.
+The pipeline stage that runs only after a document's PROD promotion succeeds, and POSTs a Beckn-shaped `publish` envelope to an external Discovery Service. Skipped entirely when PROD is disabled (`DISABLE_PROD_SETTING=true`) — a document never reaches the network without first promoting to PROD. The envelope it sends is kind-level (see Knowledge Kind), not document-level: it does not make any single document searchable, only re-announces that the document's provider (its Group Provider, or the Shared Provider) serves that kind at all.
 _Avoid_: "publish" alone — ambiguous with Publish to Dev.
 
 **Master Catalog**:
@@ -24,6 +24,18 @@ _Avoid_: "catalog" alone — ambiguous with Master Catalog and the Network Catal
 The Beckn wire-protocol `message.catalogs` object sent by Publish to Network to the Discovery Service. Structurally and semantically unrelated to the Master Catalog and Scheme Catalog tables — do not conflate the three.
 _Avoid_: "catalog" alone.
 
+**Group**:
+The state a document belongs to, named by its state code (`mh` for Maharashtra) and taken from the document's `instance`. Bharat Vistaar and `default` documents have no Group. New features say Group; existing screens, roles and data still say state or instance.
+_Avoid_: "tenant"; "state" or "instance" in new features. Never call Bharat Vistaar a Group.
+
+**Group Provider**:
+The network provider a Group's documents are announced under in Publish to Network, one per Knowledge Kind, each with its own catalog. Every Group has one; a deployment may give a Group its own provider id instead of the default.
+_Avoid_: "state provider".
+
+**Shared Provider**:
+The network provider used by documents with no Group (Bharat Vistaar, `default`), one per Knowledge Kind, keeping the original catalog ids.
+_Avoid_: "national provider", "central provider".
+
 **Knowledge Kind**:
 What a document *is* to the network — `advisory`, `scheme`, `video`, or an operator-entered slug — held in `documents.document_kind`. Asserted by a reviewer during the pipeline, never inferred from the file, and absent until then (every document starts as the default `document`). Only `advisory` and `scheme` map to a Network Catalog Envelope; the rest publish nothing.
 _Avoid_: "document type" — collides with `source_type`/`canonical_input_type`, which describe the input *format* (pdf, spreadsheet). Also avoid saying an advisory is "uploaded": what is uploaded is a file, which only becomes an advisory when someone classifies it.

ENV.md

@@ -249,10 +249,17 @@ Separate from the SQLite/Qdrant catalog above (which tracks PROD vector publicat
 
 First piece of the network-discovery push architecture: once a document's PROD promotion succeeds, the `publishing_to_network` pipeline stage POSTs a Beckn-shaped `publish` envelope to an external Discovery Service. It's a PROD-only stage (see `docs/ADR/0004-publishing-to-network-is-prod-only.md`) — skipped entirely when `DISABLE_PROD_SETTING=true`, since a document never reaches the network without first promoting to PROD. This is going to be the only way the experience layer discovers schemas going forward — the Master Scheme Catalog pull API above stays in place for backward compatibility. The catalog announces the *knowledge kind* the document was classified as — one catalog with exactly one `OnDemand` resource for `advisory`, another for `scheme`; any other `document_kind` publishes nothing. `DISCOVERY_SERVICE_ENDPOINT` and `NETWORK_SENDER_ID` are **required** whenever PROD promotion is enabled — there is no separate feature flag for this stage beyond `DISABLE_PROD_SETTING`.
 
+Each catalog's provider is named per Knowledge Kind and instance: id and descriptor code `{kind}.{instance}.provider.oan.dev` (`advisory.bv.provider.oan.dev`, `scheme.mh.provider.oan.dev`), name `Advisories|Schemes from documents`. A document filed under a state (its **Group**, taken from the document's `instance`) uses the group code as the instance, prefixes catalog and resource ids with it (`mh-cat-oan-knowledge-provider-advisories`) and names catalog and provider `<Group name> <provider name>` (`Maharashtra Advisories from documents`). Documents under the portal instance (`PORTAL_INSTANCE`, `bv`) or the default instance (`DEFAULT_INSTANCE`) have no Group: they use the portal instance in the provider id and keep the unprefixed ids. Every Group publishes. An instance can replace the default provider id per Knowledge Kind in the provider id overrides file (`PROVIDER_CONFIG_OVERRIDE_PATH`); a kind it leaves out keeps the default. A missing or malformed file, or a template with an unknown placeholder, stops both the API and the worker from booting. The file is not mounted by `docker-compose.yml` — mount your own into the `api` and `worker` containers and point the variable at it; `deploy/network_group_providers.example.json` shows the shape. See `docs/ADR/0006-publish-to-network-under-per-group-provider.md`.
+
 | Variable | Default | Purpose |
 |----------|---------|---------|
 | `DISCOVERY_SERVICE_ENDPOINT` | *(empty — required)* | Base URL of the external Discovery Service; the pipeline POSTs to `{DISCOVERY_SERVICE_ENDPOINT}/publish` |
 | `NETWORK_SENDER_ID` | *(empty — required)* | This instance's Beckn `senderId`, also sent as `networkId` |
+| `PROVIDER_CONFIG_OVERRIDE_PATH` | unset (bundled `pipeline/network_group_providers.json`, `{}` — no overrides) | Path to a JSON object of instance code → `{kind: provider id}`, set per deployment, e.g. `{"ka": {"advisory": "advisory.ka.com", "scheme": "scheme.ka.com"}}`. Codes and kinds are case-insensitive; kinds are `advisory` or `scheme`. A missing file, invalid JSON, an unknown kind, or an empty or invalid id (same rule as the template) stop the API and worker from starting |
+| `NETWORK_PROVIDER_ID_TEMPLATE` | `{kind}.{instance}.provider.oan.dev` | Default provider id and descriptor code, unless overridden in the file above. `{kind}` is `advisory` or `scheme`; `{instance}` is the group code, or `PORTAL_INSTANCE` for documents with no Group. Must use both and render a valid provider id (lowercase letters, digits, `-` and `.`, no leading, trailing or repeated `.`), or the API and worker refuse to start. A document whose instance code renders an invalid id fails its publish without retry |
+| `NETWORK_ADVISORY_PROVIDER_NAME` | `Advisories from documents` | Provider name for advisory catalogs |
+| `NETWORK_SCHEMES_PROVIDER_NAME` | `Schemes from documents` | Provider name for scheme catalogs |
+| `NETWORK_GROUP_PROVIDER_NAME_TEMPLATE` | `{group_name} {provider_name}` | Provider and catalog name for a Group, e.g. `Maharashtra Advisories from documents`. Must use `{group_name}`, or the API and worker refuse to start |
 | `DISCOVERY_SERVICE_PUBLISH_INITIAL_INTERVAL_SECONDS` | `30` | Wait before the first retry of a failed publish |
 | `DISCOVERY_SERVICE_PUBLISH_BACKOFF_COEFFICIENT` | `2.0` | Multiplier applied to the wait after each retry |
 | `DISCOVERY_SERVICE_PUBLISH_MAX_INTERVAL_SECONDS` | `300` | Cap on the wait between retries |
@@ -270,6 +277,7 @@ First piece of the network-discovery push architecture: once a document's PROD p
 | OCR / translation / chunking / domain tags | — | config / status | ✅ runs jobs |
 | Qdrant (`VECTOR_DB_*`) | — | ✅ | ✅ |
 | Publish to Network (`DISCOVERY_SERVICE_ENDPOINT`, `NETWORK_SENDER_ID`) | — | — | ✅ runs the publish activity |
+| Network provider config (`PROVIDER_CONFIG_OVERRIDE_PATH`, `NETWORK_*_PROVIDER_*`) | — | validated at boot | ✅ validated at boot, used by the publish activity |
 
 ---
 

deploy/network_group_providers.example.json

@@ -0,0 +1,6 @@
+{
+  "ka": {
+    "advisory": "advisory.ka.com",
+    "scheme": "scheme.ka.com"
+  }
+}

docs/ADR/0006-publish-to-network-under-per-group-provider.md

@@ -0,0 +1,24 @@
+---
+status: accepted
+---
+
+# Publish to Network under a per-Group provider
+
+Context: ADR 0003 announced every document under one provider, with one catalog per Knowledge Kind. A state's experience (for example Maharashtra's, through DSS) needs to consume only its own state's knowledge from the network, and one shared provider gives it no way to do that (engineering-tracker#190).
+
+Decision:
+- Every catalog has a provider per Knowledge Kind and instance. By default its id and descriptor code are `{kind}.{instance}.provider.oan.dev` (`advisory.mh.provider.oan.dev`, `scheme.bv.provider.oan.dev`). It is named `Advisories|Schemes from documents`, prefixed with the Group name for a Group (`Maharashtra Advisories from documents`).
+- Documents with a Group use the group code as the instance. Catalog and resource ids are today's ids prefixed with the group code (`mh-cat-oan-knowledge-provider-advisories`, `mh-res-oan-knowledge-provider-advisories`, and the same for schemes), and the catalog takes the provider's name.
+- The group code is derived from the document's `instance`, not from the uploader's group, so a Super Admin uploading for a state publishes under that state.
+- Documents with no Group (Bharat Vistaar `bv`, `default`) use the portal instance `bv` in the provider id and keep ADR 0003's unprefixed catalog and resource ids.
+- Every Group publishes; only an unmapped Knowledge Kind skips. A real publish failure stays fatal, per ADR 0004.
+- The provider id template, the per-kind provider names and the Group name template are environment settings, so a deployment can change them without a code change. A template with an unknown placeholder, an id template missing `{kind}` or `{instance}`, a Group name template missing `{group_name}`, or a provider id (rendered or overridden) outside lowercase letters, digits, `-` and dot-separated parts stops the API and the worker from booting, since a published id can't be removed later. A document whose instance code still renders an invalid id fails its publish, without retry, before anything is sent.
+- An instance that wants its own provider id overrides it per Knowledge Kind in a JSON file whose path is `PROVIDER_CONFIG_OVERRIDE_PATH`, e.g. `{"ka": {"advisory": "advisory.ka.com", "scheme": "scheme.ka.com"}}`. Instances and kinds it leaves out keep the template id. It defaults to a bundled empty `pipeline/network_group_providers.json`, so onboarding a Group needs no config. A missing or malformed file stops both the API and the worker from booting, so a bad deploy can't quietly publish under the wrong id.
+- "No Group" follows the configured `PORTAL_INSTANCE` and `DEFAULT_INSTANCE` values rather than hardcoding `bv` and `default`.
+- The publish endpoint stays the same for every Group.
+
+Consequences:
+- Catalog and resource ids are permanent under `MERGE` (ADR 0003). State knowledge published before this change stays in the unprefixed catalogs; nothing moves it to the Group's `mh-` catalogs.
+- The next `bv` publish re-announces the unprefixed catalogs under `advisory.bv`/`scheme.bv` instead of `knowledge-provider`, keeping their catalog and resource ids. Changing the id template or an override later does the same. Whether the network moves such a catalog to the new provider or keeps the old entry depends on the adapter, and is untested.
+- Depends on the network adapter layer accepting the new provider ids, for `bv` as well as Groups, and the Groups' prefixed catalog and resource ids.
+- Only the Publish to Network payload changes. Chunks and the search indexes gain no `group` field; they keep `instance` and `instance_name`, where Bharat Vistaar and `default` documents carry `bv`/`default` and state documents carry the state code, so #202 can filter on `instance`.

pipeline/activities.py

@@ -30,6 +30,7 @@
 from .document_repository import DocumentRepository
 from .instances import instance_display_name
 from .network_constants import RESULT_REJECTED
+from .network_provider import InvalidProviderIdError
 from .ocr import ocr_pdf as run_ocr_pdf
 from .ocr import ocr_pdf_in_segments as run_ocr_pdf_in_segments
 from .translation import load_translation_config, translate_pages
@@ -1464,11 +1465,15 @@ async def publish_catalog_to_network(workflow_id: str, transaction_id: str) -> d
 
     repository = DocumentRepository()
     document_kind = normalize_document_kind(repository.get_document_kind(workflow_id))
+    group = repository.get_document_group(workflow_id)
 
     service = DiscoveryPublishService()
-    result = await asyncio.to_thread(
-        service.publish, transaction_id, document_kind, workflow_id
-    )
+    try:
+        result = await asyncio.to_thread(
+            service.publish, transaction_id, document_kind, workflow_id, group
+        )
+    except InvalidProviderIdError as exc:
+        raise ApplicationError(str(exc), non_retryable=True) from exc
 
     if result["skipped"]:
         artifact_type = "network_publish_skipped"

pipeline/api.py

@@ -103,6 +103,7 @@
     SearchSettingsUpdate,
     SettingsAuditResponse,
 )
+from .network_group import validate_network_provider_config
 from .workflows import (
     ChunkingOnlyWorkflow,
     DocumentPipelineWorkflow,
@@ -137,6 +138,8 @@ async def lifespan(app: FastAPI):
     """Initialize Temporal and MinIO clients on startup."""
     global temporal_client, minio_client, MINIO_BUCKET
 
+    validate_network_provider_config()
+
     auth_cfg = load_auth_config()
     validate_auth_config(auth_cfg)
     if auth_cfg.disabled:

pipeline/auth/tenancy.py

@@ -26,6 +26,17 @@ def default_instance() -> str:
     return (os.environ.get("DEFAULT_INSTANCE") or "default").strip().lower() or "default"
 
 
+def portal_instance() -> str:
+    return (os.environ.get("PORTAL_INSTANCE") or PORTAL_INSTANCE).strip().lower() or PORTAL_INSTANCE
+
+
+def group_for_instance(instance: str | None) -> str | None:
+    code = (instance or "").strip().lower()
+    if not code or code in {portal_instance(), default_instance()}:
+        return None
+    return code
+
+
 def normalize_instance(value: str | None) -> str:
     text = (value or "").strip().lower()
     return text or default_instance()
@@ -76,8 +87,7 @@ def resolve_create_instance(user: AuthUser, requested: str | None = None) -> str
         if requested_norm:
             return normalize_instance(requested_norm)
         # Prefer portal tag for platform operators when nothing chosen.
-        portal = (os.environ.get("PORTAL_INSTANCE") or PORTAL_INSTANCE).strip().lower()
-        return normalize_instance(portal or default_instance())
+        return normalize_instance(portal_instance())
 
     allowed = allowed_instances(user) or set()
     if not allowed: