--- 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`