---
title: 'Airflow Plugin Endpoints — Update Domain Terms & Tags'
slug: 'airflow-plugin-update-tags-terms'
created: '2026-04-02'
status: 'done'
baseline_commit: 'a8c4240d370bb6a24f9033b2831c31c7311d2ccb'
stepsCompleted: [1, 2, 3, 4]
tech_stack: ['FastAPI', 'SQLAlchemy 1.4 (sync)', 'pymilvus', 'PostgreSQL', 'Milvus']
files_to_modify:
  - 'src/knowledge/domain_dictionary_service.py'
  - 'airflow/plugins/api/rls.py (new)'
  - 'airflow/plugins/api/milvus.py (new)'
  - 'airflow/plugins/api/tags.py (new)'
  - 'airflow/plugins/api/domain_terms.py (new)'
  - 'airflow/plugins/__init__.py'
  - 'tests/airflow/test_plugins/test_tags.py (new)'
  - 'tests/airflow/test_plugins/test_domain_terms.py (new)'
code_patterns:
  - 'Sync FastAPI on Airflow api-server (no async, no src.* imports in plugin)'
  - 'Raw SQL via sqlalchemy.text() with RLS SET LOCAL app.tenant_id'
  - 'HMAC auth middleware (same as upload.py)'
  - 'Milvus partial upsert via MilvusClient.upsert(partial_update=True) — pymilvus 2.6.4'
  - 'PG-primary consistency: PG txn → Milvus upsert → rollback PG if Milvus fails'
  - 'error_response() / success_response() from api.responses'
test_patterns:
  - 'sys.path injection for airflow plugin imports (tests/airflow/test_plugins/)'
  - 'FastAPI TestClient with mocked HMAC middleware'
  - 'unittest.mock.patch for verify_hmac_headers'
---

# Tech-Spec: Airflow Plugin Endpoints — Update Domain Terms & Tags

**Created:** 2026-04-02

## Overview

### Problem Statement

APA (Admin Portal API) currently writes directly to Doc Extraction-owned stores (`data_hierarchical_nodes`, Milvus `textiq_records_new`, `domain_term_dictionary`), violating CQRS-lite architecture (Architecture Section 4.9). Additionally, the domain term batch insert fails entirely when a single duplicate term is encountered due to a globally-scoped unique constraint and lack of `ON CONFLICT` handling.

### Solution

Build tag and domain term update API endpoints in the Airflow plugin (`airflow/plugins/api/`) so APA calls DE instead of writing directly. Fix the domain term batch insert to handle duplicates gracefully.

### Scope

**In Scope:**
- 3 tag update endpoints (update status, bulk update status, rename tag)
- 1 domain term endpoint (update term) with cascading sync to chunks
- Each tag endpoint syncs both `data_hierarchical_nodes` (PostgreSQL) and Milvus `textiq_records_new`
- RLS tenant isolation helper
- Milvus update helper (delete + re-insert pattern)
- Fix domain term batch insert to use `ON CONFLICT DO NOTHING` (global unique on `term_original` stays)

**Out of Scope:**
- Domain term create endpoints (create/bulk create terms via API) — future iteration
- File storage endpoints (SeaweedFS delete) — future iteration
- `extracted_records` table (empty, unused in hierarchical pipeline)
- Tag creation endpoints (tags are created by the extraction DAG, not by APA)

## Context for Development

### Data Model

**Chunk ID** = same UUID across both stores:
- `data_hierarchical_nodes.key` (where `namespace = 'hierarchical/data'`)
- Milvus `textiq_records_new.id`

**Tag object structure** (identical in PG and Milvus):
```json
{
  "tag_id": "c3a6a76d-bf36-4581-a62d-db5e0283cf05",
  "name": "domain:介護",
  "source": "ai",
  "confidence": 0.85,
  "status": "pending",
  "history": [
    {"action": "suggested", "by": "ai", "ts": 1774929953}
  ]
}
```

**Tag status enum:** `pending` | `verified` | `rejected` | `deleted`

**Tags location in PG:** `value->'__data__'->'metadata'->'tags'` (JSONB array)

**Domain terms in chunks:** `value->'__data__'->'metadata'->'domain_terms'` (string array, e.g. `["病棟", "入院患者数"]`)

**Domain term dictionary schema:**
| Column | Type | Notes |
|--------|------|-------|
| `id` | UUID PK | |
| `term_original` | VARCHAR(255) | Currently UNIQUE globally — needs scoping to tenant |
| `term_ja` | VARCHAR(255) | Japanese translation |
| `term_en` | VARCHAR(255) | English translation |
| `term_vi` | VARCHAR(255) | Vietnamese translation |
| `source_language` | VARCHAR(10) | `ja`, `en`, or `vi` |
| `domain_category` | VARCHAR(100) | Default `general` |
| `source_document_id` | UUID FK | Nullable |
| `translation_failed` | BOOLEAN | Default `false` |
| `tenant_id` | UUID | RLS enforced |
| `node_id` | UUID | |

**All PG tables enforce RLS** via `SET LOCAL app.tenant_id = :tid`

### Input Validation Rules (Pydantic Models)

All endpoints use Pydantic `BaseModel` for request body validation. Define in each endpoint file.

**Tag endpoints (`tags.py`):**
```python
from pydantic import BaseModel, Field, field_validator
from typing import Literal

class TagStatusUpdate(BaseModel):
    tag_id: str
    status: Literal["verified", "rejected"]

class BulkTagStatusUpdate(BaseModel):
    updates: list[TagStatusUpdate] = Field(min_length=1)

class TagRename(BaseModel):
    name: str = Field(min_length=1, max_length=255)

    @field_validator("name")
    @classmethod
    def must_have_namespace(cls, v: str) -> str:
        if ":" not in v:
            raise ValueError("Tag name must match namespace:value pattern (e.g. domain:nursing)")
        return v
```
- `name` must not duplicate an existing tag name on the same chunk (checked in handler logic)

**Domain term endpoint (`domain_terms.py`):**
```python
from pydantic import BaseModel, Field, model_validator

class DomainTermUpdate(BaseModel):
    term_original: str | None = Field(None, min_length=1, max_length=255)
    term_ja: str | None = Field(None, max_length=255)
    term_en: str | None = Field(None, max_length=255)
    term_vi: str | None = Field(None, max_length=255)
    domain_category: str | None = Field(None, min_length=1, max_length=100)

    @model_validator(mode="after")
    def at_least_one_field(self):
        if not any([self.term_original, self.term_ja, self.term_en, self.term_vi, self.domain_category]):
            raise ValueError("At least one field must be provided")
        return self
```
- If `term_original` is changed, must not conflict with existing global unique constraint (checked in handler, returns 409)

### Codebase Patterns

1. **Airflow plugin API runs inside the Airflow api-server process** — uses `/home/airflow/.local/bin/python`. **Must NOT import from `src.*`**.
2. **Sync only** — SQLAlchemy 1.4 sync engine, sync FastAPI route handlers (threadpool).
3. **Raw SQL via `text()`** — no ORM models available in plugin context.
4. **HMAC auth** — all endpoints use `HMACVerificationMiddleware`. Context dict: `{"tenant_id", "node_id", "user_id", "role"}` from `request.state.hmac_ctx`.
5. **Milvus supports partial upsert** (pymilvus 2.6.4) — `MilvusClient.upsert(collection_name, data, partial_update=True)` updates only specified fields, others remain unchanged. No need to read/delete/re-insert.
6. **Milvus collection uses LlamaIndex dynamic fields** — schema has 5 fixed fields (`id`, `doc_id`, `text`, `embedding`, `sparse_embedding`) + dynamic fields (`tags`, `domain_terms`, `tenant_id`, etc.).
7. **Consistency model: PG is primary, Milvus follows.** Start PG transaction → update PG → upsert Milvus → if Milvus fails, rollback PG. Both succeed or neither does.
7. **Tag history `by` field** — use `request.state.hmac_ctx["user_id"]` for audit trail.
8. **Response format** — `error_response(code, message, status_code)` and `success_response(data, status_code)` from `api.responses`.

### Files to Reference

| File | Purpose |
| ---- | ------- |
| `airflow/plugins/api/upload.py` | Reference pattern: FastAPI app, middleware registration, DB txn, error handling |
| `airflow/plugins/api/middleware.py` | HMAC, RequestId, CorrelationId middleware classes |
| `airflow/plugins/api/db.py` | Sync SQLAlchemy engine (lazy init, `pool_size=3`) |
| `airflow/plugins/api/responses.py` | `error_response()` and `success_response()` helpers |
| `airflow/plugins/api/auth.py` | `verify_hmac_headers()` returns `{tenant_id, node_id, user_id, role}` |
| `airflow/plugins/__init__.py` | Plugin registration (`TextIQPlugin.fastapi_apps` list) |
| `src/knowledge/domain_dictionary_service.py` | `add_terms_batch()` (line 44) — needs ON CONFLICT fix |
| `src/extraction_v2/domain_term_extractor.py` | `persist_new_terms()` (line 839) — calls add_terms_batch |
| `tests/airflow/test_plugins/test_middleware.py` | Reference for test structure: sys.path injection, TestClient, mocking |

### Technical Decisions

1. **Separate FastAPI apps** — `tags_app` at `/api/v1/blocks`, `domain_terms_app` at `/api/v1/domain-terms`, same pattern as `upload_app` at `/api/v1/documents`
2. **Chunk ID in URL** — `/blocks/{chunk_id}/tags/...` uses UUID from `data_hierarchical_nodes.key` = Milvus `id`
3. **Milvus partial upsert** — `MilvusClient.upsert(collection_name, data=[{"id": chunk_id, "tags": updated_tags}], partial_update=True)` updates only specified fields, no need for delete+re-insert
4. **Unique constraint stays global** — `(term_original)` per original spec design (global shared dictionary). The batch insert fix uses `ON CONFLICT DO NOTHING` on `term_original` only
5. **Domain term cascading sync** — when `term_original` changes or term is deleted, update/remove from all chunks' `domain_terms` arrays in both PG and Milvus

## Implementation Plan

### Tasks

- [x] **Task 0: Fix batch insert — ON CONFLICT DO NOTHING**
  - File: `src/knowledge/domain_dictionary_service.py` — replace `add_terms_batch()` (line 44-53)
  - Action: Replace `session.add_all(entries) + commit` with `insert().on_conflict_do_nothing(index_elements=["term_original", "tenant_id", "node_id"])`. Return `result.rowcount` instead of `len(entries)`.
  - ```python
    def add_terms_batch(self, entries: list[DomainTermDictionary]) -> int:
        try:
            self._set_rls_context()
            from sqlalchemy.dialects.postgresql import insert
            stmt = insert(DomainTermDictionary).values([
                {
                    "id": entry.id, "term_original": entry.term_original,
                    "term_ja": entry.term_ja, "term_en": entry.term_en, "term_vi": entry.term_vi,
                    "source_language": entry.source_language, "domain_category": entry.domain_category,
                    "source_document_id": entry.source_document_id,
                    "translation_failed": entry.translation_failed,
                    "tenant_id": entry.tenant_id, "node_id": entry.node_id,
                } for entry in entries
            ]).on_conflict_do_nothing(index_elements=["term_original"])
            result = self.session.execute(stmt)
            self.session.commit()
            return result.rowcount
        except Exception:
            self.session.rollback()
            raise
    ```

- [x] **Task 1: Create RLS helper**
  - File: `airflow/plugins/api/rls.py` (new)
  - Action: Context manager that opens sync connection, sets RLS tenant, yields, commits/rollbacks
  - ```python
    from contextlib import contextmanager
    from sqlalchemy import text
    from api.db import get_engine

    @contextmanager
    def rls_connection(tenant_id: str):
        engine = get_engine()
        conn = engine.connect()
        txn = conn.begin()
        try:
            conn.execute(text("SET LOCAL app.tenant_id = :tid"), {"tid": tenant_id})
            yield conn
            txn.commit()
        except Exception:
            txn.rollback()
            raise
        finally:
            conn.close()
    ```

- [x] **Task 2: Create Milvus helper**
  - File: `airflow/plugins/api/milvus.py` (new)
  - Action: Lazy-initialized `MilvusClient` singleton. Helper functions for partial upsert.
  - Functions:
    - `get_client() -> MilvusClient` — lazy singleton, connects using env vars
    - `upsert_chunk_field(chunk_id: str, timeout: float = 10.0, **fields)` — partial upsert: `client.upsert(COLLECTION, data=[{"id": chunk_id, **fields}], partial_update=True, timeout=timeout)`. Timeout prevents holding PG connections indefinitely if Milvus is slow.
    - `query_chunks_with_term(term: str) -> list[dict]` — find chunks containing a domain term for cascading
  - Notes:
    - Uses `MilvusClient` (not `Collection`) for `partial_update=True` support (pymilvus 2.6.4)
    - Lazy singleton pattern matching `api/db.py` (`_client = None`)
    - Env vars: `MILVUS_HOST`, `MILVUS_PORT`, `MILVUS_COLLECTION`, `MILVUS_USER`, `MILVUS_PASSWORD`

- [x] **Task 3: Create tag endpoints**
  - File: `airflow/plugins/api/tags.py` (new)
  - Action: Create `tags_app = FastAPI(title="TextIQ Tags API")` with middleware stack (HMAC, CorrelationId, RequestId). Implement 3 sync route handlers:
    - `PATCH /{chunk_id}/tags/status` — body: `{"tag_id": "uuid", "status": "verified|rejected"}`. Find tag by tag_id in JSONB array, update status, append history `{"action": "status_changed", "by": ctx["user_id"], "ts": <unix_ts>}`. PG first, then Milvus partial upsert.
    - `PATCH /{chunk_id}/tags/bulk-status` — body: `{"updates": [{"tag_id": "uuid", "status": "..."}]}`. **Validate all tag_ids exist first** — if any tag_id is not found, return 400 with the missing tag_id(s) and do not update any. If all valid, update all in one PG transaction, then one Milvus partial upsert.
    - `PATCH /{chunk_id}/tags/{tag_id}/name` — body: `{"name": "new:tag:name"}`. Update tag name, append history `{"action": "renamed", "by": ..., "ts": ..., "old_name": "..."}`. PG first, then Milvus partial upsert.
  - **Consistency pattern (all endpoints):**
    1. Begin PG transaction (via `rls_connection`)
    2. Read `value` from `data_hierarchical_nodes` with `SELECT ... FOR UPDATE` where `key = :chunk_id AND namespace = 'hierarchical/data'` (row lock prevents concurrent read-modify-write race)
    3. Manipulate tags in Python, write back with `jsonb_set(value, '{__data__,metadata,tags}', :updated_tags::jsonb)`
    4. Call `upsert_chunk_field(chunk_id, tags=updated_tags)` (Milvus partial upsert, timeout=10s)
    5. If Milvus fails → rollback PG transaction
    6. If both succeed → commit PG
  - Error cases: chunk not found → 404, tag not found → 404, invalid status → 400, Milvus failure → 500 (PG rolled back).

- [x] **Task 4: Create domain term endpoints**
  - File: `airflow/plugins/api/domain_terms.py` (new)
  - Action: Create `domain_terms_app = FastAPI(title="TextIQ Domain Terms API")` with middleware stack. Implement 1 sync route handler:
    - `PATCH /{term_id}` — body: partial `{"term_original": "...", "term_ja": "...", "term_en": "...", "term_vi": "...", "domain_category": "..."}` (all optional, at least one required). Update `domain_term_dictionary` with `COALESCE`. **If `term_original` changed**: cascade rename in `data_hierarchical_nodes` and Milvus.
  - PG cascade pattern:
    ```sql
    -- Rename term in all chunks
    UPDATE data_hierarchical_nodes
    SET value = jsonb_set(value, '{__data__,metadata,domain_terms}',
        (SELECT jsonb_agg(
            CASE WHEN elem #>> '{}' = :old_term THEN to_jsonb(:new_term::text) ELSE elem END
        ) FROM jsonb_array_elements(value->'__data__'->'metadata'->'domain_terms') AS elem))
    WHERE namespace = 'hierarchical/data'
      AND value->'__data__'->'metadata'->'domain_terms' @> to_jsonb(:old_term::text);
    ```
  - Milvus cascade: query affected chunks by term, partial upsert `domain_terms` field for each: `upsert_chunk_field(chunk_id, domain_terms=updated_terms)`.
  - Error cases: term not found → 404, no fields provided → 400.

- [x] **Task 5: Register in plugin**
  - File: `airflow/plugins/__init__.py`
  - Action: Import `tags_app` and `domain_terms_app` using the existing fallback pattern, add to `TextIQPlugin.fastapi_apps`:
    ```python
    try:
        from api.upload import upload_app
        from api.tags import tags_app
        from api.domain_terms import domain_terms_app
    except ImportError:
        from airflow.plugins.api.upload import upload_app
        from airflow.plugins.api.tags import tags_app
        from airflow.plugins.api.domain_terms import domain_terms_app

    class TextIQPlugin(AirflowPlugin):
        name = "textiq"
        fastapi_apps = [
            {"name": "upload", "app": upload_app, "url_prefix": "/api/v1/documents"},
            {"name": "tags", "app": tags_app, "url_prefix": "/api/v1/blocks"},
            {"name": "domain_terms", "app": domain_terms_app, "url_prefix": "/api/v1/domain-terms"},
        ]
    ```

- [x] **Task 6: Write tests**
  - File: `tests/airflow/test_plugins/test_tags.py` (new)
  - File: `tests/airflow/test_plugins/test_domain_terms.py` (new)
  - Action: Follow `test_middleware.py` pattern (sys.path injection, TestClient, mocked HMAC). Test happy path + error cases for each endpoint. Mock `api.db.get_engine` and `api.milvus.get_client` for unit isolation.

### Acceptance Criteria

- [ ] **AC1:** Given a chunk with a pending tag, when `PATCH /api/v1/blocks/{chunk_id}/tags/status` is called with `{"tag_id": "...", "status": "verified"}`, then the tag status is updated to "verified" in both PG and Milvus, and a history entry `{"action": "status_changed", "by": "<user_id>", "ts": <unix_ts>}` is appended.
- [ ] **AC2:** Given a chunk with 3 pending tags, when `PATCH /api/v1/blocks/{chunk_id}/tags/bulk-status` is called with updates for 2 tags, then both tags are updated in both PG and Milvus in a single transaction, and the third tag is unchanged.
- [ ] **AC2b:** Given a chunk with 3 tags, when `PATCH /api/v1/blocks/{chunk_id}/tags/bulk-status` is called with updates including a non-existent tag_id, then 400 is returned with the missing tag_id(s), and no tags are updated.
- [ ] **AC3:** Given a chunk with tag `name: "topic:coronavirus"`, when `PATCH /api/v1/blocks/{chunk_id}/tags/{tag_id}/name` is called with `{"name": "topic:covid19"}`, then the tag name is updated in both PG and Milvus, and a history entry `{"action": "renamed", "old_name": "topic:coronavirus", ...}` is appended.
- [ ] **AC4:** Given a tag status update with `{"status": "invalid_value"}`, when the endpoint is called, then 400 is returned with validation error.
- [ ] **AC4b:** Given a tag rename with `{"name": ""}` or a name exceeding 255 chars, when the endpoint is called, then 400 is returned with validation error.
- [ ] **AC4c:** Given a tag rename with `{"name": "domain:nursing"}` that already exists on the same chunk, when the endpoint is called, then 400 is returned with "duplicate tag name" error.
- [ ] **AC4d:** Given a domain term update with `{"term_original": "existing_term"}` that conflicts with the global unique constraint, when the endpoint is called, then 409 Conflict is returned.
- [ ] **AC5:** Given chunk belongs to tenant A, when request comes with tenant B HMAC context, then the operation returns 404 (RLS filters it out).
- [ ] **AC6:** Given a request without valid HMAC headers, when any endpoint is called, then 401 Unauthorized is returned.
- [ ] **AC7:** Given `domain_term_dictionary` already contains term "感染症" for tenant A, when the DAG inserts a batch of 86 terms including "感染症" for tenant A, then the duplicate is skipped and the other 85 terms are inserted successfully.
- [ ] **AC8:** Given term "感染症" exists with `term_en: "infectious disease"`, when `PATCH /api/v1/domain-terms/{term_id}` is called with `{"term_en": "infection"}`, then `term_en` is updated and other fields remain unchanged, and no chunk sync occurs.
- [ ] **AC9:** Given term "感染症" exists and 5 chunks contain it in `domain_terms`, when `PATCH /api/v1/domain-terms/{term_id}` is called with `{"term_original": "伝染病"}`, then `domain_term_dictionary.term_original` is updated, and all 5 chunks' `domain_terms` arrays are updated from "感染症" to "伝染病" in both PG and Milvus.
- [ ] **AC10:** Given term belongs to tenant A, when request comes with tenant B HMAC context, then the operation returns 404.
- [ ] **AC11:** Given term "感染症" already exists in the global dictionary, when the DAG extracts the same term from a different tenant's document, then the duplicate is skipped (global unique constraint).

## Additional Context

### Dependencies

- `pymilvus` must be available in the Airflow api-server Python environment
- APA team needs to update their `ExtractedRecordRepository` and `DomainTermService` calls to use the new HTTP endpoints with HMAC auth

### Testing Strategy

**Unit tests** (mocked DB + Milvus):
- Each endpoint: happy path, 404, 400 validation errors
- RLS isolation: verify `SET LOCAL app.tenant_id` is called with correct value
- Milvus helper: verify delete+re-insert sequence

**Integration tests** (real DB + Milvus on local):
- Insert test data into PG + Milvus, call endpoint, verify both stores updated
- Domain term cascade: rename/delete term, verify all affected chunks synced
- Batch insert: insert with known duplicate, verify partial success

### Notes

- `extracted_records` table is empty and unused in the hierarchical pipeline — not a target
- Milvus delete+re-insert means a brief moment where the record is missing — acceptable for metadata updates (not affecting search availability significantly)
- APA identifies chunks by `record_id` — this maps to `data_hierarchical_nodes.key` = Milvus `id`
- Domain term cascade to Milvus could be slow if many chunks are affected — partial upsert is faster than delete+re-insert but still O(n) Milvus calls. Consider batching or async for large cascades in future iteration
- Consistency model: PG-primary. PG transaction wraps both PG write and Milvus upsert. If Milvus fails, PG rolls back. Both succeed or neither does
- The `_node_content` dynamic field in Milvus also contains tags/domain_terms in its JSON — but this is LlamaIndex internal metadata and is not used for filtering, so we do NOT update it (only `tags` and `domain_terms` top-level dynamic fields)

## Suggested Review Order

**Shared infrastructure**

- RLS context manager — transaction boundary for all endpoints
  [`rls.py:18`](../../airflow/plugins/api/rls.py#L18)

- Milvus singleton + partial upsert helper with filter-injection escaping
  [`milvus.py:30`](../../airflow/plugins/api/milvus.py#L30)

**Tag endpoints**

- Entry point: 3 PATCH handlers with PG-first/Milvus-follow consistency
  [`tags.py:114`](../../airflow/plugins/api/tags.py#L114)

- `_EarlyReturn` pattern prevents accidental commit on error-path returns
  [`tags.py:70`](../../airflow/plugins/api/tags.py#L70)

- `SELECT ... FOR UPDATE` row lock + JSONB manipulation helpers
  [`tags.py:77`](../../airflow/plugins/api/tags.py#L77)

**Domain term endpoint**

- PATCH handler with cascading rename to PG chunks and Milvus
  [`domain_terms.py:96`](../../airflow/plugins/api/domain_terms.py#L96)

- Global uniqueness check (outside RLS) for cross-tenant constraint
  [`domain_terms.py:68`](../../airflow/plugins/api/domain_terms.py#L68)

**Batch insert fix**

- `ON CONFLICT DO NOTHING` replaces `add_all` to skip duplicates gracefully
  [`domain_dictionary_service.py:47`](../../src/knowledge/domain_dictionary_service.py#L47)

**Plugin registration**

- New apps added to `TextIQPlugin.fastapi_apps`
  [`__init__.py:17`](../../airflow/plugins/__init__.py#L17)

**Tests**

- Tag endpoint tests: happy path, 404, 400, 422, 401
  [`test_tags.py:1`](../../tests/airflow/test_plugins/test_tags.py#L1)

- Domain term tests: cascade, conflict, not-found, auth
  [`test_domain_terms.py:1`](../../tests/airflow/test_plugins/test_domain_terms.py#L1)
