# Technical Specification: Airflow Migration for Document Extraction Pipeline

**Version:** 1.0
**Status:** Draft - Ready for Review
**Created:** 2025-12-08
**Epic:** Document Extraction Pipeline Modernization
**Story:** Migrate from Celery to Apache Airflow for DAG-based Orchestration

---

## Executive Summary

This specification outlines the migration plan to refactor the current Celery-based document extraction pipeline (`src/extraction_v2/`) to use Apache Airflow as the orchestration backbone, following the architecture defined in `document-extraction-engine.md`.

**Current State:**
- Celery-based task processing with `extract_document_v2` task
- Monolithic `ExtractionPipelineV2` class orchestrating all extraction logic
- Tight coupling between orchestration and business logic
- Limited observability and workflow visualization

**Target State:**
- Apache Airflow 3.0.6 DAG-based orchestration
- Decoupled operators for each processing step
- Event-driven triggers via SeaweedFS webhooks
- Enhanced observability with DAG visualization and task monitoring

**Key Benefits:**
1. Visual workflow management and monitoring
2. Built-in retry, backfill, and scheduling capabilities
3. Task-level parallelism and dependency management
4. Better separation of concerns between orchestration and processing
5. Foundation for multi-document source support (beyond SeaweedFS)

---

## Table of Contents

1. [Current Architecture Analysis](#current-architecture-analysis)
2. [Target Architecture](#target-architecture)
3. [Component Mapping](#component-mapping)
4. [Migration Strategy](#migration-strategy)
5. [Implementation Plan](#implementation-plan)
6. [Data Models](#data-models)
7. [Operator Specifications](#operator-specifications)
8. [Configuration & Secrets](#configuration--secrets)
9. [Testing Strategy](#testing-strategy)
10. [Rollback Plan](#rollback-plan)
11. [Risk Assessment](#risk-assessment)

---

## Current Architecture Analysis

### Existing Components

```
┌─────────────────────────────────────────────────────────────────────────────┐
│                    CURRENT EXTRACTION PIPELINE V2                            │
└─────────────────────────────────────────────────────────────────────────────┘

                    Document Upload (API)
                           │
                           ▼
              ┌────────────────────────┐
              │   Celery Task Queue    │
              │  (extract_document_v2) │
              └────────────────────────┘
                           │
                           ▼
              ┌────────────────────────┐
              │  ExtractionPipelineV2  │
              │   (Orchestrator)       │
              └────────────────────────┘
                           │
    ┌──────────────────────┼──────────────────────┐
    │                      │                      │
    ▼                      ▼                      ▼
┌──────────┐        ┌──────────────┐       ┌────────────────┐
│  Excel   │        │  Word/PDF    │       │  PowerPoint    │
│ Pipeline │        │  Pipeline    │       │  Pipeline      │
└──────────┘        └──────────────┘       └────────────────┘
    │                      │                      │
    ▼                      ▼                      ▼
┌────────────────────────────────────────────────────────────┐
│              Common Processing Steps                        │
│  1. DocumentToHtmlConverter (format-specific conversion)    │
│  2. HtmlTableExtractor / SemanticChunker                   │
│  3. LlmHeaderDetector (Excel only)                         │
│  4. RecordBuilder                                          │
│  5. _save_v2_records (PostgreSQL)                          │
│  6. _index_vectors_sync (Milvus)                           │
└────────────────────────────────────────────────────────────┘
```

### Source File Inventory

| File | Purpose | Lines | Airflow Mapping |
|------|---------|-------|-----------------|
| `pipeline.py` | Orchestrator class | 1139 | DAG definition + task routing |
| `document_converter.py` | Multi-format conversion | 1818 | ParseDocumentOperator |
| `html_table_extractor.py` | HTML → Tables | 370 | ChunkContentOperator (Excel) |
| `llm_header_detector.py` | GPT-4o header detection | 517 | GenerateEmbeddingsOperator (partial) |
| `record_builder.py` | JSON record construction | 488 | StoreChunksOperator (partial) |
| `semantic_chunker.py` | HybridChunker for text docs | 294 | ChunkContentOperator (Word/PDF) |
| `large_table_extractor.py` | Chunked large table processing | ~600 | ParseDocumentOperator (Excel, integrated) |
| `detect_border.py` | Border detection preprocessing | ~400 | ParseDocumentOperator (Excel) |
| `word_preprocessor.py` | Word preprocessing | ~300 | ParseDocumentOperator (Word) |
| `pptx_preprocessor.py` | PowerPoint preprocessing | ~200 | ParseDocumentOperator (PowerPoint) |
| `libreoffice_lock.py` | Concurrency control | ~50 | Shared utility |
| `extraction_v2_tasks.py` | Celery task wrapper | 706 | **REPLACED by DAG** |

### Current Data Flow

1. **API Upload** → Document stored in filesystem, metadata in PostgreSQL
2. **Celery Task** → `extract_document_v2(document_id, file_path)` queued
3. **File Type Routing** → Excel/Word/PDF/PowerPoint pipeline selected
4. **Processing** → Format-specific conversion + chunking
5. **Storage** → Records saved to PostgreSQL via `_save_v2_records()`
6. **Vector Indexing** → Embeddings generated and stored in Milvus

### Pain Points Addressed by Migration

| Pain Point | Current Issue | Airflow Solution |
|------------|---------------|------------------|
| **Monolithic Task** | Single 700+ line task handles all logic | Separate operators per step |
| **Limited Retry Granularity** | Entire extraction retries on failure | Task-level retries |
| **No Workflow Visibility** | Black box processing | DAG visualization |
| **Hard-coded Routing** | File type routing in task code | DAG branching operators |
| **Manual Progress Tracking** | Redis-based progress updates | Built-in task state |
| **No Backfill Support** | Cannot reprocess historical data easily | Native backfill capabilities |

---

## Target Architecture

### Airflow DAG Structure

```
┌─────────────────────────────────────────────────────────────────────────────┐
│                    DOCUMENT EXTRACTION DAG (Airflow 3.x)                     │
└─────────────────────────────────────────────────────────────────────────────┘

                           SeaweedFS Webhook / API Trigger
                                     │
                                     ▼
                          ┌────────────────────┐
                          │  validate_event    │
                          │  (ValidateEventOp) │
                          └────────────────────┘
                                     │
                                     ▼
                          ┌────────────────────┐
                          │  fetch_document    │
                          │  (FetchDocumentOp) │
                          └────────────────────┘
                                     │
                                     ▼
                          ┌────────────────────┐
                          │  detect_format     │
                          │  (BranchOp)        │
                          └────────────────────┘
                                     │
           ┌─────────────────────────┼─────────────────────────┐
           │                         │                         │
           ▼                         ▼                         ▼
    ┌────────────┐           ┌────────────┐           ┌────────────┐
    │   Excel    │           │ Word/PDF   │           │ PowerPoint │
    │  Pipeline  │           │  Pipeline  │           │  Pipeline  │
    └────────────┘           └────────────┘           └────────────┘
           │                         │                         │
           ▼                         ▼                         ▼
    ┌────────────┐           ┌────────────┐           ┌────────────┐
    │ parse_doc  │           │ parse_doc  │           │ parse_doc  │
    │ (ParseOp)  │           │ (ParseOp)  │           │ (ParseOp)  │
    └────────────┘           └────────────┘           └────────────┘
           │                         │                         │
           ▼                         │                         │
    ┌────────────┐                   │                         │
    │ detect_    │                   │                         │
    │ headers    │                   │                         │
    │ (LLMOp)    │                   │                         │
    └────────────┘                   │                         │
           │                         │                         │
           ▼                         ▼                         ▼
    ┌────────────┐           ┌────────────┐           ┌────────────┐
    │ chunk_     │           │ chunk_     │           │ chunk_     │
    │ content    │           │ content    │           │ content    │
    └────────────┘           └────────────┘           └────────────┘
           │                         │                         │
           └─────────────────────────┼─────────────────────────┘
                                     │
                                     ▼
                          ┌────────────────────┐
                          │  generate_embed    │
                          │  (EmbeddingsOp)    │
                          └────────────────────┘
                                     │
                      ┌──────────────┴──────────────┐
                      ▼                              ▼
              ┌────────────┐                 ┌────────────┐
              │ store_     │                 │ store_     │
              │ chunks     │                 │ vectors    │
              └────────────┘                 └────────────┘
                      │                              │
                      ▼                              ▼
              ┌────────────┐                 ┌────────────┐
              │ PostgreSQL │                 │  Milvus    │
              └────────────┘                 └────────────┘
```

### Container Deployment

```yaml
# docker-compose.airflow.yml additions
# Requires: SeaweedFS 4.02+ for S3-compatible object storage
services:
  airflow-webserver:
    image: apache/airflow:3.1.3-python3.11
    environment:
      - AIRFLOW__CORE__EXECUTOR=LocalExecutor
      - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres:5432/airflow
    ports:
      - "8081:8080"
    volumes:
      - ./dags:/opt/airflow/dags
      - ./plugins:/opt/airflow/plugins
    depends_on:
      - postgres
      - redis

  airflow-scheduler:
    image: apache/airflow:3.1.3-python3.11
    command: scheduler
    volumes:
      - ./dags:/opt/airflow/dags
      - ./plugins:/opt/airflow/plugins
```

---

## Component Mapping

### Current → Airflow Operator Mapping

| Current Component | Current Location | Target Operator | Notes |
|-------------------|------------------|-----------------|-------|
| File type routing | `extraction_v2_tasks.py:76-100` | `BranchPythonOperator` | Branch based on extension |
| `DocumentToHtmlConverter` | `document_converter.py` | `ParseDocumentOperator` | Reuse existing class |
| `HtmlTableExtractor` | `html_table_extractor.py` | `ChunkContentOperator` | Excel path |
| `SemanticChunker` | `semantic_chunker.py` | `ChunkContentOperator` | Word/PDF/PowerPoint path |
| `LlmHeaderDetector` | `llm_header_detector.py` | `DetectHeadersOperator` | Excel-only operator |
| `RecordBuilder` | `record_builder.py` | `StoreChunksOperator` | Build + store combined |
| `_save_v2_records()` | `extraction_v2_tasks.py:331-437` | `StoreChunksOperator` | PostgreSQL persistence |
| `_index_vectors_sync()` | `extraction_v2_tasks.py:670-705` | `StoreVectorsOperator` | Milvus indexing |
| Progress tracking | `_update_progress()` | Built-in task state | Airflow native |

### Reusable Components (No Changes Required)

These components encapsulate business logic and can be imported directly by operators:

```python
# Components to import directly in operators
from src.extraction_v2.document_converter import DocumentToHtmlConverter
from src.extraction_v2.html_table_extractor import HtmlTableExtractor
from src.extraction_v2.llm_header_detector import LlmHeaderDetector
from src.extraction_v2.record_builder import RecordBuilder
from src.extraction_v2.semantic_chunker import SemanticChunker
from src.extraction_v2.large_table_extractor import LargeTableExtractor
from src.extraction_v2.detect_border import BorderTableDetector
```

### Components to Deprecate

| Component | File | Replacement |
|-----------|------|-------------|
| `extract_document_v2` task | `extraction_v2_tasks.py` | `document_extraction_dag` |
| `_fallback_to_v1()` | `extraction_v2_tasks.py:224-308` | Separate fallback DAG task |
| `ExtractionPipelineV2` orchestration | `pipeline.py:100-1013` | DAG task dependencies |

---

## Migration Strategy

### Phase 1: Parallel Deployment (2 sprints)

**Goal:** Deploy Airflow alongside existing Celery without disrupting current operations.

```
┌─────────────────────────────────────────────────────────────────────────────┐
│                         PHASE 1: PARALLEL DEPLOYMENT                         │
└─────────────────────────────────────────────────────────────────────────────┘

                          Document Upload API
                                  │
                  ┌───────────────┴───────────────┐
                  │                               │
                  ▼                               ▼
        ┌──────────────────┐           ┌──────────────────┐
        │  Celery Worker   │           │   Airflow DAG    │
        │  (CURRENT)       │           │   (NEW - Beta)   │
        │                  │           │                  │
        │  100% traffic    │           │  Shadow mode     │
        │  ✅ Production   │           │  📊 Metrics only │
        └──────────────────┘           └──────────────────┘
```

**Tasks:**
1. Deploy Airflow infrastructure (webserver, scheduler, DB)
2. Create `document_extraction_dag` with all operators
3. Implement shadow mode: Airflow processes copies of documents without saving
4. Compare results between Celery and Airflow outputs
5. Monitor performance and accuracy metrics

### Phase 2: Gradual Traffic Migration (2 sprints)

**Goal:** Progressively route traffic from Celery to Airflow.

```
Week 1:  Celery 100% | Airflow 0%   (baseline)
Week 2:  Celery 90%  | Airflow 10%  (canary)
Week 3:  Celery 70%  | Airflow 30%  (expanded canary)
Week 4:  Celery 50%  | Airflow 50%  (balanced)
Week 5:  Celery 20%  | Airflow 80%  (migration)
Week 6:  Celery 0%   | Airflow 100% (complete)
```

**Traffic Splitting Implementation:**
```python
# src/api/routes/documents.py
@router.post("/upload")
async def upload_document(...):
    # Feature flag for gradual rollout
    use_airflow = await feature_flags.check("airflow_extraction",
                                            percentage=settings.airflow_traffic_percentage)

    if use_airflow:
        # Trigger Airflow DAG via REST API
        airflow_client.trigger_dag("document_extraction_dag", conf={...})
    else:
        # Existing Celery task
        extract_document_v2.delay(document_id, file_path)
```

### Phase 3: Celery Deprecation (1 sprint)

**Goal:** Remove Celery extraction task and related code.

**Tasks:**
1. Remove `extract_document_v2` Celery task
2. Archive `extraction_v2_tasks.py`
3. Update API to use Airflow exclusively
4. Remove Celery beat schedules for extraction
5. Update documentation

---

## Implementation Plan

### Project Structure

```
textiq-doc-extraction/
├── dags/
│   ├── __init__.py
│   ├── document_extraction_dag.py        # Main DAG definition
│   ├── large_table_extraction_dag.py     # Large table handling DAG
│   └── config.py                         # DAG configuration
├── plugins/
│   ├── operators/
│   │   ├── __init__.py
│   │   ├── validate_event.py             # ValidateEventOperator
│   │   ├── fetch_document.py             # FetchDocumentOperator
│   │   ├── parse_document.py             # ParseDocumentOperator
│   │   ├── detect_headers.py             # DetectHeadersOperator
│   │   ├── chunk_content.py              # ChunkContentOperator
│   │   ├── generate_embeddings.py        # GenerateEmbeddingsOperator
│   │   ├── store_chunks.py               # StoreChunksOperator
│   │   └── store_vectors.py              # StoreVectorsOperator
│   └── hooks/
│       ├── __init__.py
│       ├── seaweedfs_hook.py             # SeaweedFS connection hook
│       └── milvus_hook.py                # Milvus connection hook
├── src/
│   └── extraction_v2/                    # Existing business logic (UNCHANGED)
│       ├── document_converter.py
│       ├── html_table_extractor.py
│       ├── llm_header_detector.py
│       ├── record_builder.py
│       ├── semantic_chunker.py
│       └── ...
└── docker/
    └── docker-compose.airflow.yml        # Airflow services
```

### DAG Definition

```python
# dags/document_extraction_dag.py
from datetime import timedelta
from airflow import DAG
from airflow.operators.python import BranchPythonOperator
from airflow.utils.dates import days_ago

from plugins.operators.validate_event import ValidateEventOperator
from plugins.operators.fetch_document import FetchDocumentOperator
from plugins.operators.parse_document import ParseDocumentOperator
from plugins.operators.detect_headers import DetectHeadersOperator
from plugins.operators.chunk_content import ChunkContentOperator
from plugins.operators.generate_embeddings import GenerateEmbeddingsOperator
from plugins.operators.store_chunks import StoreChunksOperator
from plugins.operators.store_vectors import StoreVectorsOperator

default_args = {
    'owner': 'textiq',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=2),
    'retry_exponential_backoff': True,
    'max_retry_delay': timedelta(minutes=30),
}

with DAG(
    dag_id='document_extraction_dag',
    default_args=default_args,
    description='Multi-format document extraction pipeline',
    schedule_interval=None,  # Event-triggered only
    start_date=days_ago(1),
    catchup=False,
    max_active_runs=10,
    tags=['extraction', 'v2', 'docling'],
) as dag:

    # Task 1: Validate incoming event
    validate = ValidateEventOperator(
        task_id='validate_event',
    )

    # Task 2: Fetch document from storage
    fetch = FetchDocumentOperator(
        task_id='fetch_document',
        connection_id='seaweedfs_default',
    )

    # Task 3: Detect format and branch
    def detect_format(**context):
        """Branch based on file extension."""
        event = context['ti'].xcom_pull(task_ids='validate_event')
        file_path = event.get('file_path', '')
        ext = file_path.lower().split('.')[-1]

        if ext in ['xlsx', 'xls', 'xlsm']:
            return 'parse_excel'
        elif ext in ['docx', 'doc', 'docm']:
            return 'parse_word'
        elif ext == 'pdf':
            return 'parse_pdf'
        elif ext in ['pptx', 'ppt', 'pptm']:
            return 'parse_powerpoint'
        else:
            return 'unsupported_format'

    branch = BranchPythonOperator(
        task_id='detect_format',
        python_callable=detect_format,
    )

    # Task 4a: Parse Excel
    parse_excel = ParseDocumentOperator(
        task_id='parse_excel',
        doc_type='excel',
    )

    # Task 4b: Parse Word
    parse_word = ParseDocumentOperator(
        task_id='parse_word',
        doc_type='word',
    )

    # Task 4c: Parse PDF
    parse_pdf = ParseDocumentOperator(
        task_id='parse_pdf',
        doc_type='pdf',
    )

    # Task 4d: Parse PowerPoint
    parse_powerpoint = ParseDocumentOperator(
        task_id='parse_powerpoint',
        doc_type='powerpoint',
    )

    # Task 5: Detect headers (Excel only)
    detect_headers = DetectHeadersOperator(
        task_id='detect_headers',
        trigger_rule='none_failed_min_one_success',
    )

    # Task 6: Chunk content
    chunk_excel = ChunkContentOperator(
        task_id='chunk_excel',
        chunking_strategy='table',
    )

    chunk_text = ChunkContentOperator(
        task_id='chunk_text',
        chunking_strategy='semantic',
        trigger_rule='none_failed_min_one_success',
    )

    # Task 7: Generate embeddings
    embeddings = GenerateEmbeddingsOperator(
        task_id='generate_embeddings',
        connection_id='azure_openai_default',
        batch_size=16,
        trigger_rule='none_failed_min_one_success',
    )

    # Task 8a: Store chunks in PostgreSQL
    store_chunks = StoreChunksOperator(
        task_id='store_chunks',
        connection_id='postgres_default',
    )

    # Task 8b: Store vectors in Milvus
    store_vectors = StoreVectorsOperator(
        task_id='store_vectors',
        connection_id='milvus_default',
        collection='document_embeddings',
    )

    # Define dependencies
    validate >> fetch >> branch

    # Excel path
    branch >> parse_excel >> detect_headers >> chunk_excel >> embeddings

    # Word/PDF/PowerPoint paths (no header detection)
    branch >> parse_word >> chunk_text >> embeddings
    branch >> parse_pdf >> chunk_text
    branch >> parse_powerpoint >> chunk_text

    # Final storage (parallel)
    embeddings >> [store_chunks, store_vectors]
```

---

## Data Models

### XCom Data Structures

```python
# plugins/models/xcom_schemas.py
from dataclasses import dataclass
from typing import Any, Optional
from uuid import UUID

@dataclass
class ValidatedEvent:
    """Output from ValidateEventOperator."""
    document_id: UUID
    file_path: str
    content_type: str
    size_bytes: int
    event_id: str
    bucket: str
    object_key: str

@dataclass
class FetchedDocument:
    """Output from FetchDocumentOperator."""
    local_path: str
    content_type: str
    original_filename: str
    file_extension: str

@dataclass
class ParsedDocument:
    """Output from ParseDocumentOperator."""
    doc_type: str  # 'excel', 'word', 'pdf', 'powerpoint'
    html_content: Optional[str]  # Excel → HTML
    markdown_content: Optional[str]  # Word/PDF/PowerPoint → Markdown
    tables: list[dict]  # Extracted tables (Excel)
    docling_document: Any  # Serialized DoclingDocument for chunking

@dataclass
class DetectedHeaders:
    """Output from DetectHeadersOperator."""
    header_rows: list[int]
    headers: list[dict]  # {column: int, hierarchy: list[str]}

@dataclass
class ContentChunks:
    """Output from ChunkContentOperator."""
    chunks: list[dict]  # {content: dict, _source: dict} or {text: str, metadata: dict}
    chunk_count: int
    chunking_strategy: str

@dataclass
class ChunkEmbeddings:
    """Output from GenerateEmbeddingsOperator."""
    embeddings: list[dict]  # {chunk_id: str, vector: list[float]}
    embedding_count: int
    model: str
    cached_count: int
```

### Database Schema Additions

```sql
-- Add extraction_jobs table for Airflow audit trail
CREATE TABLE IF NOT EXISTS extraction_jobs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    dag_run_id VARCHAR(255) NOT NULL,
    document_id UUID REFERENCES documents(id),
    status VARCHAR(50) NOT NULL DEFAULT 'pending',
    pipeline_version VARCHAR(10) DEFAULT 'v2',
    chunks_created INTEGER DEFAULT 0,
    vectors_created INTEGER DEFAULT 0,
    error_message TEXT,
    started_at TIMESTAMP DEFAULT NOW(),
    completed_at TIMESTAMP,
    metadata JSONB,
    UNIQUE(dag_run_id)
);

CREATE INDEX idx_extraction_jobs_document ON extraction_jobs(document_id);
CREATE INDEX idx_extraction_jobs_status ON extraction_jobs(status);
CREATE INDEX idx_extraction_jobs_dag_run ON extraction_jobs(dag_run_id);
```

---

## Operator Specifications

### ValidateEventOperator

```python
# plugins/operators/validate_event.py
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class ValidateEventOperator(BaseOperator):
    """
    Validate incoming webhook/API event and extract metadata.

    Input: dag_run.conf with event payload
    Output: ValidatedEvent XCom
    """

    ALLOWED_CONTENT_TYPES = {
        'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',  # xlsx
        'application/vnd.ms-excel',  # xls
        'application/vnd.openxmlformats-officedocument.wordprocessingml.document',  # docx
        'application/msword',  # doc
        'application/pdf',
        'application/vnd.openxmlformats-officedocument.presentationml.presentation',  # pptx
    }

    MAX_FILE_SIZE = 100 * 1024 * 1024  # 100MB

    @apply_defaults
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def execute(self, context):
        conf = context['dag_run'].conf or {}

        # Extract from SeaweedFS webhook format or direct API call
        if 'Records' in conf:
            # SeaweedFS webhook format
            record = conf['Records'][0]['s3']
            event = {
                'bucket': record['bucket']['name'],
                'object_key': record['object']['key'],
                'content_type': record['object'].get('contentType'),
                'size_bytes': record['object'].get('size', 0),
                'document_id': conf.get('document_id'),
            }
        else:
            # Direct API trigger format
            event = conf

        # Validate
        if event.get('content_type') not in self.ALLOWED_CONTENT_TYPES:
            raise ValueError(f"Unsupported content type: {event.get('content_type')}")

        if event.get('size_bytes', 0) > self.MAX_FILE_SIZE:
            raise ValueError(f"File too large: {event.get('size_bytes')} bytes")

        return event
```

### ParseDocumentOperator

```python
# plugins/operators/parse_document.py
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class ParseDocumentOperator(BaseOperator):
    """
    Parse document using Docling and existing converters.

    Reuses:
    - src.extraction_v2.document_converter.DocumentToHtmlConverter
    - src.extraction_v2.html_table_extractor.HtmlTableExtractor
    """

    @apply_defaults
    def __init__(self, doc_type: str, **kwargs):
        super().__init__(**kwargs)
        self.doc_type = doc_type

    def execute(self, context):
        from src.extraction_v2.document_converter import DocumentToHtmlConverter
        from src.extraction_v2.html_table_extractor import HtmlTableExtractor

        # Get document path from previous task
        fetch_result = context['ti'].xcom_pull(task_ids='fetch_document')
        local_path = fetch_result['local_path']

        converter = DocumentToHtmlConverter()

        if self.doc_type == 'excel':
            # Excel → HTML conversion
            html = converter.convert_to_html(local_path)
            extractor = HtmlTableExtractor()
            valid_tables, raw_text_tables = extractor.extract_all_tables(html)

            return {
                'doc_type': 'excel',
                'html_content': html,
                'tables': [{'html': t.html, 'rows': t.rows} for t in valid_tables],
                'raw_text_tables': [{'rows': t.rows} for t in raw_text_tables],
            }
        else:
            # Word/PDF/PowerPoint → Markdown
            markdown = converter.convert_to_html(local_path)  # Returns markdown for these types

            return {
                'doc_type': self.doc_type,
                'markdown_content': markdown,
                'local_path': local_path,  # Needed for DoclingDocument in chunking
            }
```

### ChunkContentOperator

```python
# plugins/operators/chunk_content.py
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults

class ChunkContentOperator(BaseOperator):
    """
    Chunk content using appropriate strategy.

    Reuses:
    - src.extraction_v2.record_builder.RecordBuilder (Excel)
    - src.extraction_v2.semantic_chunker.SemanticChunker (Word/PDF/PowerPoint)
    """

    @apply_defaults
    def __init__(self, chunking_strategy: str, max_tokens: int = 8000, **kwargs):
        super().__init__(**kwargs)
        self.chunking_strategy = chunking_strategy
        self.max_tokens = max_tokens

    def execute(self, context):
        from uuid import UUID

        validate_result = context['ti'].xcom_pull(task_ids='validate_event')
        document_id = UUID(validate_result['document_id'])

        if self.chunking_strategy == 'table':
            # Excel: Use RecordBuilder with detected headers
            from src.extraction_v2.record_builder import RecordBuilder
            from src.extraction_v2.llm_header_detector import HeaderStructure

            parse_result = context['ti'].xcom_pull(task_ids='parse_excel')
            headers_result = context['ti'].xcom_pull(task_ids='detect_headers')

            builder = RecordBuilder()
            all_records = []

            for table_data in parse_result.get('tables', []):
                headers = HeaderStructure(
                    header_rows=headers_result.get('header_rows', []),
                    headers=headers_result.get('headers', [])
                )
                # Build records using existing logic
                records = builder.build_records(
                    table=table_data,
                    headers=headers,
                    document_id=document_id,
                )
                all_records.extend(records)

            return {'chunks': all_records, 'chunk_count': len(all_records)}

        else:
            # Word/PDF/PowerPoint: Use SemanticChunker
            from src.extraction_v2.semantic_chunker import SemanticChunker
            from docling.document_converter import DocumentConverter

            parse_result = context['ti'].xcom_pull(
                task_ids=['parse_word', 'parse_pdf', 'parse_powerpoint'],
                include_prior_dates=True
            )
            # Find non-None result
            parse_result = next((r for r in parse_result if r), {})

            local_path = parse_result.get('local_path')
            if not local_path:
                return {'chunks': [], 'chunk_count': 0}

            # Re-convert to get DoclingDocument for chunking
            converter = DocumentConverter()
            result = converter.convert(local_path)

            chunker = SemanticChunker(max_tokens=self.max_tokens)
            chunks = chunker.chunk_document(
                doc=result.document,
                file_id=document_id,
                original_filename=parse_result.get('original_filename', 'document')
            )

            return {'chunks': chunks, 'chunk_count': len(chunks)}
```

---

## Configuration & Secrets

### Airflow Connections

```python
# scripts/setup_airflow_connections.py
from airflow.models import Connection
from airflow import settings

connections = [
    Connection(
        conn_id='seaweedfs_default',
        conn_type='s3',
        host='seaweedfs:8333',
        login='${SEAWEEDFS_ACCESS_KEY}',
        password='${SEAWEEDFS_SECRET_KEY}',
        extra={'endpoint_url': 'http://seaweedfs:8333'}
    ),
    Connection(
        conn_id='postgres_default',
        conn_type='postgres',
        host='postgres',
        login='${POSTGRES_USER}',
        password='${POSTGRES_PASSWORD}',
        schema='${POSTGRES_DB}',
        port=5432,
    ),
    Connection(
        conn_id='milvus_default',
        conn_type='generic',
        host='milvus',
        port=19530,
    ),
    Connection(
        conn_id='azure_openai_default',
        conn_type='http',
        host='${AZURE_OPENAI_ENDPOINT}',
        password='${AZURE_OPENAI_API_KEY}',
        extra={
            'api_version': '${AZURE_OPENAI_API_VERSION}',
            'deployment': '${AZURE_OPENAI_DEPLOYMENT}',
        }
    ),
]
```

### Environment Variables

```bash
# docker/.env.airflow
# Airflow Core
AIRFLOW__CORE__EXECUTOR=LocalExecutor
AIRFLOW__CORE__FERNET_KEY=your-fernet-key
AIRFLOW__WEBSERVER__SECRET_KEY=your-secret-key

# Database
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres:5432/airflow

# SeaweedFS (reuse existing)
SEAWEEDFS_HOST=localhost
SEAWEEDFS_PORT=8333
SEAWEEDFS_ACCESS_KEY_ID=dummy
SEAWEEDFS_SECRET_ACCESS_KEY=dummy

# PostgreSQL (reuse existing)
POSTGRES_HOST=postgres
POSTGRES_USER=${POSTGRES_USER}
POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
POSTGRES_DB=textiq

# Azure OpenAI (reuse existing)
AZURE_OPENAI_API_KEY=${AZURE_OPENAI_API_KEY}
AZURE_OPENAI_ENDPOINT=${AZURE_OPENAI_ENDPOINT}
AZURE_OPENAI_DEPLOYMENT=${AZURE_OPENAI_DEPLOYMENT}
AZURE_OPENAI_API_VERSION=2024-02-15-preview
```

---

## Testing Strategy

### Unit Tests

```python
# tests/test_operators/test_validate_event.py
import pytest
from plugins.operators.validate_event import ValidateEventOperator

class TestValidateEventOperator:

    def test_valid_seaweedfs_webhook(self, mock_context):
        """Test SeaweedFS webhook payload validation."""
        mock_context['dag_run'].conf = {
            'Records': [{
                's3': {
                    'bucket': {'name': 'documents'},
                    'object': {
                        'key': 'test/sample.xlsx',
                        'contentType': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
                        'size': 1024,
                    }
                }
            }],
            'document_id': 'uuid-here'
        }

        op = ValidateEventOperator(task_id='test')
        result = op.execute(mock_context)

        assert result['bucket'] == 'documents'
        assert result['object_key'] == 'test/sample.xlsx'

    def test_unsupported_content_type_raises(self, mock_context):
        """Test that unsupported file types raise error."""
        mock_context['dag_run'].conf = {
            'content_type': 'application/zip',
            'size_bytes': 100,
        }

        op = ValidateEventOperator(task_id='test')

        with pytest.raises(ValueError, match='Unsupported content type'):
            op.execute(mock_context)
```

### Integration Tests

```python
# tests/integration/test_dag_execution.py
import pytest
from airflow.models import DagBag

class TestDocumentExtractionDAG:

    @pytest.fixture
    def dag_bag(self):
        return DagBag(dag_folder='dags/', include_examples=False)

    def test_dag_loads_without_errors(self, dag_bag):
        """Verify DAG loads successfully."""
        assert 'document_extraction_dag' in dag_bag.dags
        assert len(dag_bag.import_errors) == 0

    def test_dag_task_count(self, dag_bag):
        """Verify expected number of tasks."""
        dag = dag_bag.dags['document_extraction_dag']
        # validate, fetch, branch, 4 parse, detect_headers, 2 chunk, embeddings, 2 store
        assert len(dag.tasks) >= 12

    def test_excel_path_dependencies(self, dag_bag):
        """Verify Excel processing path dependencies."""
        dag = dag_bag.dags['document_extraction_dag']

        # Verify parse_excel → detect_headers → chunk_excel
        parse_excel = dag.get_task('parse_excel')
        detect_headers = dag.get_task('detect_headers')
        chunk_excel = dag.get_task('chunk_excel')

        assert detect_headers in parse_excel.downstream_list
        assert chunk_excel in detect_headers.downstream_list
```

### End-to-End Tests

```python
# tests/e2e/test_full_extraction.py
import pytest
from airflow.models import DagRun
from airflow.utils.state import DagRunState

class TestFullExtraction:

    @pytest.fixture
    def sample_xlsx(self, tmp_path):
        """Create sample Excel file for testing."""
        import openpyxl
        wb = openpyxl.Workbook()
        ws = wb.active
        ws['A1'] = 'Name'
        ws['B1'] = 'Value'
        ws['A2'] = 'Test'
        ws['B2'] = '123'

        path = tmp_path / 'sample.xlsx'
        wb.save(path)
        return str(path)

    @pytest.mark.e2e
    def test_excel_extraction_e2e(self, sample_xlsx, airflow_client):
        """Test full Excel extraction through Airflow."""
        # Trigger DAG
        run_id = airflow_client.trigger_dag(
            'document_extraction_dag',
            conf={
                'document_id': 'test-uuid',
                'file_path': sample_xlsx,
                'content_type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
            }
        )

        # Wait for completion
        airflow_client.wait_for_dag_run(run_id, timeout=300)

        # Verify results
        dag_run = DagRun.find(dag_id='document_extraction_dag', run_id=run_id)[0]
        assert dag_run.state == DagRunState.SUCCESS
```

---

## Rollback Plan

### Immediate Rollback (< 5 minutes)

1. **Traffic Switch:**
   ```python
   # Update feature flag
   await feature_flags.set("airflow_extraction", percentage=0)
   ```

2. **Verify Celery Active:**
   ```bash
   celery -A src.workers.celery_app inspect active
   ```

### Full Rollback (< 30 minutes)

1. **Stop Airflow Services:**
   ```bash
   docker-compose -f docker/docker-compose.airflow.yml down
   ```

2. **Clear Incomplete DAG Runs:**
   ```sql
   UPDATE extraction_jobs
   SET status = 'cancelled',
       error_message = 'Rollback: switched to Celery'
   WHERE status IN ('pending', 'running');
   ```

3. **Re-queue Pending Documents:**
   ```python
   # Re-queue documents that were in-flight during rollback
   pending_docs = db.query(Document).filter(Document.status == 'processing').all()
   for doc in pending_docs:
       extract_document_v2.delay(str(doc.id), doc.file_path)
   ```

4. **Monitor Celery Workers:**
   ```bash
   celery -A src.workers.celery_app flower  # Open monitoring UI
   ```

---

## Risk Assessment

| Risk | Likelihood | Impact | Mitigation |
|------|------------|--------|------------|
| **Airflow DB migration issues** | Medium | High | Use separate Airflow DB, test migrations in staging |
| **XCom data size limits** | Medium | Medium | Store large data in SeaweedFS, pass references in XCom |
| **Operator execution timeout** | Medium | Medium | Configure appropriate task timeouts, add monitoring |
| **SeaweedFS webhook reliability** | Low | High | Implement webhook retry + dead letter queue (SeaweedFS 4.02+) |
| **Azure OpenAI rate limiting** | High | Medium | Existing rate limiting logic in operator |
| **LibreOffice concurrency** | Medium | Medium | Reuse existing file-based locking mechanism |
| **Memory pressure during chunking** | Medium | Medium | Configure worker resource limits, use streaming |

### Mitigation Details

**XCom Data Size:**
```python
# Store large data in SeaweedFS, pass reference
def execute(self, context):
    large_result = {...}  # Could be > 64KB

    # Store in SeaweedFS
    key = f"xcom/{context['dag_run'].run_id}/{self.task_id}.json"
    seaweedfs_client.put_object('airflow-xcom', key, json.dumps(large_result))

    # Return reference
    return {'xcom_key': key}
```

**Task Timeouts:**
```python
# DAG definition
parse_excel = ParseDocumentOperator(
    task_id='parse_excel',
    doc_type='excel',
    execution_timeout=timedelta(minutes=30),  # Large file timeout
)

generate_embeddings = GenerateEmbeddingsOperator(
    task_id='generate_embeddings',
    execution_timeout=timedelta(minutes=60),  # API call timeout
)
```

---

## Success Metrics

| Metric | Current (Celery) | Target (Airflow) | Measurement |
|--------|------------------|------------------|-------------|
| **Processing Success Rate** | 95% | ≥ 95% | DAG run success rate |
| **Average Processing Time** | 45s | ≤ 50s | DAG duration histogram |
| **P99 Processing Time** | 180s | ≤ 200s | DAG duration percentile |
| **Retry Rate** | 8% | ≤ 10% | Task retry counter |
| **Manual Intervention Rate** | 5% | ≤ 3% | Failed DAG runs requiring manual action |
| **Developer Debugging Time** | 30min avg | ≤ 15min | Time to identify failure cause |

---

## Appendix

### A. Glossary

| Term | Definition |
|------|------------|
| **DAG** | Directed Acyclic Graph - Airflow's workflow representation |
| **Operator** | Airflow task template that defines what work to perform |
| **XCom** | Cross-communication mechanism for passing data between tasks |
| **Sensor** | Operator that waits for an external condition |
| **Hook** | Interface to external systems (SeaweedFS, Milvus, etc.) |
| **Docling** | Open-source document conversion library |

### B. Related Documents

- [Document Extraction Engine Architecture](./document-extraction-engine.md)
- [Extraction Pipeline V2 Tech Spec](./extraction-pipeline-v2-tech-spec.md)
- [Apache Airflow 3.x Documentation](https://airflow.apache.org/docs/)
- [Docling Documentation](https://github.com/docling-project/docling)

### C. Change Log

| Version | Date | Author | Changes |
|---------|------|--------|---------|
| 1.0 | 2025-12-08 | AI | Initial specification |

---

_Generated: 2025-12-08_