---
title: 'Doc-Extraction Tenant/Node Isolation (GAP-018)'
slug: 'de-tenant-node-isolation'
created: '2026-03-10'
completed: '2026-03-13'
status: 'done'
stepsCompleted: [1, 2, 3, 4]
commit: 'e18d1e7'
tech_stack: ['Python 3.11', 'SQLAlchemy 2.x', 'Alembic', 'PostgreSQL 16', 'Milvus 2.x (pymilvus + LlamaIndex MilvusVectorStore)', 'Apache Airflow 3.x', 'FastAPI', 'Redis']
files_modified:
  - 'src/db/models.py'
  - 'src/api/tenant_context.py (NEW)'
  - 'src/api/routes/documents.py'
  - 'src/api/routes/query.py'
  - 'src/services/document_service.py'
  - 'src/services/seaweedfs_service.py'
  - 'src/knowledge/vector_store.py'
  - 'src/knowledge/domain_aware_retriever.py'
  - 'src/retrieval/semantic_retriever.py'
  - 'src/extraction_v2/domain_term_extractor.py'
  - 'src/core/config.py'
  - 'airflow/dags/document_extraction_dag.py'
  - 'airflow/dags/tasks/processing_tasks.py'
  - 'airflow/plugins/operators/validate_event.py'
  - 'airflow/plugins/operators/fetch_document.py'
  - 'scripts/upload_documents.py'
  - 'src/db/migrations/versions/20260313_100000_add_tenant_node_columns.py (NEW)'
  - 'src/db/migrations/versions/20260313_100001_backfill_tenant_node.py (NEW)'
  - 'src/db/migrations/versions/20260313_100002_enforce_tenant_node_not_null_rls.py (NEW)'
code_patterns:
  - 'SQLAlchemy DeclarativeBase with UUID PKs and mapped_column'
  - 'TenantIsolationMixin for tenant_id/node_id on all isolated models'
  - 'Alembic 3-phase migration (nullable -> backfill -> NOT NULL + RLS)'
  - 'Airflow @task.external_python with /opt/airflow/.venv'
  - 'pymilvus FieldSchema/CollectionSchema for raw Milvus access'
  - 'LlamaIndex MilvusVectorStore for store tasks (write path)'
  - 'FastAPI Depends() with TenantContextDep from X-Tenant-ID/X-Node-ID headers'
  - 'set_rls_tenant() for parameterized SET LOCAL before queries'
  - 'Async session factory via async_sessionmaker (src/db/session.py)'
  - 'Airflow tasks create their own AsyncEngine (not shared session factory)'
  - 'validate_s3_path() for cross-tenant S3 key validation'
test_patterns: []
---

# Tech-Spec: Doc-Extraction Tenant/Node Isolation (GAP-018)

**Created:** 2026-03-10

## Overview

### Problem Statement

Doc-extraction has zero tenant/node isolation at the data layer. All core PostgreSQL tables (`documents`, `extracted_records`, `domain_term_dictionary`, `symbol_dictionaries`) lack `tenant_id` and `node_id` columns. No PostgreSQL RLS policies exist. The Milvus `textiq_records` collection has no tenant/node fields. This violates SYS-FR-10, SYS-FR-11, SYS-FR-12, and SYS-NFR-09, enabling cross-tenant and cross-node data leakage.

Current isolation relies solely on `api_key_id` foreign key on the `documents` table — insufficient for multi-tenant/multi-node deployment and incompatible with the system-wide RBAC model.

### Solution

Add `tenant_id` (UUID, NOT NULL) and `node_id` (UUID, NOT NULL) columns to all core PostgreSQL tables with full RLS enforcement matching IAM's pattern. Add `tenant_id`/`node_id` fields to the Milvus `textiq_records` collection schema with INVERTED indexes. Propagate tenant/node context through Airflow DAG params. Update all retrieval and query code paths to filter by tenant/node.

### Scope

**In Scope:**
- Alembic migration: add `tenant_id` + `node_id` to `documents`, `extracted_records`, `domain_term_dictionary`, `symbol_dictionaries`
- PostgreSQL RLS policies on all affected tables (matching IAM's `current_setting('app.tenant_id')` pattern)
- Tenant-aware DB session: `SET LOCAL app.tenant_id` + `app.node_id` on every connection (API + Airflow)
- Milvus schema update: add `tenant_id` + `node_id` VARCHAR fields + INVERTED indexes to `textiq_records`
- Airflow DAG: add `tenant_id` + `node_id` as DAG params, propagate through all tasks
- All retrieval/query paths updated to include tenant/node filters (domain_aware_retriever, semantic_retriever, query routes)
- API routes: extract and pass tenant/node context on document creation and queries

**Out of Scope:**
- Chatbot-service `LlamaIndexRetrieverClient` changes (different repo — contract documented here)
- Deprecating `api_key_id` (keep as-is alongside new columns)
- GAP-017 HMAC header implementation (interim: tenant/node passed via API params or headers)
- SeaweedFS tenant isolation (GAP-012)
- MongoDB session scoping (GAP-013)

## Context for Development

### Codebase Patterns

**DB Session Management:**
- `src/db/session.py`: `async_session_maker` creates `AsyncSession` via `create_async_engine`. No middleware wrapping connections. Sessions provided via `get_async_session()` generator dependency.
- `src/api/dependencies.py`: `AsyncSessionDep = Annotated[AsyncSession, Depends(get_async_session)]` and `ApiKeyDep = Annotated[ApiKey, Depends(get_current_api_key)]`.
- **Airflow tasks create their own `AsyncEngine`** from `settings.database_url` — they do NOT use the shared session factory. RLS session vars must be set independently in each task.

**Two Milvus Write Paths (both via LlamaIndex):**
- `store_to_llamaindex_task` (hierarchical nodes): `MilvusVectorStore` + `PostgresDocumentStore` → stores ALL nodes to docstore, LEAF nodes to Milvus
- `store_large_table_to_llamaindex_task` (flat records): `MilvusVectorStore` only → 1 record = 1 Document

**Two Milvus Read Paths (both raw pymilvus):**
- `SemanticRetriever` (`src/retrieval/semantic_retriever.py:29`): Used by query API route. Requires `tenant_id` + `node_id` (raises `ValueError` if empty). Mandatory tenant/node filter prepended to all queries.
- `DomainAwareRetriever` (`src/knowledge/domain_aware_retriever.py:67`): Requires `tenant_id` + `node_id`. `_build_filter()` (L212) always returns non-None — tenant/node are mandatory first-level filters, followed by domain_category, domain_terms, tags.

**Milvus Collection Schema (`src/knowledge/vector_store.py:254-328`):**
13 fields: `id`, `vector`, `record_id`, `document_id`, `filename`, `document_type`, `text_content`, `metadata`, `domain_category`, `domain_terms`, `tags`, `tenant_id`, `node_id`. INVERTED indexes on both tenant/node fields.

**Known Gap — DomainTermDictionary unique constraint:**
- `DomainTermDictionary.term_original` still has `unique=True` (L295 in models.py). This was planned to change to a composite `UniqueConstraint('tenant_id', 'term_original')` but was **not implemented**. Two tenants currently cannot have the same domain term. Tracked for follow-up.

### Files to Reference

| File | Purpose |
| ---- | ------- |
| `src/db/models.py` | SQLAlchemy models — TenantIsolationMixin (L23), Document (L68), ExtractedRecord (L112), SymbolDictionary (L179), DomainTermDictionary (L287) |
| `src/api/tenant_context.py` | TenantContext dataclass, `get_tenant_context()` dependency, `set_rls_tenant()` helper |
| `src/db/session.py` | Session factory, `get_async_session()` dependency (unchanged — RLS set in route handlers, not here) |
| `src/api/dependencies.py` | `AsyncSessionDep`, `ApiKeyDep` (unchanged) |
| `src/api/routes/documents.py` | All document endpoints — `TenantContextDep` injected, defense-in-depth RLS + WHERE |
| `src/api/routes/query.py` | Query endpoint — `TenantContextDep` injected, `validate_document_ids()` scoped by tenant |
| `src/services/document_service.py` | `create_document()` — accepts and sets `tenant_id`/`node_id` |
| `src/knowledge/vector_store.py:254-328` | Milvus collection schema (13 fields) + INVERTED indexes on tenant/node |
| `src/knowledge/domain_aware_retriever.py:212` | `_build_filter()` — mandatory tenant/node as first-level filters |
| `src/retrieval/semantic_retriever.py:29` | `retrieve()` — requires tenant/node, raises ValueError if empty |
| `src/knowledge/term_index.py` | Tenant-scoped Redis cache (already uses `tenant_id` for cache keys) |
| `src/extraction_v2/domain_term_extractor.py` | `persist_new_terms()` — accepts tenant/node, sets on new DomainTermDictionary entries |
| `src/services/seaweedfs_service.py` | `validate_s3_path()`, tenant-scoped S3 keys for image upload/delete |
| `airflow/dags/document_extraction_dag.py` | DAG params include `tenant_id`/`node_id`, wired through all task calls |
| `airflow/plugins/operators/validate_event.py` | Extracts `tenant_id`/`node_id` from `dag_run.conf` |
| `airflow/plugins/operators/fetch_document.py` | Validates S3 key prefix matches `{tenant_id}/{node_id}/` |
| `airflow/dags/tasks/processing_tasks.py` | All store/extract tasks accept and propagate tenant/node context |
| `src/db/migrations/versions/20260313_10000*.py` | 3-phase migration: add columns → backfill → NOT NULL + RLS |

### Technical Decisions

- **RLS pattern**: Match IAM's approach — `SET LOCAL app.tenant_id = '{uuid}'` per transaction, RLS policy uses `current_setting('app.tenant_id')::uuid` with nil UUID fallback
- **RLS injection for API**: Created new `src/api/tenant_context.py` with `set_rls_tenant()` helper. Called explicitly by route handlers before queries (`src/db/session.py` was not modified — RLS is not injected at session factory level)
- **RLS injection for Airflow**: Not yet implemented — Airflow tasks create their own `AsyncEngine` and do not call `SET LOCAL`. RLS enforcement on Airflow worker connections is a follow-up item
- **Milvus collection**: `textiq_records` is now the code default in `src/core/config.py` (changed from `dsol_records`)
- **Milvus write path**: LlamaIndex `MilvusVectorStore` stores node metadata — `tenant_id`/`node_id` added to node metadata dict, picked up as top-level Milvus scalar fields
- **Milvus read path**: Raw pymilvus filter expressions in `DomainAwareRetriever._build_filter()` and `SemanticRetriever.retrieve()` — `tenant_id`/`node_id` are mandatory first-level filters (always non-None)
- **DAG propagation**: `tenant_id` + `node_id` added as explicit DAG params via `dag_run.conf`, extracted by `ValidateEventOperator`, passed through task chain via `get_tenant_id()`/`get_node_id()` TaskFlow tasks
- **`api_key_id`**: Retained as-is, not deprecated
- **Backfill strategy**: Existing data gets a default tenant/node UUID (single-tenant pilot per ADR-001, env var `PILOT_TENANT_ID`/`PILOT_NODE_ID`)
- **`DomainTermDictionary` unique constraint**: Planned to change `unique=True` on `term_original` to composite `UniqueConstraint('tenant_id', 'term_original')` — **not yet implemented** (follow-up required, two tenants cannot share the same term)
- **Chatbot-service contract**: Milvus `textiq_records` now has `tenant_id`/`node_id` fields. CB must add metadata pre-filters on queries — documented as interface contract, not implemented here

## Implementation Plan

### Tasks

#### Task 1: PostgreSQL Schema — TenantIsolationMixin + 3-Phase Migration
**Files:** `src/db/models.py`, `src/db/migrations/versions/20260313_10000{0,1,2}_*.py`

1. Created `TenantIsolationMixin` class with `tenant_id` and `node_id` (`UUID`, nullable initially, `index=False` — composite indexes added separately).
2. Applied mixin to `Document`, `ExtractedRecord`, `SymbolDictionary`, `DomainTermDictionary`. (`QueryLog` also has the mixin in code but is not actively used and has no migration.)
3. Added composite indexes: `idx_{table}_tenant_node` on `(tenant_id, node_id)` for each table.
4. **Migration Phase 1** (`7a1b2c3d4e5f`): Add nullable `tenant_id`/`node_id` columns + composite indexes to `documents`, `extracted_records`, `domain_term_dictionary`, `symbol_dictionaries`.
5. **Migration Phase 2** (`8a2b3c4d5e6f`): Backfill existing rows with pilot UUIDs (`PILOT_TENANT_ID`/`PILOT_NODE_ID` env vars, default `00000000-0000-0000-0000-000000000001`). Inherits from parent `documents` where possible; orphaned rows get pilot defaults.
6. **Migration Phase 3** (`9a3b4c5d6e7f`): `ALTER COLUMN ... NOT NULL`, `ENABLE ROW LEVEL SECURITY`, `FORCE ROW LEVEL SECURITY`, `CREATE POLICY tenant_isolation_{table}` using `COALESCE(NULLIF(current_setting('app.tenant_id', true), '')::uuid, '00000000-...'::uuid)` for both `USING` and `WITH CHECK`.

#### Task 2: FastAPI Tenant Context Dependency
**Files:** `src/api/tenant_context.py` (NEW)

1. Created `TenantContext` frozen dataclass with `tenant_id: UUID` and `node_id: UUID`.
2. Created `get_tenant_context()` FastAPI dependency — extracts from `X-Tenant-ID` / `X-Node-ID` headers, validates UUID format, returns 400 with structured error codes (`MISSING_TENANT_ID`, `INVALID_TENANT_ID`, etc.) on failure.
3. Exported `TenantContextDep = Annotated[TenantContext, Depends(get_tenant_context)]`.
4. Created `set_rls_tenant(session, tenant_id)` — executes `SET LOCAL app.tenant_id = :tid` using parameterized query (re-validates UUID to prevent injection).

#### Task 3: API Routes — Tenant Scoping
**Files:** `src/api/routes/documents.py`, `src/api/routes/query.py`, `src/services/document_service.py`

1. Added `ctx: TenantContextDep` to all document endpoints: `upload_document`, `upload_documents_batch`, `get_document_status`, `delete_document`, `list_documents`, `upload_custom_symbols`, `delete_custom_symbol`, `list_document_symbols`.
2. Added `ctx: TenantContextDep` to `query_documents` endpoint.
3. For read/delete endpoints: call `set_rls_tenant(session, ctx.tenant_id)` before queries + application-level `.where(Document.tenant_id == ctx.tenant_id).where(Document.node_id == ctx.node_id)` (defense-in-depth).
4. For symbol endpoints: added tenant/node ownership check alongside existing `api_key_id` check.
5. Updated `create_document()` service to accept and pass `tenant_id`/`node_id` to `Document()` constructor.
6. Updated `validate_document_ids()` in query route to accept optional `tenant_id`/`node_id` and add `.where()` clauses.
7. Updated `process_query()` to pass `tenant_id`/`node_id` to `SemanticRetriever.retrieve()`.

#### Task 4: Milvus Schema + Retrieval Filters
**Files:** `src/knowledge/vector_store.py`, `src/retrieval/semantic_retriever.py`, `src/knowledge/domain_aware_retriever.py`

1. Added `tenant_id` and `node_id` `FieldSchema(VARCHAR, max_length=36)` to collection schema.
2. Added INVERTED indexes on both fields.
3. Added `tenant_id`/`node_id` to `MilvusEntity` dataclass and columnar insert data.
4. Updated `_prepare_entities()`, `index_records()`, `index_with_vectors()` to accept and propagate `tenant_id`/`node_id`.
5. Updated `delete_vectors_by_document_id()` to include tenant/node in delete expression.
6. **SemanticRetriever**: Made `tenant_id`/`node_id` required params on `retrieve()`. Added `ValueError` if empty. Prepends mandatory `tenant_id == "..." and node_id == "..."` to all filter expressions.
7. **DomainAwareRetriever**: Changed `tenant_id` from `default` to required. Added `node_id` as required. `_build_filter()` now always returns non-None (mandatory tenant/node filters first).

#### Task 5: Airflow DAG — Param Propagation
**Files:** `airflow/dags/document_extraction_dag.py`, `airflow/plugins/operators/validate_event.py`, `airflow/plugins/operators/fetch_document.py`, `airflow/dags/tasks/processing_tasks.py`

1. Added `tenant_id` and `node_id` as `Param(type="string")` to DAG definition.
2. **ValidateEventOperator**: Extracts `tenant_id`/`node_id` from `dag_run.conf` for both S3 webhook and direct API trigger formats.
3. **FetchDocumentOperator**: Validates object_key starts with `{tenant_id}/{node_id}/` prefix — blocks cross-tenant file access with `ValueError`. Passes `tenant_id`/`node_id` through to output dict.
4. Added `get_tenant_id()` and `get_node_id()` TaskFlow tasks to extract from validate event result.
5. Wired `t_id`/`n_id` through to all downstream tasks: `save_document_metadata_task`, `upload_images_to_seaweedfs`, `extract_domain_terms_task`, `extract_domain_terms_nodes_task`, `store_to_llamaindex_task`, `store_large_table_to_llamaindex_task`.

#### Task 6: Processing Tasks — Tenant Context
**Files:** `airflow/dags/tasks/processing_tasks.py`

1. **save_document_metadata_task**: Accepts `tenant_id`/`node_id`, passes as `UUID()` to `Document()` constructor.
2. **upload_images_to_seaweedfs**: Accepts `tenant_id`/`node_id`, passes to `SeaweedFSService.upload_image()` for tenant-scoped S3 keys.
3. **extract_domain_terms_task / extract_domain_terms_nodes_task**: Passes `tenant_id`/`node_id` to `DomainTermExtractor.persist_new_terms()`.
4. **store_to_llamaindex_task**: Injects `tenant_id`/`node_id` into all `node.metadata` dicts before storage (picked up by LlamaIndex MilvusVectorStore as scalar fields).
5. **store_large_table_to_llamaindex_task**: Adds `tenant_id`/`node_id` to Document metadata for each large table record.

#### Task 7: SeaweedFS Tenant-Scoped Keys
**Files:** `src/services/seaweedfs_service.py`

1. Created `validate_s3_path()` — raises `PermissionError` if S3 key doesn't match `{tenant_id}/{node_id}/` prefix.
2. Updated `upload_image()`: key pattern changed from `documents/{doc_id}/images/...` to `{tenant_id}/{node_id}/{doc_id}/images/...` when tenant context provided. Legacy fallback retained.
3. Updated `upload_all_images()` and `delete_document_images()` to accept and propagate tenant/node context.

#### Task 8: Domain Term Extractor — Tenant Scoping
**Files:** `src/extraction_v2/domain_term_extractor.py`

1. Updated `persist_new_terms()` to accept optional `tenant_id`/`node_id` params.
2. New `DomainTermDictionary` entries created with `tenant_id`/`node_id` set.

#### Task 9: Upload Script + Config
**Files:** `scripts/upload_documents.py`, `src/core/config.py`

1. Added `--tenant-id` and `--node-id` CLI args (default: pilot UUID `00000000-0000-0000-0000-000000000001`).
2. Object key generation changed to `{tenant_id}/{node_id}/{document_id}/{filename}` format.
3. DAG trigger payload now includes `tenant_id`/`node_id`.
4. Changed `milvus_collection` default from `dsol_records` to `textiq_records`.

### Acceptance Criteria

**AC-1: PostgreSQL columns exist and are NOT NULL**
- Given the 3-phase migration has run
- When querying `documents`, `extracted_records`, `domain_term_dictionary`, `symbol_dictionaries`
- Then `tenant_id` and `node_id` columns are NOT NULL with composite indexes

**AC-2: RLS enforces tenant isolation**
- Given RLS policies are active on all tables
- When a session has `app.tenant_id` set to tenant A's UUID
- Then only rows with `tenant_id = A` are visible (SELECT/INSERT/UPDATE/DELETE)
- And if `app.tenant_id` is unset, the nil UUID fallback matches nothing

**AC-3: API endpoints require tenant headers**
- Given a request without `X-Tenant-ID` or `X-Node-ID` headers
- When calling any document or query endpoint
- Then a 400 response with `MISSING_TENANT_ID` / `MISSING_NODE_ID` error code is returned

**AC-4: API applies defense-in-depth filtering**
- Given a valid tenant context
- When querying documents or executing deletions
- Then both RLS (`SET LOCAL`) and application-level `.where()` clauses filter by tenant + node

**AC-5: Milvus collection has tenant/node fields**
- Given the Milvus collection is (re)created
- When inspecting the schema
- Then `tenant_id` and `node_id` VARCHAR(36) fields exist with INVERTED indexes

**AC-6: All Milvus queries filter by tenant + node**
- Given a query via `SemanticRetriever` or `DomainAwareRetriever`
- When the filter expression is built
- Then `tenant_id == "..."` and `node_id == "..."` are always the first filters (mandatory, never None)

**AC-7: SemanticRetriever rejects empty tenant context**
- Given `tenant_id` or `node_id` is empty string
- When `SemanticRetriever.retrieve()` is called
- Then `ValueError` is raised before any Milvus query

**AC-8: Airflow DAG propagates tenant context end-to-end**
- Given a DAG trigger with `tenant_id` and `node_id` in conf
- When the DAG executes
- Then `validate_event` extracts them, `fetch_document` validates the S3 prefix, and all downstream tasks receive and use them

**AC-9: FetchDocumentOperator blocks cross-tenant access**
- Given `tenant_id=A` and `node_id=B` in the event
- When `object_key` does not start with `A/B/`
- Then `ValueError` is raised and the task fails

**AC-10: SeaweedFS keys follow tenant-scoped pattern**
- Given image upload with `tenant_id` and `node_id`
- When `SeaweedFSService.upload_image()` runs
- Then S3 key is `{tenant_id}/{node_id}/{doc_id}/images/{sheet}/{image}.{ext}`
- And `validate_s3_path()` confirms the prefix

**AC-11: Existing data is backfilled**
- Given the Phase 2 migration runs
- When inspecting previously-existing rows
- Then all have `tenant_id` and `node_id` set to pilot UUIDs (env var or default `00000000-0000-0000-0000-000000000001`)

**AC-12: Domain terms are tenant-scoped**
- Given a new domain term is persisted
- When `DomainTermExtractor.persist_new_terms()` is called with tenant context
- Then the `DomainTermDictionary` entry has `tenant_id` and `node_id` set

## Additional Context

### Dependencies

- **GAP-017 (HMAC headers)**: Not yet implemented. Interim: tenant/node context arrives via API request headers (`X-Tenant-ID`, `X-Node-ID`) or DAG params from the triggering system (Admin Portal API currently sends `X-Scope-ID`)
- **GAP-003 (Retrieval scope control)**: Depends on this GAP — `node_id` must exist in data layer before scoped retrieval is possible
- **GAP-012 (SeaweedFS isolation)**: Independent but related — key prefix convention uses same `tenant_id`
- **ADR-001 (Single tenant pilot)**: Backfill uses single default tenant UUID; RLS is forward-compatible for multi-tenant

### Testing Strategy

**Unit Tests:**
- `TenantContext` dependency: valid headers, missing headers (400), invalid UUID (400)
- `set_rls_tenant()`: parameterized query injection safety
- `validate_s3_path()`: matching prefix passes, mismatched prefix raises `PermissionError`
- `SemanticRetriever.retrieve()`: raises `ValueError` on empty tenant/node
- `DomainAwareRetriever._build_filter()`: always includes tenant/node as first filters

**Integration Tests:**
- 3-phase migration: run forward + backward, verify column state at each phase
- RLS: insert rows for tenant A, set session to tenant B, verify SELECT returns empty
- API endpoints: confirm 400 without headers, confirm scoped results with headers
- Milvus: insert with tenant A, query with tenant B filter, verify zero results

**End-to-End:**
- Upload document via script with `--tenant-id`/`--node-id`, trigger DAG, verify:
  - PostgreSQL records have correct `tenant_id`/`node_id`
  - Milvus vectors have `tenant_id`/`node_id` fields populated
  - SeaweedFS keys follow `{tenant_id}/{node_id}/...` pattern
  - Query API with matching tenant headers returns results
  - Query API with different tenant headers returns zero results

### Known Gaps / Follow-ups

- **`DomainTermDictionary.term_original` unique constraint**: Still global `unique=True` — needs composite `UniqueConstraint('tenant_id', 'term_original')` for true multi-tenant support
- **Airflow RLS**: Airflow tasks create their own `AsyncEngine` but do not call `SET LOCAL app.tenant_id`. RLS policies exist on the tables but are not enforced on Airflow worker connections. Tasks rely on application-level `tenant_id`/`node_id` propagation only
- **`query_logs` migration**: `QueryLog` model has `TenantIsolationMixin` in code but the table is not actively used and was excluded from the 3-phase migration (no column add, no backfill, no RLS policy)
- **Milvus schema migration**: Adding `tenant_id`/`node_id` fields requires collection recreation — existing Milvus data must be re-indexed
- **LlamaIndex auto-create**: `MilvusVectorStore` auto-creates collection if missing — tenant/node fields are injected via node metadata, picked up as scalar fields by the store