---
title: 'Ingest .md files through the document extraction DAG'
type: 'feature'
created: '2026-05-18'
status: 'done'
baseline_commit: 'f46be6d'
context:
  - '_planning/docs/services/textiq-doc-extraction/architecture.md'
  - 'docs/implementation-artifacts/spec-md-ingestion-s3-upload.md'
---

<frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">

## Intent

**Problem:** `.md` uploads are stored in S3 but never extracted — the upload endpoint marks them `status="completed"`/`approval_status="approved"` at insert time, so the extraction DAG never runs and markdown content (incl. Mermaid flowcharts) is absent from Milvus + the PG docstore. The existing `SentenceSplitter`-only hierarchy also fragments long fenced blocks across many ~256-token leaves where they're un-retrievable.

**Approach:** Treat `.md` as a first-class text format that needs no parsing — it is already markdown. (1) Flip the upload endpoint so `.md` rows are `pending`/`pending`, reusing the existing approval-gated DAG trigger. (2) `detect_format` for `.md` routes directly to `hierarchical_chunk_text` (no parse task, no `select_parse_result` for this branch). (3) `hierarchical_chunk_text_task` gains an optional `fetch_result_md` parameter; when set **and** `original_filename` ends in `.md` it reads the `.md` from `fetch_result.local_path`, decodes UTF-8, extracts each ` ```mermaid ... ``` ` fence and **removes it from the prose entirely** (no placeholder), chunks the remaining prose with `MarkdownNodeParser → SentenceSplitter(1024) → SentenceSplitter(256)`, then appends each mermaid fence as a **standalone orphan `TextNode`** (no parent wiring). Fences exceeding `settings.embedding_model_context_tokens` are split along mermaid line boundaries into multiple orphan leaves sharing the same `diagram_index`. When the file is mermaid-only (no prose left after fence removal) the prose hierarchy step is skipped and only mermaid leaves are emitted. Other formats unchanged.

**Mermaid wiring decision (revised):** the original design wired each mermaid leaf as a child of the mid-level parent that owned the placeholder leaf so AutoMergingRetriever could collapse both into a single rich chunk. In practice the merger collapsed the mermaid leaf into a parent whose visible text was still `[MERMAID_DIAGRAM_N]` (the placeholder), dropping the 7000-char diagram body from the retrieved result. We now keep mermaid leaves orphan/standalone so each diagram surfaces as its own retrievable chunk; the placeholder is no longer useful and is removed to avoid polluting retrieval with a near-empty prose chunk (e.g. `## フロー図 (Flow Diagram)\n\n[MERMAID_DIAGRAM_1]`).

## Boundaries & Constraints

**Always:**
- Reuse the existing text pipeline (chunk → domain terms → tags → `store_to_llamaindex_task`). No new store task, no new DAG.
- No new parse task. The `.md` file is already markdown — `hierarchical_chunk_text_task` reads it directly via the existing `fetch_result.local_path` produced by `fetch_document_from_s3`.
- `hierarchical_chunk_text_task` gains an optional `fetch_result_md: dict = None` parameter. The DAG wires `fetch_result_md=fetch.output` unconditionally (Airflow XCom kwargs can't be conditionally omitted at parse time), so the task **gates on the filename extension**: the markdown branch fires only when `fetch_result_md is not None and original_filename.lower().endswith(".md")`. When the branch fires, the task: (a) decodes `local_path` as UTF-8 (let `UnicodeDecodeError` propagate); (b) reads embedding limit from `Settings().embedding_model_context_tokens`; (c) extracts each ` ```mermaid ... ``` ` fence via regex (`r"```mermaid\b[^\n]*\n(.*?)\n```"`, DOTALL) and **removes the fence from the prose entirely** (the regex callback returns `""`, not a placeholder); (d) tokenises each fence with `tiktoken.get_encoding("cl100k_base")` — if a fence exceeds the limit, splits it along mermaid line boundaries into K chunks (each ≤ limit, each re-wrapped as ` ```mermaid ... ``` `); (e) if the stripped prose is non-empty, builds `HierarchicalNodeParser(node_parser_ids=["markdown", "chunk_size_1024", "chunk_size_256"], node_parser_map={"markdown": MarkdownNodeParser(), ...SentenceSplitters})` and runs it on the prose; if the stripped prose is empty (mermaid-only file), the prose hierarchy is skipped entirely (no anchor leaf, no synthetic Document); (f) for each atomic chunk (or split chunk) appends a **standalone** `TextNode(text=<fence-chunk>, metadata={node_type:"mermaid_diagram", diagram_index:N, split_index:K, split_total:M, document_id, filename, document_type:"markdown"})` to the returned node list. **No PARENT/CHILD relationships are set on mermaid leaves** — they are orphan leaves so each fence surfaces as its own retrievable chunk. When no fences exist, only prose leaves are emitted; when a fence fits in one piece `split_index=1, split_total=1`.
- The `parse_result` path inside `hierarchical_chunk_text_task` is unchanged (xlsx/docx/pdf/pptx behaviour is byte-identical).
- `detect_format` routes extension `md` (only) directly to `['hierarchical_chunk_text']`. `.markdown` is not accepted — matches the upload endpoint's `VALID_EXTENSIONS` which only contains `.md`.
- DAG wiring adds a direct edge `branch >> hierarchical_chunk_text`. `hierarchical_chunk_text` gets `trigger_rule='none_failed_min_one_success'` so it fires whether its upstream is `select_parse_result` (other formats) or `branch` (md). `select_parse_result` and the four `parse_*` tasks are unchanged — they are skipped on the md branch and never produce output for md.
- For `.md`, the upload endpoint inserts `status="pending"` / `approval_status="pending"`. Approval-side DAG triggering is reused as-is.
- `cleanup_temp_files` patterns are unchanged — there is no parse-md pickle to clean up.

**Ask First:**
- Extracting code fences other than `mermaid` (e.g. `python`, `sql`) as atomic leaves.
- LLM-summarising each fence into prose before embedding (rejected for this PR — Option B in design notes).
- Changing chunk sizes for non-markdown formats.
- Auto-triggering the DAG at upload time (skipping the approval gate).

**Never:**
- Do not change behaviour for any non-markdown extension. Upload, parse, DAG routing, and chunking for `.xlsx/.docx/.pdf/.pptx/...` must be byte-identical.
- Do not import from `src.*` inside `airflow/plugins/api/upload.py`.
- Do not change the upload-endpoint response shape or remove the `validate_markdown_utf8` check.
- Do not embed any atomic leaf whose token count exceeds `settings.embedding_model_context_tokens` (default 8191 for Azure `text-embedding-3-large`). Oversize fences must be split along line boundaries before emission; if a single line still exceeds the limit, raise a clear error in `parse_markdown_document` — do not silently truncate.
- The embedding model's max input is configured via `EMBEDDING_MODEL_CONTEXT_TOKENS` in `.env.example` / `settings.embedding_model_context_tokens` (Pydantic Settings). Hardcoding `8191` anywhere in extraction code is forbidden; always read from settings.

## I/O & Edge-Case Matrix

| Scenario | Input / State | Expected Output / Behaviour | Error Handling |
|---|---|---|---|
| Upload happy path | `POST /api/v1/documents/upload` with valid `foo.md` | 202 `{document_id, filename, status:"pending"}`; DB `pending`/`pending`; S3 object exists; no DAG trigger | N/A |
| Approve `.md` | `POST /{doc_id}/approve` on pending `.md` | `document_extraction_dag` triggered with the usual conf; response `dag_triggered=true` | Existing handler |
| DAG run, no fences | `.md` with only prose | `branch` routes to `hierarchical_chunk_text` directly; parse_* and `select_parse_result` are skipped. Chunking reads the `.md` and produces only prose leaves. `store_text` indexes leaves; `update_document_status` writes `completed` | Existing terminal-resolution path |
| DAG run, with `mermaid` fences | `.md` with N mermaid fences, all under embed limit | Each fence becomes one **standalone orphan leaf** (no parent wiring). `node_type="mermaid_diagram"`, `split_index=split_total=1`. Prose chunked normally — fences are removed from prose (no placeholder), so the resulting leaves contain only the surrounding text. | As above |
| Fence over embed-model limit | A mermaid fence > `embedding_model_context_tokens` | `hierarchical_chunk_text` splits the fence along line boundaries into K pieces, each ≤ limit. K orphan leaves emitted sharing `diagram_index=N` and `split_index=1..K, split_total=K` (no parent wiring). DAG completes normally. | None — split is transparent |
| Mermaid-only file (no prose) | `.md` whose only content is one or more ` ```mermaid ``` ` fences | Prose hierarchy is skipped entirely. Only orphan mermaid leaves are emitted; no placeholder/anchor prose leaf is produced. | None |
| Non-`.md` file in mistakenly-wired branch | DAG passes `fetch_result_md` but `original_filename` is not `.md` (e.g. `.pdf`) | Task falls through to the `parse_result` branch as if `fetch_result_md` were `None`. No UTF-8 decode of binary content. | N/A — gate prevents misuse |
| Single mermaid line over limit | One line in the fence > limit on its own | `hierarchical_chunk_text` raises `ValueError("Mermaid fence N: single line exceeds embedding model context")`. DAG fails chunk stage. | Raised in `hierarchical_chunk_text_task` (md branch) |
| Empty / whitespace-only `.md` | No prose, no fences | `hierarchical_chunk_text` raises `ValueError("Markdown file is empty")` | Raised in `hierarchical_chunk_text_task` (md branch) |
| Non-UTF-8 bytes reach DAG | Invalid bytes (upload should catch first) | `hierarchical_chunk_text` raises `UnicodeDecodeError`; terminal status `failed` | Existing failure path |
| Existing formats | Any `.xlsx/.docx/.pdf/.pptx` upload + approve | Byte-identical to today | Unchanged |

</frozen-after-approval>

## Code Map

- `.env.example` / `src/core/config.py` — add `EMBEDDING_MODEL_CONTEXT_TOKENS` env var and `Settings.embedding_model_context_tokens: int = 8191`.
- `airflow/plugins/api/upload.py` — remove `is_markdown` terminal-status special-case so `.md` inserts `pending`/`pending`; keep `validate_markdown_utf8` and the skip of magic-bytes/temp-file/password checks.
- `airflow/dags/tasks/processing_tasks.py` — add `fetch_result_md: dict = None` parameter to `hierarchical_chunk_text_task`. The markdown branch fires only when `fetch_result_md is not None and original_filename.lower().endswith(".md")` (the DAG wires this kwarg unconditionally so the task itself must gate). When the branch fires, the task reads the `.md` directly, extracts mermaid fences (with split-on-overflow), removes them from prose (no placeholder), and appends each fence as a standalone orphan mermaid leaf (no parent wiring). Empty-prose files skip the prose hierarchy entirely. When the gate doesn't match, existing `parse_result` behaviour is byte-identical.
- `airflow/dags/document_extraction_dag.py` — add `md` branch in `detect_format` that returns `['hierarchical_chunk_text']`; pass `fetch_result_md=fetch.output` to `hierarchical_chunk_text_task.override(...)`; change its `trigger_rule` to `'none_failed_min_one_success'`; add direct edge `branch >> hierarchical_chunk_text`. No new parse task, no changes to `select_parse_result`, no changes to `cleanup_temp_files`.

## Tasks & Acceptance

**Execution:**
- [x] `.env.example` / `src/core/config.py` -- Add `EMBEDDING_MODEL_CONTEXT_TOKENS=8191` to `.env.example` (in the Embedding Provider Selection block) and `embedding_model_context_tokens: int = 8191` to `Settings` in `src/core/config.py`.
- [x] `airflow/plugins/api/upload.py` -- Remove the `row_status = "completed" if is_markdown else "pending"` / `row_approval_status = "approved" if is_markdown else "pending"` branch so `.md` rows insert with `"pending"`/`"pending"`. Response naturally returns `status="pending"`. Keep the upstream `is_markdown` branch (UTF-8 validation, skip magic-bytes/password).
- [x] `airflow/dags/tasks/processing_tasks.py` -- Add an optional `fetch_result_md: dict = None` parameter to `hierarchical_chunk_text_task`. Gate the markdown branch on `fetch_result_md is not None and (fetch_result_md.get("original_filename") or "").lower().endswith(".md")`; the DAG passes `fetch_result_md=fetch.output` unconditionally, so this extension check is what keeps PDFs/Word/etc out of the markdown path. When the gate doesn't match: existing `parse_result`-driven path (no change). When it matches: (1) read `fetch_result_md['local_path']` as bytes, decode UTF-8 (let `UnicodeDecodeError` propagate); (2) read embedding limit via `Settings().embedding_model_context_tokens`; (3) scan for ` ```mermaid\n...\n``` ` fences using regex `r"```mermaid\b[^\n]*\n(.*?)\n```"` (DOTALL); for each fence, tokenize with `tiktoken.get_encoding("cl100k_base")` — if tokens ≤ limit, emit a single chunk; otherwise call an inline `_split_fence(raw, limit)` helper that splits along line boundaries, re-wraps each piece with ` ```mermaid ... ``` `, and emits K chunks sharing the same `diagram_index` with `split_index/split_total` set; if any single line still exceeds the limit, raise `ValueError(f"Mermaid fence {N}: single line exceeds embedding model context")`; (4) **remove each fence from the prose entirely** (the regex sub callback returns `""`, not a `[MERMAID_DIAGRAM_N]` placeholder); (5) if the stripped prose is empty AND there are no atomic chunks, raise `ValueError("Markdown file is empty")`; (6) if the stripped prose is non-empty, build a single `Document(text=<prose>, metadata={"filename": original_filename})` and the parser `HierarchicalNodeParser(node_parser_ids=["markdown", "chunk_size_1024", "chunk_size_256"], node_parser_map={"markdown": MarkdownNodeParser(), "chunk_size_1024": SentenceSplitter(chunk_size=1024, chunk_overlap=204, tokenizer=tokenizer, paragraph_separator="\n\n"), "chunk_size_256": SentenceSplitter(chunk_size=256, chunk_overlap=51, tokenizer=tokenizer, paragraph_separator="\n\n")}); otherwise (mermaid-only file) start with an empty nodes list and skip the parser entirely; (7) for each atomic chunk: build `TextNode(text=<chunk>, metadata={"node_type":"mermaid_diagram", "diagram_index":N, "split_index":K, "split_total":M, "document_id":document_id, "filename":original_filename, "document_type":"markdown", "page_number":1})` and append to the nodes list — **do not set PARENT/CHILD relationships**; mermaid leaves are orphan/standalone so each fence surfaces as its own retrievable chunk. (8) The output pickle/XCom shape matches the existing `hierarchical_chunk_text_task` contract (`store_to_llamaindex_task` consumes it unchanged).
- [x] `airflow/dags/document_extraction_dag.py` -- In `detect_format`, add `elif ext == 'md': return ['hierarchical_chunk_text']` before `else`. Change the `hierarchical_chunk_text_task.override(...)` call: add `trigger_rule='none_failed_min_one_success'`; pass `fetch_result_md=fetch.output` in addition to existing `parse_result=text_parse_result`. Add the edge `branch >> hierarchical_chunk_text` (direct, parallel to the existing `text_parse_result >> hierarchical_chunk_text`). Do not touch `select_parse_result` or `cleanup_temp_files`.

**Acceptance Criteria:**
- Given a valid `.md` upload, when `POST /upload` is called, then the response is 202 with `status="pending"` and the DB row is `pending`/`pending`.
- Given a pending `.md`, when `POST /{doc_id}/approve` is called, then `document_extraction_dag` is triggered and the response includes `dag_triggered=true`.
- Given an approved `.md` with N mermaid fences (all within embed limit), when the DAG completes, then Milvus contains N orphan leaf vectors with `node_type="mermaid_diagram"` (full fence body, `split_total=1`, no `parent_id`) plus M prose leaves from the `MarkdownNodeParser` hierarchy. Prose leaves do not contain `[MERMAID_DIAGRAM_N]` placeholders. `documents.status='completed'`; mermaid leaves surface directly in retrieval (AutoMergingRetriever still merges sibling prose leaves; mermaid leaves are not eligible for merging by design).
- Given an approved `.md` whose mermaid fence exceeds `embedding_model_context_tokens`, when the DAG completes, then the fence is split into K orphan leaves all sharing `diagram_index=N`, with `split_index/split_total` set (no `parent_id`). All leaves embed successfully. `documents.status='completed'`.
- Given an approved mermaid-only `.md` (no prose outside fences), when the DAG completes, then Milvus contains only mermaid leaves — no placeholder/anchor prose chunk is produced.
- Given an approved non-`.md` upload (e.g. `.pdf`), the markdown branch must not trigger inside `hierarchical_chunk_text_task` (no UTF-8 decode of binary content) — verified by the extension gate.
- Given a `.md` whose mermaid fence contains a single line exceeding the embedding limit, when `hierarchical_chunk_text` runs, then it raises `ValueError` and `update_document_status` writes `failed`.
- Given an approved `.md` with no fences, when the DAG completes, then only prose leaves are indexed; no atomic mermaid leaves are produced.
- Given an approved `.md`, when the DAG runs, then no `parse_*` task and no `select_parse_result` task runs for it (all skipped via `detect_format` branching) — only `validate → fetch → save_document_metadata → detect_format → hierarchical_chunk_text → ...` executes.
- Given any non-markdown upload, when uploaded and approved, then behaviour is byte-identical to today's DAG run.

## Design Notes

`MarkdownNodeParser` splits the prose by H1/H2/... headers, producing one top-level node per section. Verified on `BK20190313.md`: 6 clean sections at the top (sizes 26 / 189 / 212 / 3465 / 107 / 78 tokens), prose sections smaller than 256 tokens pass through `SentenceSplitter` unchanged. `HierarchicalNodeParser` supports heterogeneous `node_parser_map` values (verified in llama-index-core 0.14.13 source `relational/hierarchical.py`); parent/child relationships are wired across levels so AutoMergingRetriever works on the prose side.

**Orphan mermaid leaves (revised from original design).** An earlier iteration wired each mermaid leaf as a child of the mid-level parent that owned the `[MERMAID_DIAGRAM_N]` placeholder leaf, so AutoMergingRetriever could collapse sibling prose + mermaid into a single rich chunk. Empirically this backfired: the merger collapsed retrieved siblings into a parent whose visible text was the placeholder string itself (`## フロー図 (Flow Diagram)\n\n[MERMAID_DIAGRAM_1]`, 43 chars), silently dropping the 7000-char diagram body. We now (a) **drop the fence from prose entirely** (no placeholder), and (b) **emit each mermaid chunk as a standalone orphan leaf** with no `PARENT`/`CHILD` relationships. Consequences: each diagram surfaces as its own retrievable chunk in the top-k; AutoMergingRetriever cannot collapse it (which is exactly what we want); the placeholder-only prose chunk is gone, so retrieval is no longer polluted by 22–43 char near-empty leaves. The whole fence is embedded as a single vector; `text-embedding-3-large` accepts 8191 input tokens, which comfortably covers the `BK20190313` diagram (3465 tokens).

**Extension gate inside the task.** Airflow XCom kwargs are wired at DAG-parse time; the DAG cannot conditionally omit `fetch_result_md`, so it is passed unconditionally to `hierarchical_chunk_text_task`. To prevent the markdown branch from running for non-`.md` uploads (which would attempt to UTF-8-decode binary PDF/Word bytes and crash with `UnicodeDecodeError: 'utf-8' codec can't decode byte 0x8f`), the task itself gates the branch on `original_filename.lower().endswith(".md")`. Non-`.md` files fall through to the existing `parse_result` path as if `fetch_result_md` were `None`.

Soft trade-off: embedding raw Mermaid DSL retrieves less reliably from natural-language queries than a prose summary would (Option B) — accepted for this PR. Follow-up spec can add an LLM-describe pass that stores both the prose summary (embedded) and the raw fence (metadata) if retrieval quality on diagrams proves insufficient.

## Verification

**Commands:**
- `python scripts/test_upload_endpoint.py BK20190313.md` then `python scripts/test_approval_endpoint.py approve <doc-id>` — end-to-end DAG run for a normal md+mermaid file. Expected: hierarchy has 8 prose leaves + 1 orphan mermaid leaf; no `[MERMAID_DIAGRAM_1]` placeholder chunk in Milvus.
- `python scripts/test_upload_endpoint.py BK20190313_mm_only.md` then approve — end-to-end DAG run for a mermaid-only file. Expected: hierarchy has 0 prose leaves + 1 orphan mermaid leaf; no anchor/placeholder leaf.
- `python scripts/test_automerge_mermaid.py --doc-id <doc-id> --top-k 20` — exercise AutoMergingRetriever against the indexed doc; mermaid leaves should surface as their own rows (`type=mermaid_diagram`, `chars≈7096` for BK20190313), prose sibling hits should still collapse to merged parents.

## Suggested Review Order

**DAG routing for `.md`**

- `.md` extension routes directly to `hierarchical_chunk_text`, skipping all parse tasks.
  [`document_extraction_dag.py:121`](../../airflow/dags/document_extraction_dag.py#L121)

- `fetch_result_md=fetch.output` is wired unconditionally; `trigger_rule` becomes `none_failed_min_one_success`.
  [`document_extraction_dag.py:512`](../../airflow/dags/document_extraction_dag.py#L512)

- Direct edge `branch >> hierarchical_chunk_text` complements the existing text path.
  [`document_extraction_dag.py:607`](../../airflow/dags/document_extraction_dag.py#L607)

**Markdown chunking branch (orphan mermaid, no placeholder)**

- Extension gate prevents PDFs/Word from entering the markdown UTF-8 path.
  [`processing_tasks.py:451`](../../airflow/dags/tasks/processing_tasks.py#L451)

- Mermaid fences are removed from prose entirely (regex sub returns `""`).
  [`processing_tasks.py:474`](../../airflow/dags/tasks/processing_tasks.py#L474)

- Oversize fence splitter (line-boundary, respects `EMBEDDING_MODEL_CONTEXT_TOKENS`).
  [`processing_tasks.py:488`](../../airflow/dags/tasks/processing_tasks.py#L488)

- Prose hierarchy is built only when prose is non-empty (mermaid-only files skip it).
  [`processing_tasks.py:539`](../../airflow/dags/tasks/processing_tasks.py#L539)

- Each mermaid fence appended as a standalone orphan `TextNode` — no `PARENT`/`CHILD` wiring.
  [`processing_tasks.py:575`](../../airflow/dags/tasks/processing_tasks.py#L575)

**Upload endpoint + config**

- `.md` rows now insert with `pending`/`pending`, reusing the approval-gated DAG trigger.
  [`upload.py:199`](../../airflow/plugins/api/upload.py#L199)

- `embedding_model_context_tokens` setting (default 8191) gates fence splitting.
  [`config.py:42`](../../src/core/config.py#L42)

- `.env.example` documents the new `EMBEDDING_MODEL_CONTEXT_TOKENS` knob.
  [`.env.example:121`](../../.env.example#L121)
- `docker compose -f docker/docker-compose.yml -f docker/docker-compose.airflow.yml exec airflow-webserver python -c "from tasks.processing_tasks import hierarchical_chunk_text_task; print('import ok')"` -- expected: `import ok`.
- `docker compose -f docker/docker-compose.yml -f docker/docker-compose.airflow.yml exec airflow-webserver airflow dags list 2>&1 | grep document_extraction_dag` -- expected: DAG listed, no import errors.

**Manual checks:**
- `python scripts/test_upload_endpoint.py BK20190313.md` — expect response `status="pending"`. Then call `POST /api/v1/documents/{doc_id}/approve` with HMAC; expect `dag_triggered=true`.
- Watch the Airflow UI for one run of `document_extraction_dag` on the `.md`. Verify task graph: `validate → fetch → save_document_metadata → detect_format → hierarchical_chunk_text → extract_domain_terms_text → tag_records_text → store_text_llamaindex → resolve_terminal_outcome → update_document_status → cleanup_temp_files`. All `parse_*` tasks and `select_parse_result` skipped on the md branch.
- After success: query Milvus for `document_id`, confirm at least one chunk has `metadata.node_type == "mermaid_diagram"` whose `text` contains the full ` ```mermaid ... ``` ` fence; confirm prose leaves exist with `document_type == "markdown"`.
- Regression: upload + approve an `.xlsx` end-to-end; confirm the parallel Excel pipelines still run and produce the same final status as before.