feat: add AgenticComponentActivity for agent-authored bundle components

src/flowx/bundler/dab_writer.py

@@ -30,6 +30,7 @@
 from flowx.models.dab import DabNotebook
 from flowx.models.ir import (
     Activity,
+    AgenticComponentActivity,
     AppendVariableActivity,
     CopyActivity,
     DbtFactoryActivity,
@@ -133,6 +134,7 @@ def write_bundle(
     )
 
     pipeline_resources = _collect_pipeline_resources(workflow)
+    _check_pipeline_resource_keys(pipeline_resources, known_bundle_jobs)
     pipeline_variable_declarations = _build_pipeline_variable_declarations(pipeline_resources, catalog, schema)
     # sql_task references ${var.warehouse_id}; declare it (no default -> user supplies at deploy).
     if _bundle_uses_sql_task(workflow):
@@ -234,11 +236,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))
@@ -656,6 +654,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)
@@ -692,7 +697,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}"
@@ -1057,6 +1062,33 @@ def _collect_pipeline_resources(workflow: PreparedWorkflow) -> list[dict[str, An
     return resources
 
 
+def _check_pipeline_resource_keys(pipeline_resources: list[dict[str, Any]], job_resource_keys: set[str]) -> None:
+    """Fail when a pipeline resource key clashes with another resource in the bundle.
+
+    Every pipeline resource is written to ``resources/<key>.yml``, as is every static job, so a
+    pipeline key that matches a job key or an earlier pipeline key would silently replace that file
+    and drop the other resource. Bundle resource keys must also be unique across resource types, so
+    a match with a Python-generated dbt-factory job would fail at deploy time instead.
+
+    Raises:
+        ValueError: A pipeline resource key matches a job resource key or repeats an earlier one.
+    """
+    seen_pipeline_keys: set[str] = set()
+    for resource in pipeline_resources:
+        pipeline_key = resource["resource_key"]
+        if pipeline_key in job_resource_keys:
+            raise ValueError(
+                f"Pipeline resource key {pipeline_key!r} matches a job resource key in this bundle; "
+                "bundle resource keys must be unique"
+            )
+        if pipeline_key in seen_pipeline_keys:
+            raise ValueError(
+                f"Pipeline resource key {pipeline_key!r} is used by more than one pipeline resource; "
+                f"each would overwrite resources/{pipeline_key}.yml"
+            )
+        seen_pipeline_keys.add(pipeline_key)
+
+
 def _collect_pydabs_resource_entries(workflow: PreparedWorkflow) -> list[str]:
     """Returns the ``python.resources`` entries for every dbt-factory PyDABs hook in *workflow*.
 
@@ -2276,6 +2308,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,
     ControlEdge,
     CopyActivity,
@@ -296,6 +297,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

@@ -688,6 +688,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,95 @@
+"""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
+from flowx.utils import normalize_task_key
+
+# flowx owns a task's identity, its upstream wiring, and its run policy; these come from
+# the activity itself, so an authored task fragment may only say what the task runs.
+FLOWX_OWNED_TASK_FIELDS = frozenset(
+    {
+        "task_key",
+        "depends_on",
+        "run_if",
+        "timeout_seconds",
+        "max_retries",
+        "min_retry_interval_millis",
+        "retry_on_timeout",
+    }
+)
+
+
+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 _check_resource_key(task_key: str, resource_key: object) -> None:
+    """Reject a resource key that is not a plain identifier.
+
+    Each resource is written to ``resources/<resource_key>.yml``, so a key with path characters
+    could land outside the ``resources`` directory or overwrite another bundle file.
+    """
+    if not isinstance(resource_key, str) or not resource_key or resource_key != normalize_task_key(resource_key):
+        raise ValueError(
+            f"Agentic component {task_key!r} resource key {resource_key!r} must be a plain identifier "
+            "of lowercase letters, digits, and single underscores"
+        )
+
+
+def prepare(activity: AgenticComponentActivity, *, scope: str = "") -> PreparedActivity:
+    """Pass authored files, resources, and the authored task payload to the bundle writer.
+
+    The task's key, dependencies, run condition, timeout, and retries always come from
+    the activity, never from the authored fragment, so an agent cannot rewire or re-time
+    a task behind flowx's back. A fragment that tries to set any of them is rejected.
+
+    Raises:
+        ValueError: The authored task fragment sets a flowx-owned field, a file path
+            escapes the bundle's ``src`` directory, a resource key is not a plain
+            identifier, or a binary file is not valid base64.
+    """
+    del scope
+    owned_fields = sorted(FLOWX_OWNED_TASK_FIELDS & activity.task.keys())
+    if owned_fields:
+        raise ValueError(
+            f"Agentic component {activity.task_key!r} task wiring sets flowx-owned field(s) "
+            f"{', '.join(owned_fields)}; set dependencies and run policy on the activity instead"
+        )
+    for resource in activity.resources:
+        _check_resource_key(activity.task_key, resource.get("resource_key"))
+    notebooks = [
+        DabNotebook(
+            relative_path=_source_relative_path(file["path"]),
+            content=str(file.get("content", "")),
+            binary_content=(
+                base64.b64decode(str(file["binary_content"]), validate=True) if "binary_content" in file else None
+            ),
+            write_to_bundle_root=False,
+        )
+        for file in activity.files
+    ]
+    owned_task_fields = build_common_task_fields(activity)
+    task = {**owned_task_fields, **activity.task}
+    task.update(owned_task_fields)
+    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