---
title: "DE Airflow Gap Audit — System Gaps & Changes"
author: nguyennh25
date: 2026-04-20
document_type: audit-report
source: _planning/docs/planning-artifacts/2-planning/system-gaps-and-changes.md
scope: airflow/ directory of textiq-doc-extraction
status: final
---

# DE Airflow Gap Audit — 2026-04-20

Per-gap assessment of the current state of `airflow/` (DAGs + FastAPI plugin) in
the `textiq-doc-extraction` repository against every DE-tagged gap in
[`system-gaps-and-changes.md`](../../_planning/docs/planning-artifacts/2-planning/system-gaps-and-changes.md).

## Executive Summary

| Category | Count | Gaps |
|---|---|---|
| Resolved | 9 | GAP-011 (TEXTIQ-389), GAP-012, GAP-018, GAP-024 (DE scope), GAP-030, GAP-033, GAP-038, GAP-040, GAP-010 §10.1 (code-level) |
| Partially resolved | 2 | GAP-041, GAP-042 |
| Not resolved | 5 | GAP-027, GAP-029, GAP-031, GAP-039, GAP-043 |

Biggest wins: tenant/node isolation (PostgreSQL RLS + Milvus scalar filters + SeaweedFS key prefix), HMAC authentication on the plugin, correlation/request ID middleware, the APA→DE multipart upload path that unblocks the CQRS-lite single-writer model, and domain-term confidence scoring (TEXTIQ-389).

Largest remaining work: Socket.IO pipeline events, observability (Prometheus + circuit breaker), and the LLM provider abstraction.

---

## Audit Matrix

| Gap | Title | Priority | Status | One-line verdict |
|---|---|---|---|---|
| GAP-010 §10.1 | DE schema divergence | Medium | 🟡 Code aligned, docs pending | Code uses the real table names; architecture doc still lags. |
| GAP-011 | Domain-term confidence scores | Medium | ✅ Resolved (TEXTIQ-389) | `confidence FLOAT NOT NULL CHECK 0..1` column live; rule=0.95, LLM=min-prob from logprobs, API default=1.0, fallback=0.70. |
| GAP-012 | SeaweedFS tenant/node key prefix | High | ✅ Resolved | `{tenant}/{node}/{doc_id}/{filename}` enforced on upload + delete. |
| GAP-018 | Core tables missing tenant/node/RLS | High | ✅ Resolved | 4 core tables + hierarchical nodes have columns, NOT NULL, RLS. |
| GAP-024 | Milvus collection name | Medium | ✅ Resolved (DE scope) | DE default aligned to `textiq_records` (matches `src/core/config.py` + `.env.example`). |
| GAP-027 | Socket.IO pipeline events | Medium | ❌ Not resolved | No `socketio`, still polling-only. |
| GAP-029 | Prometheus metrics | Medium | ❌ Not resolved | No `prometheus_client`, no `/metrics`. |
| GAP-030 | X-Correlation-ID / X-Request-ID | Medium | ✅ Resolved (plugin) | Both middlewares present; IDs propagated into DAG conf. |
| GAP-031 | Circuit breaker | Medium | ❌ Not resolved | No circuit breaker library/logic. |
| GAP-033 | Milvus Strategy C (scalar filtering) | High | ✅ Resolved | `tenant_id` + `node_id` scalar filters on all Milvus query/delete. |
| GAP-038 | APA→DE multipart upload (CQRS) | High (P1) | ✅ Resolved | DE receives multipart → writes SeaweedFS → triggers DAG. |
| GAP-039 | LLM provider abstraction | High | ❌ Not resolved | `text-embedding-3-large` hardcoded; no `LLMProvider`. |
| GAP-040 | Hierarchical chunk sizes (4096/1024/256) | High | ✅ Resolved | Defaults canonical in code and docstrings. |
| GAP-041 | HITL tag cross-service flow | High | 🟡 Partially addressed | Tag status/rename/delete APIs exist; no formal cross-service spec. |
| GAP-042 | Domain term dictionary cross-service sync | Medium | 🟡 Partially addressed | DE CRUD + cascade to Milvus/PG done; push to DG undefined. |
| GAP-043 | Chunk schema versioning | High | ❌ Not resolved | No `schema_version` on chunks or DAG metadata. |

Legend: ✅ resolved · 🟡 partially resolved · ❌ not resolved.

---

## Detailed Findings

### GAP-010 §10.1 — DE schema divergence — 🟡 Code aligned, docs pending

**Summary:** Airflow code matches the real table names (which diverge from architecture spec); doc sync is still pending.

**Evidence:**
- `airflow/plugins/api/upload.py` writes to `documents`, reads/cascades through `extracted_records`, `domain_term_dictionary`, `symbol_dictionaries`, `data_hierarchical_nodes`.
- Upload bucket default: `airflow/plugins/api/config.py:9` → `source-files`.
- Image bucket default: `airflow/plugins/api/upload.py:225` → `textiq-images`.

**Action remaining:** Update `system-architecture.md` §4.0–4.1 to adopt the real names; keep `textiq-images` (or rename once, across code + docs).

---

### GAP-011 — Domain-term confidence scores — ✅ Resolved (TEXTIQ-389)

**Summary:** SYS-FR-47 confidence scoring is now live end-to-end. `domain_term_dictionary.confidence FLOAT NOT NULL CHECK (BETWEEN 0 AND 1)` is populated on every write path; the API, the DAG, and the batch insert all stamp a score per term.

**Evidence:**
- Migration `src/db/migrations/versions/20260422_100000_add_domain_term_confidence.py` — reversible; upgrade backfills `0.70` then drops the temporary default.
- Model `src/db/models.py:300` — `confidence: Mapped[float]` + `CheckConstraint("confidence >= 0 AND confidence <= 1")`.
- `src/extraction_v2/domain_term_extractor.py` — rule-based hits = **0.95**; LLM terms = `exp(min(token_logprobs))` (min-prob aggregation, `logprobs=True, top_logprobs=0` on the chat call); alignment/logprob failure = **0.70** fallback; rule∪LLM overlap takes `max`; `persist_new_terms(confidence_by_term=…)` threads the score into the ORM entity.
- `src/knowledge/domain_dictionary_service.py:94` — `add_terms_batch` includes `confidence` in the `postgresql.insert().values(...)` dict (fix for initial NotNullViolation).
- `airflow/plugins/api/domain_terms.py` — `DomainTermCreate.confidence: float | None` (Field `ge=0, le=1`); POST/bulk/PATCH accept and echo it; `None` defaults server-side to **1.0** (tag-spec Human=1.0 precedent).
- `airflow/dags/tasks/processing_tasks.py:795-860,1099-1121` — both DAG call sites aggregate per-chunk confidences via `max` across chunks and forward to `persist_new_terms`.

**Verification (end-to-end):**
- Fresh-node DAG run (`node_id=44444444-…`, 1706.03762v7.pdf / 150 nodes) produced 377 terms with distribution 265 GREEN (≥0.90) / 54 YELLOW (0.70–0.90) / 58 RED (<0.70); range 0.33–1.00, mean 0.89.
- Plugin API round-trip (omit → 1.0; explicit 0.82; bulk mixed) persisted every value verbatim and echoed in response bodies.
- Tests: 77/77 pass across `tests/knowledge/test_domain_dictionary_service.py`, `tests/extraction_v2/test_domain_term_extractor.py`, `tests/airflow/test_plugins/test_domain_terms.py`.

**Residual:** HITL review UI thresholding (Green/Yellow/Red per SYS-FR-32) — owned by UPA; DE exposes the scalar field, no GET endpoint introduced.

---

### GAP-012 — SeaweedFS tenant/node object-key prefix — ✅ Resolved

**Summary:** All DE SeaweedFS writes use the exact `{tenant_id}/{node_id}/{document_id}/{filename}` format required by SYS-FR-56.

**Evidence:**
- `airflow/plugins/api/upload.py:146` — `s3_key = f"{tenant_id}/{node_id}/{document_id}/{safe_filename}"`.
- `airflow/plugins/api/upload.py:149` — `validate_s3_path(tenant_id, node_id, s3_key)` rejects any mismatched path (raises `PermissionError` → HTTP 403).
- `airflow/plugins/api/upload.py:224` — `_delete_s3_folder` scopes cleanup to the same tenant/node/doc prefix across both `source-files` and `textiq-images` buckets.

---

### GAP-018 — Core tables missing tenant_id/node_id + RLS — ✅ Resolved

**Summary:** The four core DE tables plus the LlamaIndex hierarchical-nodes table now carry `tenant_id`/`node_id` (NOT NULL) and are protected by RLS policies keyed on `app.tenant_id`.

**Evidence:**
- `src/db/migrations/versions/20260313_100000_add_tenant_node_columns.py` — adds nullable `tenant_id`/`node_id` + composite indexes on `documents`, `extracted_records`, `domain_term_dictionary`, `symbol_dictionaries`.
- `src/db/migrations/versions/20260313_100001_backfill_tenant_node.py` — backfills data.
- `src/db/migrations/versions/20260313_100002_enforce_tenant_node_not_null_rls.py` — sets NOT NULL, `ENABLE`+`FORCE ROW LEVEL SECURITY`, and creates `tenant_isolation_{table}` policies for all four tables.
- `src/db/migrations/versions/20260330_100000_rls_hierarchical_nodes.py` — RLS on `data_hierarchical_nodes`.
- `airflow/plugins/api/rls.py:28` — every request uses `SET LOCAL app.tenant_id = :tid` inside a transactional `rls_connection()`.
- `airflow/plugins/api/upload.py` — every INSERT/SELECT/UPDATE/DELETE runs under `rls_connection(tenant_id)` or sets `app.tenant_id` directly.
- Milvus filters include `tenant_id` and `node_id`: `airflow/plugins/api/milvus.py:97-101` (`delete_chunks_by_document_id`), `:119-121` (`query_chunks_with_term`).

**Residual:** No current gap in airflow/; ensure new endpoints follow the same `rls_connection()` pattern.

---

### GAP-024 — Milvus collection naming — ✅ Resolved (DE scope)

**Summary:** DE code defaults now align on `textiq_records`, matching `src/core/config.py`, `.env.example`, and architecture doc §10.1.

**Evidence:**
- `airflow/plugins/api/milvus.py:20` — `os.getenv("MILVUS_COLLECTION", "textiq_records")`.
- `src/core/config.py:18` — `milvus_collection: str = "textiq_records"`.
- `.env.example:60` — `MILVUS_COLLECTION=textiq_records`.
- `docs/airflow-setup.md:283` — `MILVUS_COLLECTION=textiq_records`.

**Residual (out of DE repo scope):**
- Chatbot-service env default alignment (separate repo).
- Operational migration from any deployed `textiq_records_new` / `dsol_records` collection to `textiq_records` (ops runbook, not a code change).

---

### GAP-027 — Socket.IO pipeline events — ✅ Resolved (TEXTIQ-390)

**Summary:** DE publishes stage-transition events to Redis pub/sub on per-tenant+node channels; UPA subscribes and relays over Socket.IO `/pipeline`. Implementation tracked in [spec-textiq-390-redis-pubsub-pipeline-events.md](./spec-textiq-390-redis-pubsub-pipeline-events.md).

**Resolution:**
- New module `src/services/pipeline_events.py` — sync publisher, flag-gated by `PIPELINE_EVENTS_ENABLED`, swallows `redis.RedisError`.
- Emits at DAG stage boundaries: `pending`, `processing`, `parsing`, `chunking`, `indexing`, `completed`|`failed`.
- Channel format `pipeline:{tenant_id}:{node_id}`; payload includes `correlation_id` (= `dag_run_id`), `document_id`, `tenant_id`, `node_id`, `stage`, `status`, `timestamp`, optional `error`.
- Also fixed a latent bug: `update_document_status_task` now resolves terminal outcome from upstream TI states and writes `documents.status='failed'` when any upstream task failed (previously wrote `completed` unconditionally).
- Also tightened retry policy: `default_args.retries=0`; only the five format parsers retain `retries=2` with 30s exponential backoff.

**Architecture doc:** `docs/architecture.md` — new `## Pipeline Events (DE → UPA)` section.

---

### GAP-029 — Prometheus metrics — ❌ Not resolved

**Summary:** No operational Prometheus metrics in the Airflow plugin.

**Evidence:**
- No `prometheus_client` import anywhere in `airflow/`.
- No `/metrics` endpoint on the upload/tags/domain-terms FastAPI sub-apps.

**Action remaining:** Minimum viable — add `prometheus_client`, expose `/metrics` on each sub-app (or a shared router), emit counters for upload/approve/reject and histograms for DAG task latency via Airflow's existing StatsD.

---

### GAP-030 — X-Correlation-ID / X-Request-ID — ✅ Resolved (plugin scope)

**Summary:** The DE Airflow plugin generates/forwards both dual correlation headers, echoes them in responses, and propagates them to downstream systems.

**Evidence:**
- `airflow/plugins/api/middleware.py:30-39` — `RequestIdMiddleware` accepts or generates `X-Request-ID`, stores on `request.state`, echoes in response.
- `airflow/plugins/api/middleware.py:42-60` — `CorrelationIdMiddleware` forwards/generates `X-Correlation-ID`, logs a warning on fallback, echoes in response.
- `airflow/plugins/api/upload.py:382-383` — correlation/request IDs are injected into `trigger_dag` `conf` so Airflow task logs stay correlatable across the pipeline.

**Residual:** Propagation into external_python task logging context (if not already) is a good follow-up.

---

### GAP-031 — Circuit breaker — ❌ Not resolved

**Summary:** No per-upstream circuit breaker on DE's outbound calls (Milvus, SeaweedFS).

**Evidence:**
- `grep -ri circuit airflow/` → only `middleware.py:64` (`Short-circuits with 401/500 on failure.` docstring). No library, no state machine.

**Action remaining:** Either adopt `pybreaker` locally or consume the shared `textiq-client-sdk` circuit breaker once available.

---

### GAP-033 — Milvus tenant/node isolation (Strategy C) — ✅ Resolved

**Summary:** Single-collection strategy with scalar-field filtering is implemented in code, matching the architecture decision.

**Evidence:**
- `airflow/plugins/api/milvus.py:85-106` — `delete_chunks_by_document_id` builds `document_id == … AND tenant_id == … AND node_id == …` filter.
- `airflow/plugins/api/milvus.py:109-123` — `query_chunks_with_term` filters by `tenant_id` in addition to the term predicate.
- DAG store tasks pass `tenant_id` and `node_id` through to Milvus writes (e.g., `document_extraction_dag.py:345-349, 395-401, 461-465`).

---

### GAP-038 — APA→DE multipart upload (CQRS) — ✅ Resolved

**Summary:** Target architecture is in place: Client → APA/BFF (multipart) → DE `POST /api/v1/documents/upload` (multipart) → DE writes SeaweedFS → DE triggers Airflow with object-key string.

**Evidence:**
- `airflow/plugins/api/upload.py:56` — `upload_app = FastAPI(title="TextIQ Upload API")`, mounted via plugin at `/api/v1/documents`.
- `airflow/plugins/api/upload.py:81-207` — `POST /upload` accepts `UploadFile`, validates (extension/magic bytes/password/filename sanitation), computes canonical S3 key, writes to SeaweedFS, inserts `documents` row (status `pending`, approval `pending`).
- `airflow/plugins/api/upload.py:314-398` — `POST /{document_id}/approve` gates DAG trigger; `trigger_dag(DAG_ID, conf={..., object_key: file_path})` passes the URL string (not the file) to Airflow — satisfying the Airflow `dag_run.conf` constraint.
- HMAC verification (`HMACVerificationMiddleware`) ensures only APA can call this endpoint.

**Residual:** Monitor latency/error behavior under batch uploads; ensure DE container resources sized for multipart streaming.

---

### GAP-039 — LLM provider abstraction — ❌ Not resolved

**Summary:** Azure OpenAI / `text-embedding-3-large` remains hardcoded.

**Evidence:**
- `airflow/dags/tasks/processing_tasks.py:293` — `encoding = tiktoken.encoding_for_model("text-embedding-3-large")`.
- `airflow/dags/tasks/processing_tasks.py:445` — comment "Initialize tiktoken encoder for text-embedding-3-large".
- `airflow/dags/document_extraction_dag.py:241` — DAG doc string still references `text-embedding-3-large` + `3072-dim`.

**Action remaining:** Adopt the `LLMProvider` interface per `specs/llm-provider-abstraction-spec.md`; expose `EMBEDDING_MODEL` + `EMBEDDING_DIM` env vars; add an embedding-migration DAG.

---

### GAP-040 — Hierarchical chunk sizes (4096/1024/256) — ✅ Resolved

**Summary:** Canonical sizes are now consistent in both runtime code and docstrings.

**Evidence (code):**
- `airflow/dags/tasks/processing_tasks.py:418` — `chunk_sizes = [4096, 1024, 256]` (text pipeline).
- `airflow/dags/tasks/processing_tasks.py:2396` — `chunk_sizes = [4096, 1024, 256]` (Excel Detailed pipeline).

**Evidence (docstrings, now aligned):**
- `airflow/dags/document_extraction_dag.py:12` — "3 levels [4096/1024/256 tokens]".
- `airflow/dags/document_extraction_dag.py:214` — "3-level hierarchy: [4096, 1024, 256] tokens".
- `airflow/dags/tasks/processing_tasks.py:394` — "default: 4096/1024/256 tokens".
- `airflow/dags/tasks/processing_tasks.py:400` — "default: [4096, 1024, 256]".
- `airflow/dags/tasks/processing_tasks.py:2379` — "default: [4096, 1024, 256]".

---

### GAP-041 — HITL tag cross-service flow — 🟡 Partially addressed

**Summary:** DE plugin exposes the tag-lifecycle APIs with history tracking, but the end-to-end DE↔APA↔UP contract is still not captured in a shared spec.

**Evidence:**
- `airflow/plugins/api/tags.py` — `PATCH /{chunk_id}/tags/status` (single), `PATCH /{chunk_id}/tags/bulk-status`, `PATCH /{chunk_id}/tags/{tag_id}/name`, `DELETE /{chunk_id}/tags/{tag_id}`; each appends a history entry (`action`, `by`, `ts`, optional `old_name`).
- Tag status is constrained to `verified | rejected` (Pydantic `Literal` at `tags.py:50`).
- Updates cascade to Milvus via `upsert_chunk_field(chunk_id, tags=…)`.

**Action remaining:** Produce a cross-service spec under `_planning/docs/planning-artifacts/2-planning/specs/hitl-tag-integration.md` locking down the shared tag schema, confidence thresholds, and state machine.

---

### GAP-042 — Domain term dictionary cross-service sync — 🟡 Partially addressed

**Summary:** DE owns CRUD + cascade of domain-term changes to both PG chunks and Milvus, but there is still no defined push mechanism to DG.

**Evidence:**
- `airflow/plugins/api/domain_terms.py` — `POST /`, `POST /bulk`, `PATCH /{term_id}`, `DELETE /{term_id}`.
- Cascade on rename: updates `data_hierarchical_nodes` JSONB metadata and batch upserts affected Milvus chunks (`domain_terms.py:358-384`).
- Cascade on delete: removes the term from JSONB metadata and Milvus chunks (`domain_terms.py:280-303`).

**Action remaining:** Choose sync mechanism (webhook to DG or shared Redis channel), document the expected propagation latency, and update the integration map.

---

### GAP-043 — Chunk schema versioning — ❌ Not resolved

**Summary:** Hierarchical chunk payloads carry no version field, so consumers cannot detect or handle schema changes.

**Evidence:**
- No `schema_version` / `chunk_schema_version` field in `airflow/dags/tasks/processing_tasks.py` chunk builders, in Milvus collection schema, or in `data_hierarchical_nodes.value`.

**Action remaining:** Add a `schema_version` field (e.g. in chunk metadata and on the Milvus record), document backward-compatibility rules, and publish a version contract (DE producer ↔ CB/DG consumers).

---

## Quick-win Backlog

Suggested order for the next sprint, highest value / lowest cost first:

1. ~~**GAP-040 docstring cleanup**~~ — done (TEXTIQ-387).
2. ~~**GAP-024 Milvus collection name**~~ — DE default aligned (TEXTIQ-388). CB + ops migration tracked separately.
3. ~~**GAP-011 domain-term confidence**~~ — done (TEXTIQ-389). HITL threshold UI remains UPA-side.
4. **GAP-043 chunk `schema_version`** — add constant + metadata field; document compat rules.
5. **GAP-029 Prometheus** — expose `/metrics` on the plugin sub-apps.
6. **GAP-027 Socket.IO** — Redis pub/sub emitter on DAG stage transitions.
7. **GAP-031 circuit breaker** and **GAP-039 LLM abstraction** — track with owning architecture specs.

## Reference

- Source gaps: `_planning/docs/planning-artifacts/2-planning/system-gaps-and-changes.md`
- Isolation spec: `_planning/docs/planning-artifacts/2-planning/specs/data-isolation-spec.md`
- Tenant/node tech spec: `_bmad-output/implementation-artifacts/tech-spec-de-tenant-node-isolation.md`
- Airflow plugin entry points:
  - `airflow/plugins/api/upload.py`
  - `airflow/plugins/api/tags.py`
  - `airflow/plugins/api/domain_terms.py`
  - `airflow/plugins/api/milvus.py`
  - `airflow/plugins/api/rls.py`
  - `airflow/plugins/api/middleware.py`
