Add generic agentic component passthrough

src/flowx/bundler/dab_writer.py

@@ -29,6 +29,7 @@
 from flowx.models.dab import DabNotebook
 from flowx.models.ir import (
     Activity,
+    AgenticComponentActivity,
     AppendVariableActivity,
     ControlEdge,
     CopyActivity,
@@ -238,11 +239,7 @@ def write_bundle(
     src_dir = output_dir / "src"
 
     def _write_generated(notebooks: list[DabNotebook]) -> None:
-        root_artifacts = [
-            notebook
-            for notebook in notebooks
-            if notebook.relative_path.startswith("resources/") or notebook.relative_path == "pyproject.toml"
-        ]
+        root_artifacts = [notebook for notebook in notebooks if _is_bundle_root_artifact(notebook)]
         rest = [notebook for notebook in notebooks if notebook not in root_artifacts]
         if rest:
             created_files.extend(write_notebooks(rest, src_dir))
@@ -660,6 +657,13 @@ def _known_bundle_job_keys(workflow: PreparedWorkflow, resource_key: str) -> set
     return keys
 
 
+def _is_bundle_root_artifact(notebook: DabNotebook) -> bool:
+    """Return whether a generated file belongs at the bundle root instead of below ``src``."""
+    if notebook.write_to_bundle_root is not None:
+        return notebook.write_to_bundle_root
+    return notebook.relative_path.startswith("resources/") or notebook.relative_path == "pyproject.toml"
+
+
 def _namespace_workflow_assets(workflow: PreparedWorkflow) -> PreparedWorkflow:
     """Namespaces generated source files by DAG while preserving workspace paths."""
     cloned = copy.deepcopy(workflow)
@@ -696,7 +700,7 @@ def _namespace_workflow_assets(workflow: PreparedWorkflow) -> PreparedWorkflow:
                 notebook.content = notebook.content.replace("src/dbt_project", f"src/{prefix}/dbt_project")
                 notebook.content = notebook.content.replace("src/dbt_profiles", f"src/{prefix}/dbt_profiles")
                 continue
-            if original_path.startswith("resources/") or original_path == "pyproject.toml":
+            if _is_bundle_root_artifact(notebook):
                 continue
             notebook.relative_path = f"{prefix}/{original_path}"
             replacements[f"../src/{original_path}"] = f"../src/{notebook.relative_path}"
@@ -2279,6 +2283,14 @@ def _reconstruct_ir(task_ir: dict[str, Any]) -> Activity:
             consolidate_metadata_driven=bool(task_ir.get("consolidate_metadata_driven", False)),
             lookup_values=list(task_ir.get("lookup_values") or []),
         )
+    if task_type == "AgenticComponentActivity":
+        return AgenticComponentActivity(
+            **base,
+            files=list(task_ir.get("files") or []),
+            resources=list(task_ir.get("resources") or []),
+            task=dict(task_ir.get("task") or {}),
+            raw_definition=task_ir.get("raw_definition"),
+        )
     if task_type == "UnsupportedActivity":
         return UnsupportedActivity(
             **base,

src/flowx/ir_serde.py

@@ -21,6 +21,7 @@
 
 from flowx.models.ir import (
     Activity,
+    AgenticComponentActivity,
     AppendVariableActivity,
     CopyActivity,
     DataAsset,
@@ -242,6 +243,12 @@ def activity_extra_fields(activity: Activity) -> dict[str, Any]:
     extra: dict[str, Any] = {}
 
     match activity:
+        case AgenticComponentActivity():
+            extra["files"] = activity.files
+            extra["resources"] = activity.resources
+            extra["task"] = activity.task
+            if activity.raw_definition is not None:
+                extra["raw_definition"] = activity.raw_definition
         case NotebookActivity():
             extra["notebook_path"] = activity.notebook_path
             if activity.base_parameters:

src/flowx/models/dab.py

@@ -20,12 +20,16 @@ class DabNotebook:
         language: Notebook language (``"python"``, ``"sql"``, ``"scala"``, ``"r"``).
         binary_content: Raw bytes for binary files (e.g. JARs).  When set,
             the notebook writer writes these bytes instead of ``content``.
+        write_to_bundle_root: Controls whether the bundle writer places this
+            file at the bundle root instead of below ``src``. ``None`` keeps
+            the legacy path-based routing for PyDABs artifacts.
     """
 
     relative_path: str
     content: str = ""
     language: str = "python"
     binary_content: bytes | None = None
+    write_to_bundle_root: bool | None = None
 
 
 # ---------------------------------------------------------------------------

src/flowx/models/ir.py

@@ -685,6 +685,27 @@ class DbtFactoryActivity(Activity):
     nodes: list[dict[str, Any]] = field(default_factory=list)
 
 
+@dataclass(slots=True, kw_only=True)
+class AgenticComponentActivity(Activity):
+    """Bundle components authored for a source activity the typed engine cannot express.
+
+    Attributes:
+        files: Files to write below the bundle's ``src`` directory. Each entry
+            carries ``path`` and either UTF-8 ``content`` or base64-encoded
+            ``binary_content``.
+        resources: Pipeline resources in the existing ``resource_key`` plus
+            raw ``definition`` shape used by the bundle writer.
+        task: Raw Databricks task fragment containing either ``pipeline_task``
+            or ``notebook_task`` wiring to an authored resource or file.
+        raw_definition: Original source definition retained for auditing.
+    """
+
+    files: list[dict[str, Any]] = field(default_factory=list)
+    resources: list[dict[str, Any]] = field(default_factory=list)
+    task: dict[str, Any] = field(default_factory=dict)
+    raw_definition: dict[str, Any] | None = None
+
+
 @dataclass(slots=True, kw_only=True)
 class UnsupportedActivity(Activity):
     """Sentinel for activities that could not be translated.

src/flowx/preparer/activity_preparers/agentic_component.py

@@ -0,0 +1,45 @@
+"""Lower agent-authored bundle components through the existing output channels."""
+
+from __future__ import annotations
+
+import base64
+from pathlib import PurePosixPath, PureWindowsPath
+
+from flowx.models.dab import DabNotebook
+from flowx.models.ir import AgenticComponentActivity
+from flowx.preparer.workflow_preparer import PreparedActivity, build_common_task_fields
+
+
+def _source_relative_path(raw_path: object) -> str:
+    """Return a safe path relative to the bundle's ``src`` directory."""
+    relative_path = PurePosixPath(str(raw_path))
+    windows_path = PureWindowsPath(str(raw_path))
+    if (
+        relative_path.is_absolute()
+        or windows_path.is_absolute()
+        or relative_path.as_posix() == "."
+        or ".." in relative_path.parts
+        or ".." in windows_path.parts
+    ):
+        raise ValueError(f"Agentic component file path {raw_path!r} must be relative to the bundle src directory")
+    return relative_path.as_posix()
+
+
+def prepare(activity: AgenticComponentActivity, *, scope: str = "") -> PreparedActivity:
+    """Pass authored files, resources, and task wiring to the bundle writer."""
+    del scope
+    notebooks = [
+        DabNotebook(
+            relative_path=_source_relative_path(file["path"]),
+            content=str(file.get("content", "")),
+            binary_content=(base64.b64decode(str(file["binary_content"])) if "binary_content" in file else None),
+            write_to_bundle_root=False,
+        )
+        for file in activity.files
+    ]
+    task = {**build_common_task_fields(activity), **activity.task, "task_key": activity.task_key}
+    return PreparedActivity(
+        task=task,
+        notebooks=notebooks,
+        pipeline_resources=list(activity.resources),
+    )

src/flowx/preparer/workflow_preparer.py

@@ -9,6 +9,7 @@
 from flowx.models.dab import DabNotebook, ParameterApproximation, SecretInstruction, SetupTask
 from flowx.models.ir import (
     Activity,
+    AgenticComponentActivity,
     AppendVariableActivity,
     CopyActivity,
     DbtFactoryActivity,
@@ -142,6 +143,7 @@ def prepare_activity(
 ) -> PreparedActivity:
     """Dispatches to the appropriate activity preparer based on activity type."""
     from flowx.preparer.activity_preparers import (
+        agentic_component,
         append_variable,
         copy,
         databricks_job,
@@ -164,6 +166,7 @@ def prepare_activity(
     )
 
     dispatch: dict[type, Any] = {
+        AgenticComponentActivity: agentic_component.prepare,
         NotebookActivity: notebook.prepare,
         SparkJarActivity: spark_jar.prepare,
         SparkPythonActivity: spark_python.prepare,

src/flowx/validate/bundle_invariants.py

@@ -25,6 +25,7 @@
 _ANCHOR_RE = re.compile(r"[&*]id\d+\b")
 _JOB_PARAM_REF_RE = re.compile(r"\{\{\s*job\.parameters\.([A-Za-z0-9_]+)\s*\}\}")
 _JOB_RESOURCE_ID_RE = re.compile(r"\$\{resources\.jobs\.([^.}]+)\.id\}")
+_PIPELINE_RESOURCE_ID_RE = re.compile(r"\$\{resources\.pipelines\.([^.}]+)\.id\}")
 _PYDABS_JOB_RE = re.compile(r"resources\.add_job\(\s*['\"]([^'\"]+)['\"]")
 
 
@@ -246,10 +247,15 @@ def check_bundle_dir(bundle_dir: Path) -> BundleInvariantResult:
             documents.append((path, document))
 
     known_jobs: set[str] = set()
+    known_pipelines: set[str] = set()
     for _path, document in documents:
-        jobs = (document.get("resources") or {}).get("jobs") or {}
+        document_resources = document.get("resources") or {}
+        jobs = document_resources.get("jobs") or {}
         if isinstance(jobs, dict):
             known_jobs.update(str(job_key) for job_key in jobs)
+        pipelines = document_resources.get("pipelines") or {}
+        if isinstance(pipelines, dict):
+            known_pipelines.update(str(pipeline_key) for pipeline_key in pipelines)
         python_resources = (document.get("python") or {}).get("resources") or []
         for resource in python_resources:
             if not isinstance(resource, str):
@@ -268,6 +274,22 @@ def check_bundle_dir(bundle_dir: Path) -> BundleInvariantResult:
             if not isinstance(job, dict):
                 continue
             for task in _iter_tasks(job.get("tasks") or []):
+                pipeline_task = task.get("pipeline_task") or {}
+                pipeline_id = pipeline_task.get("pipeline_id") if isinstance(pipeline_task, dict) else None
+                pipeline_match = (
+                    _PIPELINE_RESOURCE_ID_RE.fullmatch(pipeline_id) if isinstance(pipeline_id, str) else None
+                )
+                if pipeline_match is not None and pipeline_match.group(1) not in known_pipelines:
+                    findings.append(
+                        BundleFinding(
+                            code="dangling_pipeline_reference",
+                            location=f"{path.name}, job '{job_key}', task '{task.get('task_key', '')}'",
+                            message=(
+                                f"pipeline_task references bundle pipeline '{pipeline_match.group(1)}', which is not "
+                                "declared in static resource YAML."
+                            ),
+                        )
+                    )
                 run_job = task.get("run_job_task") or {}
                 job_id = run_job.get("job_id") if isinstance(run_job, dict) else None
                 match = _JOB_RESOURCE_ID_RE.fullmatch(job_id) if isinstance(job_id, str) else None