Add AI Trace observability panel to the common.ai provider

.pre-commit-config.yaml

@@ -783,6 +783,7 @@ repos:
           ^airflow-core/src/airflow/ui/openapi-gen/|
           ^airflow-core/src/airflow/ui/pnpm-lock\.yaml$|
           ^providers/common/ai/src/airflow/providers/common/ai/plugins/www/pnpm-lock\.yaml$|
+          ^providers/common/ai/src/airflow/providers/common/ai/plugins/ai_trace_www/pnpm-lock\.yaml$|
           ^airflow-core/src/airflow/ui/public/i18n/locales/de/README\.md$|
           ^airflow-core/src/airflow/ui/src/i18n/config\.ts$|
           ^\.agents/skills/airflow-translations/|

providers/common/ai/.pre-commit-config.yaml

@@ -43,7 +43,10 @@ repos:
         name: Compile Common AI provider assets
         language: node
         stages: ['pre-commit', 'manual']
-        files: ^src/airflow/providers/common/ai/plugins/www/
+        files: |
+          (?x)
+          ^src/airflow/providers/common/ai/plugins/www/|
+          ^src/airflow/providers/common/ai/plugins/ai_trace_www/
         entry: ../../../scripts/ci/prek/compile_provider_assets.py ai
         pass_filenames: false
         additional_dependencies: ['pnpm@10.25.0']

providers/common/ai/hatch_build.py

@@ -47,10 +47,12 @@ def clean(self, directory: str, versions: Iterable[str]) -> None:
         work_dir = Path(self.root)
         log.warning("Cleaning generated files in directory: %s", work_dir)
         ai_package_src = work_dir / "src" / "airflow" / "providers" / "common" / "ai"
-        ai_ui_path = ai_package_src / "plugins" / "www"
-        self.clean_dir(ai_ui_path / ".pnpm-store")
-        self.clean_dir(ai_ui_path / "dist")
-        self.clean_dir(ai_ui_path / "node_modules")
+        # common.ai ships two plugin bundles (HITL Review + AI Trace); clean both.
+        for bundle in ("www", "ai_trace_www"):
+            ai_ui_path = ai_package_src / "plugins" / bundle
+            self.clean_dir(ai_ui_path / ".pnpm-store")
+            self.clean_dir(ai_ui_path / "dist")
+            self.clean_dir(ai_ui_path / "node_modules")
         (work_dir / "www-hash.txt").unlink(missing_ok=True)
 
     def get_version_api(self) -> dict[str, Callable[..., str]]:

providers/common/ai/provider.yaml

@@ -80,6 +80,8 @@ hooks:
 plugins:
   - name: hitl_review
     plugin-class: airflow.providers.common.ai.plugins.hitl_review.HITLReviewPlugin
+  - name: ai_trace
+    plugin-class: airflow.providers.common.ai.plugins.ai_trace.AITracePlugin
 
 
 config:
@@ -140,6 +142,46 @@ config:
         type: boolean
         example: "False"
         default: "False"
+      trace_store_path:
+        description: |
+          ObjectStorage URI for the backend-free trace store -- local dev and
+          quick testing without Langfuse or any OTLP backend. When set, GenAI
+          spans from ``AgentOperator`` / ``@task.agent`` / ``@task.llm`` are
+          written as standard OTLP JSON lines (one file per task-instance try)
+          under this path, and the "AI Trace" UI reads them back directly.
+          Requires neither ``[traces] otel_on`` nor ``otel_export_enabled``.
+          Any scheme supported by ``airflow.sdk.ObjectStoragePath`` is accepted
+          (``file://``, ``s3://``, ``gs://``, ...); the API server needs read
+          access to the same path. Files are plain OTLP JSON, so a collector's
+          ``otlpjsonfilereceiver`` can replay them into a real backend later.
+          Takes precedence over the Langfuse read path when set. Content
+          capture is still gated by ``capture_content``. Costs shown in the
+          UI are read-time estimates from the ``genai-prices`` library's
+          bundled price data (no runtime price fetch) -- approximate, not
+          billing data; unknown or self-hosted models show no cost. No
+          retention is applied; clean the path yourself.
+
+          Spike-status: proof of concept, not a stable layout.
+        version_added: 0.5.0
+        type: string
+        example: "file:///tmp/airflow_ai_traces"
+        default: ""
+      trace_backend_conn_id:
+        description: |
+          Airflow connection used by the "AI Trace" task instance panel to read
+          a trace back from your tracing backend (Langfuse today) once
+          ``otel_export_enabled`` has emitted it. The panel resolves a task
+          instance's trace_id from its own span context, so no data is written
+          anywhere new; this connection is read-only credentials for looking a
+          trace back up. Connection ``host`` is the backend's base URL,
+          ``login``/``password`` are its API key pair.
+
+          Spike-status: this is a proof of concept, not a stable API. The
+          panel only supports Langfuse today.
+        version_added: 0.5.0
+        type: string
+        example: "langfuse_default"
+        default: "langfuse_default"
 
 
 connection-types:

providers/common/ai/pyproject.toml

@@ -183,6 +183,7 @@ provider_info = "airflow.providers.common.ai.get_provider_info:get_provider_info
 
 [project.entry-points."airflow.plugins"]
 hitl_review = "airflow.providers.common.ai.plugins.hitl_review:HITLReviewPlugin"
+ai_trace = "airflow.providers.common.ai.plugins.ai_trace:AITracePlugin"
 
 [tool.hatch.version]
 path = "src/airflow/providers/common/ai/__init__.py"
@@ -199,23 +200,29 @@ exclude = [
     "src/airflow/providers/__init__.py",
     "src/airflow/providers/common/ai/plugins/www/.pnpm-store",
     "src/airflow/providers/common/ai/plugins/www/node_modules",
+    "src/airflow/providers/common/ai/plugins/ai_trace_www/.pnpm-store",
+    "src/airflow/providers/common/ai/plugins/ai_trace_www/node_modules",
 ]
 
 [tool.hatch.build.targets.custom]
 path = "./hatch_build.py"
 
 artifacts = [
     "src/airflow/providers/common/ai/plugins/www/dist",
+    "src/airflow/providers/common/ai/plugins/ai_trace_www/dist",
 ]
 
 [tool.hatch.build.targets.wheel]
 packages = ['src/airflow']
 artifacts = [
     "src/airflow/providers/common/ai/plugins/www/dist",
+    "src/airflow/providers/common/ai/plugins/ai_trace_www/dist",
 ]
 exclude = [
     "src/airflow/__init__.py",
     "src/airflow/providers/__init__.py",
     "src/airflow/providers/common/ai/plugins/www/.pnpm-store",
     "src/airflow/providers/common/ai/plugins/www/node_modules",
+    "src/airflow/providers/common/ai/plugins/ai_trace_www/.pnpm-store",
+    "src/airflow/providers/common/ai/plugins/ai_trace_www/node_modules",
 ]

providers/common/ai/src/airflow/providers/common/ai/_otlp_json.py

@@ -0,0 +1,172 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Minimal OTLP/JSON span encoder for the ``[common.ai] trace_store_path`` store.
+
+Writes one OTLP ``TracesData`` JSON object per line, following the OTLP
+JSON-Protobuf encoding rules (https://opentelemetry.io/docs/specs/otlp/):
+trace/span ids hex-encoded, 64-bit integers as decimal strings, enums as
+integers, camelCase field names. That makes the files replayable into any real
+backend later via an OpenTelemetry Collector's ``otlpjsonfilereceiver``.
+
+This is vendored rather than depended on: the upstream
+``opentelemetry-exporter-otlp-json-file`` exporter (0.64b0, the only release)
+is uninstallable from PyPI -- its ``opentelemetry-proto-json`` dependency was
+never published (verified 2026-07-04). Swap this module for the official
+``FileSpanExporter(stream=...)`` once a fixed upstream release exists.
+
+Only imported when the trace store is configured, so the OpenTelemetry SDK
+import below never taxes the tracing-off path.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+from collections.abc import Mapping, Sequence
+from typing import IO, TYPE_CHECKING, Any
+
+from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+from opentelemetry.trace import StatusCode
+
+if TYPE_CHECKING:
+    from opentelemetry.sdk.trace import ReadableSpan
+
+log = logging.getLogger(__name__)
+
+# SDK enum value -> OTLP wire value (OTLP reserves 0 for UNSPECIFIED, so the
+# SDK's INTERNAL=0..CONSUMER=4 shift up by one).
+_OTLP_STATUS = {StatusCode.UNSET: 0, StatusCode.OK: 1, StatusCode.ERROR: 2}
+
+
+def _any_value(value: Any) -> dict[str, Any]:
+    # bool before int: bool is an int subclass.
+    if isinstance(value, bool):
+        return {"boolValue": value}
+    if isinstance(value, int):
+        return {"intValue": str(value)}
+    if isinstance(value, float):
+        return {"doubleValue": value}
+    if isinstance(value, (bytes, bytearray)):
+        return {"bytesValue": base64.b64encode(value).decode()}
+    if isinstance(value, Mapping):
+        return {"kvlistValue": {"values": _key_values(value)}}
+    if isinstance(value, Sequence) and not isinstance(value, str):
+        return {"arrayValue": {"values": [_any_value(v) for v in value]}}
+    return {"stringValue": str(value)}
+
+
+def _key_values(attributes: Mapping[str, Any]) -> list[dict[str, Any]]:
+    return [{"key": k, "value": _any_value(v)} for k, v in attributes.items()]
+
+
+def _encode_span(span: ReadableSpan) -> dict[str, Any]:
+    ctx = span.get_span_context()
+    out: dict[str, Any] = {
+        "traceId": format(ctx.trace_id if ctx else 0, "032x"),
+        "spanId": format(ctx.span_id if ctx else 0, "016x"),
+        "name": span.name,
+        "kind": (span.kind.value if span.kind is not None else 0) + 1,
+        "startTimeUnixNano": str(span.start_time or 0),
+        "endTimeUnixNano": str(span.end_time or 0),
+        "attributes": _key_values(span.attributes or {}),
+        "status": {},
+    }
+    if span.parent is not None:
+        out["parentSpanId"] = format(span.parent.span_id, "016x")
+    if span.events:
+        # Exception details land here via Span.record_exception.
+        out["events"] = [
+            {
+                "timeUnixNano": str(event.timestamp),
+                "name": event.name,
+                "attributes": _key_values(event.attributes or {}),
+            }
+            for event in span.events
+        ]
+    if span.status is not None:
+        code = _OTLP_STATUS.get(span.status.status_code, 0)
+        if code:
+            out["status"]["code"] = code
+        if span.status.description:
+            out["status"]["message"] = span.status.description
+    return out
+
+
+def encode_traces_data(spans: Sequence[ReadableSpan]) -> dict[str, Any]:
+    """Encode finished spans as one OTLP ``TracesData`` object, grouped by resource/scope."""
+    grouped: dict[Any, dict[Any, list[dict[str, Any]]]] = {}
+    resources: dict[Any, Any] = {}
+    scopes: dict[Any, Any] = {}
+    for span in spans:
+        resource, scope = span.resource, span.instrumentation_scope
+        resources[id(resource)] = resource
+        scopes[id(scope)] = scope
+        grouped.setdefault(id(resource), {}).setdefault(id(scope), []).append(_encode_span(span))
+    return {
+        "resourceSpans": [
+            {
+                "resource": {"attributes": _key_values(getattr(resources[rid], "attributes", None) or {})},
+                "scopeSpans": [
+                    {
+                        "scope": {
+                            "name": getattr(scopes[sid], "name", "") or "",
+                            **(
+                                {"version": scopes[sid].version}
+                                if getattr(scopes[sid], "version", None)
+                                else {}
+                            ),
+                        },
+                        "spans": scope_spans,
+                    }
+                    for sid, scope_spans in by_scope.items()
+                ],
+            }
+            for rid, by_scope in grouped.items()
+        ]
+    }
+
+
+class OTLPJsonStreamExporter(SpanExporter):
+    """
+    Write each export batch as one OTLP/JSON ``TracesData`` line to a text stream.
+
+    Flushes after every line so spans are readable while the task still runs.
+    On ``file://`` this makes each finished span immediately visible; object
+    stores (s3/gcs) buffer locally and upload on close, so their content lands
+    at task-process exit -- acceptable for a dev/test store.
+    """
+
+    def __init__(self, stream: IO[str]) -> None:
+        self._stream = stream
+
+    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
+        try:
+            self._stream.write(json.dumps(encode_traces_data(spans), separators=(",", ":")) + "\n")
+            self._stream.flush()
+        except Exception:
+            log.exception("Failed to write %d span(s) to the common.ai trace store", len(spans))
+            return SpanExportResult.FAILURE
+        return SpanExportResult.SUCCESS
+
+    def force_flush(self, timeout_millis: int = 30_000) -> bool:
+        self._stream.flush()
+        return True
+
+    def shutdown(self) -> None:
+        self._stream.flush()

providers/common/ai/src/airflow/providers/common/ai/get_provider_info.py

@@ -85,7 +85,11 @@ def get_provider_info():
             {
                 "name": "hitl_review",
                 "plugin-class": "airflow.providers.common.ai.plugins.hitl_review.HITLReviewPlugin",
-            }
+            },
+            {
+                "name": "ai_trace",
+                "plugin-class": "airflow.providers.common.ai.plugins.ai_trace.AITracePlugin",
+            },
         ],
         "config": {
             "common.ai": {
@@ -112,6 +116,20 @@ def get_provider_info():
                         "example": "False",
                         "default": "False",
                     },
+                    "trace_store_path": {
+                        "description": "ObjectStorage URI for the backend-free trace store -- local dev and\nquick testing without Langfuse or any OTLP backend. When set, GenAI\nspans from ``AgentOperator`` / ``@task.agent`` / ``@task.llm`` are\nwritten as standard OTLP JSON lines (one file per task-instance try)\nunder this path, and the \"AI Trace\" UI reads them back directly.\nRequires neither ``[traces] otel_on`` nor ``otel_export_enabled``.\nAny scheme supported by ``airflow.sdk.ObjectStoragePath`` is accepted\n(``file://``, ``s3://``, ``gs://``, ...); the API server needs read\naccess to the same path. Files are plain OTLP JSON, so a collector's\n``otlpjsonfilereceiver`` can replay them into a real backend later.\nTakes precedence over the Langfuse read path when set. Content\ncapture is still gated by ``capture_content``. Costs shown in the\nUI are read-time estimates from the ``genai-prices`` library's\nbundled price data (no runtime price fetch) -- approximate, not\nbilling data; unknown or self-hosted models show no cost. No\nretention is applied; clean the path yourself.\n\nSpike-status: proof of concept, not a stable layout.\n",
+                        "version_added": "0.5.0",
+                        "type": "string",
+                        "example": "file:///tmp/airflow_ai_traces",
+                        "default": "",
+                    },
+                    "trace_backend_conn_id": {
+                        "description": "Airflow connection used by the \"AI Trace\" task instance panel to read\na trace back from your tracing backend (Langfuse today) once\n``otel_export_enabled`` has emitted it. The panel resolves a task\ninstance's trace_id from its own span context, so no data is written\nanywhere new; this connection is read-only credentials for looking a\ntrace back up. Connection ``host`` is the backend's base URL,\n``login``/``password`` are its API key pair.\n\nSpike-status: this is a proof of concept, not a stable API. The\npanel only supports Langfuse today.\n",
+                        "version_added": "0.5.0",
+                        "type": "string",
+                        "example": "langfuse_default",
+                        "default": "langfuse_default",
+                    },
                 },
             }
         },