# Airflow Setup and Configuration Guide

> Comprehensive documentation for deploying and operating Apache Airflow 3.x in the TextIQ document extraction pipeline.

## Table of Contents

1. [Overview](#overview)
2. [Architecture](#architecture)
3. [Prerequisites](#prerequisites)
4. [Quick Start](#quick-start)
5. [Environment Configuration](#environment-configuration)
6. [Docker Services](#docker-services)
7. [DAG Structure](#dag-structure)
8. [Operations](#operations)
9. [Troubleshooting](#troubleshooting)
10. [Development Workflow](#development-workflow)

---

## Overview

TextIQ uses Apache Airflow 3.x for DAG-based orchestration of the document extraction pipeline. The system processes multi-format documents (Excel, Word, PDF, PowerPoint) through extraction, chunking, embedding generation, and vector storage in Milvus.

### Key Features

- **Event-driven processing**: Triggered via REST API or SeaweedFS webhooks
- **File-type branching**: Different processing paths for spreadsheets vs. text documents
- **External Python execution**: Heavy tasks run in isolated virtual environment (`/opt/airflow/.venv`)
- **Vector storage**: Document vectors stored in Milvus for similarity search
- **LocalExecutor**: Single-node deployment without Celery workers

### Supported File Types

| Type | Extensions | Processing Strategy |
|------|------------|---------------------|
| Excel | `.xlsx`, `.xls`, `.xlsm` | Table-based chunking with header detection |
| Word | `.docx`, `.doc`, `.docm` | Semantic chunking (converts legacy formats via LibreOffice) |
| PDF | `.pdf` | Semantic chunking via Docling |
| PowerPoint | `.pptx`, `.ppt`, `.pptm` | Semantic chunking |

---

## Architecture

### System Components

```
┌─────────────────────────────────────────────────────────────────────────────┐
│                            Docker Network (textiq-network)                   │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  ┌─────────────┐    ┌──────────────┐    ┌────────────────┐                 │
│  │  Airflow    │    │   Airflow    │    │    Airflow     │                 │
│  │  Webserver  │◄───│  Scheduler   │◄───│  DAG Processor │                 │
│  │  (port 8081)│    │              │    │                │                 │
│  └──────┬──────┘    └──────┬───────┘    └───────┬────────┘                 │
│         │                  │                    │                           │
│         └──────────────────┼────────────────────┘                           │
│                            │                                                │
│                   ┌────────▼────────┐                                       │
│                   │    PostgreSQL   │  (Airflow metadata only)              │
│                   │   (port 5432)   │                                       │
│                   └─────────────────┘                                       │
│                                                                              │
│  ┌─────────────┐    ┌──────────────┐    ┌────────────────┐                 │
│  │   Milvus    │    │  SeaweedFS   │    │     Redis      │                 │
│  │ (port 19530)│    │ (port 8333)  │    │  (port 6379)   │                 │
│  └─────────────┘    └──────────────┘    └────────────────┘                 │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘
```

### DAG Processing Flow

```
validate_event → fetch_document → detect_format (branch)
                                        │
        ┌───────────────┬───────────────┼───────────────┬───────────────┐
        ↓               ↓               ↓               ↓               ↓
   parse_excel    parse_word      parse_pdf     parse_powerpoint  unsupported
        │               │               │               │
        ↓               └───────────────┴───────────────┘
  detect_headers                        │
        │                               ↓
        ↓                        select_parse_result
   chunk_excel                          │
        │                               ↓
        ↓                          chunk_text
generate_embeddings_excel               │
        │                               ↓
        ↓                    generate_embeddings_text
  store_vectors_excel                   │
   (Milvus)                             ↓
                                 store_vectors_text
                                   (Milvus)
```

### Directory Structure

```
textiq-doc-extraction/
├── airflow/
│   ├── config/                    # Airflow configuration (gitkeep)
│   ├── dags/
│   │   ├── __init__.py
│   │   ├── config.py              # Shared DAG configuration
│   │   ├── document_extraction_dag.py  # Main DAG definition
│   │   └── tasks/
│   │       ├── __init__.py
│   │       ├── parse_tasks.py     # @task.external_python parse functions
│   │       └── processing_tasks.py # @task.external_python processing functions
│   ├── logs/                      # Task logs (auto-generated)
│   └── plugins/
│       ├── __init__.py            # TextIQPlugin registration
│       ├── hooks/                 # Custom Airflow hooks (future)
│       ├── models/
│       │   └── xcom_schemas.py    # XCom data schema definitions
│       └── operators/
│           ├── __init__.py
│           ├── validate_event.py  # ValidateEventOperator
│           └── fetch_document.py  # FetchDocumentOperator
├── docker/
│   ├── docker-compose.yml         # Core services (PostgreSQL, Milvus, Redis, SeaweedFS)
│   ├── docker-compose.airflow.yml # Airflow services
│   ├── Dockerfile.airflow         # Custom Airflow image with uv
│   ├── airflow-entrypoint.sh      # Container startup script
│   └── init-db.sh                 # PostgreSQL initialization
└── .env                           # Environment configuration
```

---

## Prerequisites

### System Requirements

- **Docker**: 20.10+ with Docker Compose V2
- **Memory**: Minimum 8GB RAM (16GB recommended)
- **Disk**: 20GB+ free space for Docker images and volumes
- **Ports**: 8081 (Airflow UI), 5432 (PostgreSQL), 8333 (SeaweedFS), 19530 (Milvus)

### Required Accounts/Services

- **Azure OpenAI**: API key and endpoint for embeddings and LLM header detection
- Optional: Azure Redis Cache for production deployments

---

## Quick Start

### 1. Clone and Configure

```bash
# Clone repository
git clone <repository-url>
cd textiq-doc-extraction

# Copy environment template
cp .env.example .env

# Edit .env with your Azure OpenAI credentials
# Required: AZURE_OPENAI_API_KEY, AZURE_OPENAI_ENDPOINT, etc.
```

**Note on file locations**:
```
textiq-doc-extraction/
├── .env                 # Environment config (project root)
├── .env.example         # Template (project root)
└── docker/
    ├── docker-compose.yml          # Core services
    ├── docker-compose.airflow.yml  # Airflow services
    ├── Dockerfile.airflow          # Custom Airflow image
    ├── airflow-entrypoint.sh       # Container startup
    └── init-db.sh                  # PostgreSQL init
```

All Docker commands should be run from the **project root** directory, specifying the compose file path with `-f docker/docker-compose.yml`.

### 2. Set Airflow UID (Linux only)

```bash
# Get your user ID
echo "AIRFLOW_UID=$(id -u)" >> .env
```

### 3. Start Core Services

```bash
# From project root directory
# Start PostgreSQL, Redis, Milvus, SeaweedFS
docker compose -f docker/docker-compose.yml up -d

# Wait for services to be healthy
docker compose -f docker/docker-compose.yml ps
```

### 4. Initialize Airflow Database

```bash
# Run one-time database migration (init profile)
docker compose -f docker/docker-compose.airflow.yml --profile init up airflow-init

# This creates:
# - Airflow metadata tables in PostgreSQL
# - Admin user (credentials from .env)
```

### 5. Start Airflow Services

```bash
# Start all Airflow components
docker compose -f docker/docker-compose.airflow.yml up -d

# Verify all services are running
docker compose -f docker/docker-compose.airflow.yml ps
```

### 6. Access Airflow UI

- **URL**: http://localhost:8081
- **Username**: `admin` (or value of `_AIRFLOW_WWW_USER_USERNAME`)
- **Password**: `admin` (or value of `_AIRFLOW_WWW_USER_PASSWORD`)

### 7. Unpause the DAG

The DAG is paused by default. Enable it via:
- **UI**: Toggle the DAG switch in the Airflow web interface
- **CLI**: `docker exec textiq-airflow-scheduler airflow dags unpause document_extraction_dag`

---

## Environment Configuration

### Required Environment Variables

Create a `.env` file in the project root with these variables:

```bash
# ===================
# Airflow User Configuration
# ===================
# Linux user ID for Airflow containers (run `id -u` to get your UID)
AIRFLOW_UID=50000

# ===================
# Airflow Web UI Credentials
# ===================
_AIRFLOW_WWW_USER_USERNAME=admin
_AIRFLOW_WWW_USER_PASSWORD=admin

# ===================
# Airflow Internal Security
# ===================
# Generate with: python -c "import secrets; print(secrets.token_urlsafe(32))"
AIRFLOW__API_AUTH__JWT_SECRET=your-secure-jwt-secret

# ===================
# Airflow DAG Timeouts (in minutes)
# ===================
AIRFLOW_PARSE_TIMEOUT_MINUTES=120    # Document parsing timeout
AIRFLOW_EMBEDDING_TIMEOUT_MINUTES=60 # Embedding generation timeout
AIRFLOW_STORE_TIMEOUT_MINUTES=30     # Storage operation timeout

# ===================
# Database Configuration
# ===================
POSTGRES_USER=textiq
POSTGRES_PASSWORD=textiq_dev_password
POSTGRES_DB=textiq
DATABASE_URL=postgresql+asyncpg://textiq:textiq_dev_password@postgres:5432/textiq

# ===================
# Redis Configuration
# ===================
REDIS_URL=redis://redis:6379

# ===================
# Milvus Configuration
# ===================
MILVUS_HOST=milvus
MILVUS_PORT=19530
MILVUS_COLLECTION=textiq_records

# ===================
# SeaweedFS Configuration
# ===================
SEAWEEDFS_HOST=seaweedfs
SEAWEEDFS_PORT=8333
SEAWEEDFS_ACCESS_KEY_ID=dummy
SEAWEEDFS_SECRET_ACCESS_KEY=dummy

# ===================
# Azure OpenAI Configuration (REQUIRED)
# ===================
AZURE_OPENAI_API_KEY=your-api-key-here
AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com/
AZURE_OPENAI_API_VERSION=2024-02-15-preview
AZURE_OPENAI_DEPLOYMENT=gpt-4o
AZURE_OPENAI_EMBEDDING_DEPLOYMENT=text-embedding-3-large
```

### Airflow-Specific Configuration

These settings are configured in `docker-compose.airflow.yml`:

| Setting | Value | Description |
|---------|-------|-------------|
| `AIRFLOW__CORE__EXECUTOR` | `LocalExecutor` | Single-node execution without Celery |
| `AIRFLOW__CORE__LOAD_EXAMPLES` | `false` | Disable example DAGs |
| `AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION` | `true` | DAGs paused on first load |
| `AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_SEC` | `30` | Heartbeat interval for long tasks |
| `AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_TIMEOUT` | `120` | Heartbeat timeout (120 * 30s = 1 hour) |
| `AIRFLOW__WORKERS__EXECUTION_API_TIMEOUT` | `300` | API timeout for external_python tasks |

---

## Docker Services

### Core Services (`docker-compose.yml`)

| Service | Image | Port | Purpose |
|---------|-------|------|---------|
| `postgres` | `postgres:18` | 5432 | Database for Airflow metadata |
| `redis` | `redis:7` | 6379 | Caching and Celery broker (optional) |
| `milvus` | `milvusdb/milvus:v2.4.13` | 19530 | Vector database for similarity search |
| `seaweedfs` | `chrislusf/seaweedfs:latest` | 8333 | S3-compatible document storage |
| `attu` | `zilliz/attu:v2.4` | 3000 | Milvus web UI (optional) |

### Airflow Services (`docker-compose.airflow.yml`)

| Service | Purpose | Health Check |
|---------|---------|--------------|
| `airflow-init` | One-time database migration | Profile: `init` |
| `airflow-webserver` | Web UI and REST API (port 8081) | `/api/v2/monitor/health` |
| `airflow-scheduler` | Task scheduling | Port 8974 |
| `airflow-dag-processor` | DAG file parsing | DagProcessorJob check |
| `airflow-triggerer` | Async trigger handling | TriggererJob check |
| `airflow-cli` | CLI access | Profile: `cli` |

### Custom Airflow Image (`Dockerfile.airflow`)

Built from `apache/airflow:3.0.6-python3.11` with:

- System dependencies: `libgl1`, `libglib2.0-0`, `libreoffice` (for document conversion)
- **uv** package manager for fast dependency installation
- Virtual environment at `/opt/airflow/.venv` with all TextIQ dependencies
- Project dependencies installed via `uv sync`

```dockerfile
FROM apache/airflow:3.0.6-python3.11

# Install system dependencies
RUN apt-get update && apt-get install -y \
    build-essential curl libgl1 libglib2.0-0 libreoffice

# Install uv and project dependencies
RUN curl -LsSf https://astral.sh/uv/install.sh | sh
RUN uv sync --frozen --no-install-project

# Set PATH to use virtual environment
ENV PATH="/opt/airflow/.venv/bin:$PATH"
```

---

## DAG Structure

### Main DAG: `document_extraction_dag`

- **DAG ID**: `document_extraction_dag`
- **Schedule**: `None` (event-triggered only)
- **Max Active Runs**: 1 (see note below)
- **Tags**: `['extraction', 'v2', 'docling']`

#### Why `max_active_runs=1`?

The DAG is configured with `max_active_runs=1`, meaning only one DAG run can execute at a time. This is a **critical limitation** due to LibreOffice:

**The Constraint**: LibreOffice can only run one instance at a time per user profile. When converting legacy formats:
- `.xls`, `.xlsm` → `.xlsx` (Excel)
- `.doc`, `.docm` → `.docx` (Word)

LibreOffice uses a single user profile directory and locks it during conversion. Running multiple LibreOffice processes simultaneously causes:
- File lock conflicts
- Conversion failures
- Corrupted output files

**Impact**:
- Documents are processed sequentially, not in parallel
- Queue builds up during high volume periods
- Throughput is limited by single-document processing time

**Future Improvements** (if parallel processing is needed):
1. **Separate LibreOffice profiles**: Configure isolated `$HOME` directories per worker
2. **LibreOffice server mode**: Run LibreOffice as a service with connection pooling
3. **Alternative converters**: Use `unoconv` with multiple instances or cloud-based conversion
4. **Pre-conversion**: Convert legacy formats before uploading to SeaweedFS

**Current Workaround**: If you only process modern formats (`.xlsx`, `.docx`, `.pdf`, `.pptx`), LibreOffice is not invoked and parallel processing would be safe. However, the DAG currently enforces `max_active_runs=1` to handle any file type safely.

### Task Types

#### 1. Custom Operators (Light Tasks)

Located in `airflow/plugins/operators/`:

| Operator | Purpose |
|----------|---------|
| `ValidateEventOperator` | Validates incoming events, extracts metadata |
| `FetchDocumentOperator` | Downloads document from SeaweedFS to local storage |

#### 2. TaskFlow Functions (`@task.external_python`)

Run in `/opt/airflow/.venv` with full TextIQ dependencies.

**Why `@task.external_python`?**

The Airflow base image (`apache/airflow:3.0.6-python3.11`) ships with a minimal Python environment optimized for workflow orchestration. However, TextIQ's document processing requires heavy dependencies that:

1. **Conflict with Airflow's environment**: Packages like `docling`, `torch`, and `transformers` have complex dependency trees that can conflict with Airflow's pinned versions
2. **Are too large for the base image**: ML libraries and document processing tools (OpenCV, LibreOffice bindings) add significant size and build complexity
3. **Require specific system libraries**: Document parsing needs `libgl1`, `libglib2.0-0`, and LibreOffice for format conversion
4. **Have different lifecycle requirements**: TextIQ dependencies update more frequently than Airflow itself

By using `@task.external_python(python='/opt/airflow/.venv/bin/python')`, we:
- **Isolate environments**: Airflow orchestration runs in its own environment, processing runs in the uv venv
- **Simplify dependency management**: Use `uv sync` from `pyproject.toml` without affecting Airflow
- **Enable independent updates**: Update TextIQ dependencies without rebuilding Airflow
- **Prevent import errors**: Tasks only import heavy dependencies inside the function body, avoiding DAG parsing failures

**Important**: All imports must be inside the function body for `@task.external_python` to work:

```python
@task.external_python(python='/opt/airflow/.venv/bin/python')
def my_task(input_data: dict) -> dict:
    # Imports INSIDE the function - runs in external venv
    from src.extraction_v2.document_converter import DocumentToHtmlConverter

    converter = DocumentToHtmlConverter()
    return converter.process(input_data)
```

**Processing Tasks using `@task.external_python`**:

**Parse Tasks** (`tasks/parse_tasks.py`):
- `parse_excel_document`: Docling + custom table extraction
- `parse_word_document`: Docling with .doc/.docm conversion
- `parse_pdf_document`: Docling PDF processing
- `parse_powerpoint_document`: Docling PPT processing

**Processing Tasks** (`tasks/processing_tasks.py`):
- `detect_headers_task`: LLM-based header detection (Excel only)
- `chunk_excel_task`: Table-based chunking with `RecordBuilder`
- `chunk_text_task`: Semantic chunking with `SemanticChunker`
- `generate_embeddings_task`: Azure OpenAI embeddings
- `store_vectors_task`: Milvus vector indexing

### DAG Parameters

Trigger the DAG with these parameters:

```json
{
  "document_id": "550e8400-e29b-41d4-a716-446655440000",
  "object_key": "uploads/2025/12/report.xlsx",
  "content_type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
}
```

### XCom Data Limits and Large Data Handling

**The Problem**: Airflow's XCom (cross-communication) system stores task return values in the metadata database. By default, XCom has size limits that vary by database backend:
- **PostgreSQL**: ~1GB theoretical, but practical limit is much lower due to performance
- **Recommended limit**: Keep XCom values under 48KB for reliable operation

**Why This Matters**: Document processing generates large data at every stage:
- **Parse results**: HTML/markdown content can be several MB
- **Chunk data**: Lists of dictionaries with content, headers, and source info
- **Embedding vectors**: 3072 floats per chunk × hundreds of chunks = megabytes of data

**All heavy tasks use file-based data transfer**:

| Task | Pickle File Location |
|------|---------------------|
| `parse_excel_document` | `/opt/airflow/logs/parse_excel_{document_id}.pkl` |
| `parse_word_document` | `/opt/airflow/logs/parse_word_{document_id}.pkl` |
| `parse_pdf_document` | `/opt/airflow/logs/parse_pdf_{document_id}.pkl` |
| `parse_powerpoint_document` | `/opt/airflow/logs/parse_ppt_{document_id}.pkl` |
| `chunk_excel_task` | `/opt/airflow/logs/chunks_excel_{document_id}.pkl` |
| `chunk_text_task` | `/opt/airflow/logs/chunks_text_{document_id}.pkl` |
| `generate_embeddings_task` | `/opt/airflow/logs/embedding_data/embeddings_{uuid}.pkl` |

**Never return large data directly from tasks**:
```python
# BAD - Will fail or crash Airflow
@task.external_python(python=EXTERNAL_PYTHON)
def parse_document_bad(fetch_result: dict) -> dict:
    content = converter.convert(fetch_result['local_path'])
    return {
        "markdown_content": content,  # Could be MB of data
        "tables": tables,             # Large nested structures
    }
```

**Solution: File-based data transfer pattern**

All parse, chunk, and embedding tasks follow this pattern:

```python
@task.external_python(python=EXTERNAL_PYTHON)
def parse_document(fetch_result: dict) -> dict:
    import pickle

    # ... process document ...

    # Save large data to pickle file
    parse_data = {
        'doc_type': 'word',
        'markdown_content': markdown,
        'tables': tables,
        'document_id': document_id,
    }

    data_file = f"/opt/airflow/logs/parse_word_{document_id}.pkl"
    with open(data_file, 'wb') as f:
        pickle.dump(parse_data, f)

    # Return only small metadata via XCom
    return {
        "data_file": data_file,      # Path to the pickle file
        "doc_type": "word",          # Short string
        "document_id": document_id,  # UUID string
        "content_length": len(markdown),  # Small integer
    }
```

**Downstream tasks load and clean up**:

```python
@task.external_python(python=EXTERNAL_PYTHON)
def chunk_text_task(document_id: str, parse_result: dict) -> dict:
    import pickle
    import os

    # Load data from pickle file
    data_file = parse_result.get('data_file')
    if data_file and os.path.exists(data_file):
        with open(data_file, 'rb') as f:
            parse_result = pickle.load(f)
        # Clean up after loading
        os.remove(data_file)

    # ... process chunks ...
```

**Data flow through pickle files**:
```
parse_*_document → pickle → detect_headers/chunk_*_task → pickle → generate_embeddings_task → pickle → store_vectors_task
```

Each task loads data from the previous pickle file and cleans it up after processing.

**Guidelines for XCom usage**:
| Data Type | Size | Approach |
|-----------|------|----------|
| IDs, counts, status | < 1KB | Return directly |
| File paths, small metadata | < 10KB | Return directly |
| Parsed content, chunks | 10KB - 1MB | Consider file storage |
| Embeddings, large lists | > 1MB | **Always use file storage** |

---

## Operations

### Starting/Stopping Services

All commands should be run from the **project root** directory.

```bash
# Start core services (PostgreSQL, Redis, Milvus, SeaweedFS)
docker compose -f docker/docker-compose.yml up -d

# Start Airflow services
docker compose -f docker/docker-compose.airflow.yml up -d

# Stop Airflow only
docker compose -f docker/docker-compose.airflow.yml down

# Stop all services
docker compose -f docker/docker-compose.airflow.yml down
docker compose -f docker/docker-compose.yml down

# Stop and remove volumes (CAUTION: deletes all data)
docker compose -f docker/docker-compose.yml down -v
docker compose -f docker/docker-compose.airflow.yml down -v
```

**Tip**: Create shell aliases to simplify commands:
```bash
# Add to ~/.bashrc or ~/.zshrc
alias dc-core='docker compose -f docker/docker-compose.yml'
alias dc-airflow='docker compose -f docker/docker-compose.airflow.yml'

# Then use:
dc-core up -d
dc-airflow up -d
dc-airflow logs -f
```

### Triggering DAG Runs

#### Via REST API

```bash
# Basic auth with admin credentials
curl -X POST "http://localhost:8081/api/v1/dags/document_extraction_dag/dagRuns" \
  -H "Content-Type: application/json" \
  -u "admin:admin" \
  -d '{
    "conf": {
      "document_id": "550e8400-e29b-41d4-a716-446655440000",
      "object_key": "uploads/report.xlsx",
      "content_type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    }
  }'
```

#### Via Airflow CLI

```bash
docker exec textiq-airflow-scheduler airflow dags trigger \
  document_extraction_dag \
  --conf '{"document_id": "...", "object_key": "...", "content_type": "..."}'
```

### Viewing Logs

```bash
# All Airflow logs
docker compose -f docker/docker-compose.airflow.yml logs -f

# Specific service logs
docker logs -f textiq-airflow-scheduler
docker logs -f textiq-airflow-webserver

# Task logs (in Airflow UI or filesystem)
ls airflow/logs/dag_id=document_extraction_dag/
```

### DAG Management

```bash
# List DAGs
docker exec textiq-airflow-scheduler airflow dags list

# Pause/Unpause DAG
docker exec textiq-airflow-scheduler airflow dags pause document_extraction_dag
docker exec textiq-airflow-scheduler airflow dags unpause document_extraction_dag

# Test a specific task
docker exec textiq-airflow-scheduler airflow tasks test \
  document_extraction_dag validate_event 2024-01-01
```

### Health Checks

```bash
# Check Airflow health
curl http://localhost:8081/api/v2/monitor/health

# Check scheduler health
curl http://localhost:8974/health

# Check all container status
docker compose -f docker/docker-compose.airflow.yml ps
```

---

## Troubleshooting

### Common Issues

#### 1. DAG Import Errors

**Symptom**: DAG doesn't appear in UI, errors in dag-processor logs

**Solution**:
```bash
# Check DAG processor logs
docker logs textiq-airflow-dag-processor

# Test DAG import manually
docker exec textiq-airflow-scheduler python -c \
  "from airflow.models import DagBag; db = DagBag(); print(db.import_errors)"
```

#### 2. Task Stuck in "Running"

**Symptom**: Tasks show "running" indefinitely

**Causes & Solutions**:
- **Heartbeat timeout**: Increase `AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_TIMEOUT`
- **Memory issues**: Check container memory with `docker stats`
- **Deadlock**: Restart scheduler: `docker restart textiq-airflow-scheduler`

#### 3. External Python Tasks Fail

**Symptom**: `@task.external_python` tasks fail with import errors

**Solution**:
```bash
# Verify venv exists and has dependencies
docker exec textiq-airflow-scheduler /opt/airflow/.venv/bin/python -c \
  "from src.extraction_v2.document_converter import DocumentToHtmlConverter; print('OK')"

# Rebuild image if dependencies changed
docker compose -f docker/docker-compose.airflow.yml build --no-cache
```

#### 4. XCom Size Exceeded

**Symptom**: `XCom value is too large` errors

**Solution**: The DAG uses pickle files for large data. If issues persist:
```python
# Verify MAX_XCOM_SIZE in config.py
MAX_XCOM_SIZE = 48 * 1024  # 48KB
```

#### 5. Database Connection Issues

**Symptom**: `connection refused` or `database does not exist`

**Solution**:
```bash
# Ensure PostgreSQL is running and healthy
docker compose -f docker/docker-compose.yml ps postgres

# Check Airflow database exists
docker exec textiq-postgres psql -U textiq -d postgres -c "\l" | grep airflow

# Reinitialize if needed
docker compose -f docker/docker-compose.airflow.yml --profile init up airflow-init
```

#### 6. Permission Errors (Linux)

**Symptom**: `Permission denied` in logs

**Solution**:
```bash
# Set correct UID in .env
echo "AIRFLOW_UID=$(id -u)" >> .env

# Fix existing log permissions
sudo chown -R $(id -u):0 airflow/logs
```

### Log Locations

| Log Type | Location |
|----------|----------|
| Scheduler | `docker logs textiq-airflow-scheduler` |
| Webserver | `docker logs textiq-airflow-webserver` |
| DAG Processor | `docker logs textiq-airflow-dag-processor` |
| Task Logs | `airflow/logs/dag_id=document_extraction_dag/` |
| Embedding Data | `airflow/logs/embeddings_*.pkl` (temporary) |

---

## Development Workflow

### Local DAG Development

1. **Edit DAG files**: Changes in `airflow/dags/` are auto-detected
2. **Check for errors**: View dag-processor logs
3. **Test tasks individually**:
   ```bash
   docker exec textiq-airflow-scheduler airflow tasks test \
     document_extraction_dag <task_id> 2024-01-01
   ```

### Adding New Tasks

1. For light tasks without `src.*` imports: Create operator in `plugins/operators/`
2. For heavy tasks with dependencies: Create `@task.external_python` in `dags/tasks/`

Example `@task.external_python`:
```python
from airflow.decorators import task

EXTERNAL_PYTHON = '/opt/airflow/.venv/bin/python'

@task.external_python(python=EXTERNAL_PYTHON)
def my_new_task(input_data: dict) -> dict:
    # All imports inside the function
    from src.mymodule import MyClass

    result = MyClass().process(input_data)
    return {"result": result}
```

### Updating Dependencies

1. Update `pyproject.toml` with new dependencies
2. Rebuild the Airflow image:
   ```bash
   docker compose -f docker/docker-compose.airflow.yml build --no-cache
   docker compose -f docker/docker-compose.airflow.yml up -d
   ```

### Testing

```bash
# Run DAG unit tests
cd airflow
pytest tests/

# Test specific operator
pytest tests/operators/test_validate_event.py -v
```

---

## API Reference

### Airflow REST API

Base URL: `http://localhost:8081/api/v1`

| Endpoint | Method | Description |
|----------|--------|-------------|
| `/dags` | GET | List all DAGs |
| `/dags/{dag_id}/dagRuns` | POST | Trigger DAG run |
| `/dags/{dag_id}/dagRuns/{run_id}` | GET | Get DAG run status |
| `/dags/{dag_id}/dagRuns/{run_id}/taskInstances` | GET | Get task instances |

### Health Endpoints

| Endpoint | Description |
|----------|-------------|
| `/api/v2/monitor/health` | Overall Airflow health |
| `localhost:8974/health` | Scheduler health |

---

## Related Documentation

- [Tech Spec: airflow](specs/airflow.md) - Detailed component specification
- [Migration Epics](feats/airflow-migration/epics.md) - Migration planning and status
- [Sprint Status](feats/airflow-migration/sprint-status.yaml) - Current implementation progress

---

*Last updated: 2025-12-18*