Add Airflow UI plugin: Blueprint tab showing YAML and blueprint code

[jlaneve]

What

Airflow's Code tab shows the loader file that builds YAML DAGs, not the YAML itself. This adds a UI plugin, registered automatically via the airflow.plugins entry point, that puts a Blueprint tab on dag, dag run, task, and task instance pages in Airflow 3.1+:

• dag / dag run: the source YAML the DAG was built from, with each blueprint's Python underneath — syntax highlighted with line numbers, like the Code tab
• task / task instance: that step's config and blueprint source

How it resolves what to show

build_all_airflow_dags() now stamps each DAG with a blueprint:<yaml path> tag (opt out with source_tags=False) and records the path in each task's blueprint_step_config. Both survive DAG serialization, so the plugin needs one lookup instead of re-scanning the dags folder.

Task-level content walks a ladder and labels which source you're seeing:

1. the run's rendered fields — the exact config the task ran with, param overrides included
2. the serialized DAG — config as built
3. the YAML / blueprint file on disk

Values Airflow truncated under [core] max_templated_field_length (blueprint source files usually exceed it) fall through to the next source instead of showing the placeholder.

Notes

• The pages are served by a FastAPI app mounted at /blueprint under the API server and render inside Airflow's sandboxed iframes: links drive the parent SPA's router through its history API (the sandbox blocks target="_top"), and the pages follow Airflow's light/dark toggle.
• Task group pages render task-destination views but leave {TASK_ID} unsubstituted; the group id is recovered from the Referer, so groups get the tab too.
• On Airflow 2 the plugin loads but registers nothing.
• Adds pygments as a direct dependency (already present transitively via rich).

Testing

• 307 unit tests (tests/test_plugin.py, tests/test_builder.py additions)
• 92 integration tests against Astro Runtime 3.1, including a triggered run asserting as-run provenance
• Playwright click-through of all four tab levels in the advanced example

🤖 Generated with Claude Code

https://claude.ai/code/session_01TQnbVv8WAkqMzwKbQsbBER

[github-actions]

Claude finished @jlaneve's task in 2m 39s —— View job

---

Code Review: Add Airflow UI plugin

Tasks:

• Gather context and understand the PR
• Read all changed files
• Synthesize findings

---

This is a solid, well-structured implementation. The provenance ladder (RTIF → serialized DAG → YAML on disk) is a clever design, and the graceful-degradation pattern throughout is consistent. Below are findings ranked by severity.

---

Security

Path traversal guard is correct but load-bearing — resolve_dag_source (app.py:195–198) correctly uses is_relative_to to prevent arbitrary file reads when a blueprint:<path> tag is present. One subtlety: the candidate.resolve() call follows symlinks, so a symlink inside the dags folder pointing outside it would still pass the check. If symlinks are common in your deployment, consider an explicit is_symlink() guard.

XSS via | safe — the templates mark Pygments-rendered HTML as safe, which is correct. Autoescape is on for everything else (jinja2.Environment(autoescape=True) in create_app). No issue here.

dag_id unencoded in task_step.html — line 13 of task_step.html:

<a href="/blueprint/dags/{{ dag_id }}/yaml"
  >

Jinja2's autoescaping HTML-encodes the value (so < → <), but does not URL-encode it. A DAG ID containing # or ? would produce a broken link. DAG IDs are validated by Airflow but the constraints aren't strict enough to rule this out. Fix this →

---

Correctness

Fragile "Truncated." heuristic — _clean_stamped_value (app.py:273) detects Airflow's truncation placeholder with value.startswith("Truncated."). If Airflow changes the placeholder text (it's not a stable API), this check will silently stop working and the truncated placeholder will be rendered to users instead of falling through to the next source. Worth a comment noting the dependency on this string, and possibly a regression test that pins the Airflow version where this was observed.

**"+" → space run_id workaround** — app.py:425–430handles the Airflow UI not URL-encoding{RUN_ID}before substituting it, so a+ in timestamped run IDs arrives as a space. This is well-tested (test_run_id_plus_recovered_from_space`) and the comment explains it clearly. The retry approach (try space first, then convert) is reasonable, though it means a legitimate run_id containing a literal space would also trigger the retry unnecessarily. Low risk, no action needed.

"{" in task_id relies on undocumented Airflow behavior — _normalize_task_ref (app.py:390) detects unsubstituted task group pages by looking for { in the task_id query param. This works because Airflow currently leaves {TASK_ID} literally unsubstituted on group pages, but this is not a documented contract. A future Airflow version that substitutes the group ID here would silently break this recovery path. Hard to avoid given the constraint, but worth a comment.

_collect_operators only recurses into TaskGroup children — builder.py:446–457. The children attribute of a TaskGroup can also contain other node types in Airflow 3 (mapped task groups, etc.). If a Blueprint renders anything other than BaseOperator or TaskGroup, _inject_step_context silently skips it. This is probably fine for current Blueprint usage but could cause silent failures as Airflow 3 APIs evolve.

_cache is not thread-safe — app.py:37–45. FastAPI runs handlers in an asyncio event loop (thread-safe for single-thread async), but Uvicorn with workers > 1 spins up separate processes (each with their own cache), not threads — so no race condition there. However, if the API server ever runs with threading, the dict read/write is not atomic. A threading.Lock or functools.lru_cache would be safer. Low-priority given current Airflow deployment patterns.

---

Code Quality

Private function imported across module boundary — app.py:23:

from blueprint.builder import SOURCE_TAG_PREFIX, _relative_source

_relative_source is a private function (underscore prefix). Importing it into app.py creates a fragile coupling — if it's refactored or moved, the import breaks without any IDE or linter warning. Either make it public (relative_source) or move it to a shared utils.py. SOURCE_TAG_PREFIX as a constant is fine to import. Fix this →

Stale docstring reference — _stamped_step_fields docstring (app.py:412) refers to _clean_rtif_value but the actual function is named _clean_stamped_value. Fix this →

Module-level create_app() at import time — plugin/__init__.py:89 calls _build_surfaces() at module import time, which in turn calls create_app(). Any non-ImportError exception in create_app during plugin loading would propagate and break Airflow startup. The ImportError catch is too narrow — consider catching Exception (or logging and returning [], []) so a bug in app setup doesn't take down the plugin system entirely.

import inspect inside dag_code_sections — app.py:350. inspect is a stdlib module; putting it at the top-level import would be cleaner and more consistent with the rest of the file.

Unbounded _cache growth — the cache at app.py:34 has no maximum size. In practice the keys are bounded (sources:{folder} and registry:{folder}), so this is fine. Worth a comment to clarify.

---

Performance

Double-parsing YAML in dag_yaml_page — app.py:583–587:

content_html = code_html(path.read_text(encoding="utf-8"), "yaml", ...)
code_sections = dag_code_sections(_parse_dag_yaml(path), folder)

path.read_text() is called once for rendering, and _parse_dag_yaml(path) calls path.read_text() again internally. These could be combined. Minor since the result is cached for 30 seconds, but easy to clean up.

---

Test Coverage

Coverage is excellent overall. A few gaps worth noting:

• code_html Pygments fallback — the except Exception branch that returns a raw <pre> is not explicitly tested. A test with pygments monkeypatched to raise would cover this.
• group_from_referer unit tests — this function is exercised indirectly via test_group_page_recovers_step_from_referer and test_group_page_referer_with_run, but the URL parsing cases (malformed Referer, Referer without a run) aren't tested directly.
• _apply_source_tag when dag.tags is None — builder.py:528 does {*(dag.tags or ()), tag} which handles None. This is tested by test_source_tag_added, but it's worth confirming dag.tags starts as None vs [] in the test fixture.

---

Summary

Severity	Issue
Should fix	dag_id unencoded in task_step.html href
Should fix	_relative_source private function imported across module boundary
Low	Stale docstring reference (_clean_rtif_value)
Low	ImportError-only catch in _build_surfaces() too narrow
Low	Double YAML read in dag_yaml_page
FYI	startswith("Truncated.") is fragile heuristic
FYI	{" in task_id relies on undocumented Airflow behavior

The architecture is clean — the tag-first / scan-fallback resolution, the independent config and code provenance tracking, and the sandboxed iframe navigation approach are all well-considered. The builder changes (source_tags, _inject_step_context, _apply_source_tag) are minimal and don't break existing behavior.