[matthewmoorcroft]

Intent

Add AgenticComponentActivity, the in-engine escape hatch for a source activity the typed engine can't express. The agent authors raw bundle files, Lakeflow pipeline resources, job environments and a Databricks task fragment, and package writes them into the bundle as given. flowx still owns the task's identity, its dependencies and its run policy (D4): those always come from the activity and win over anything the agent wrote.

What changed

• models/ir.py, ir_serde.py, dab_writer._reconstruct_ir: the new IR type (files, resources, environments, task, raw_definition). It round-trips through the serialized report.
• Merging agentic results (ir_serde._find_and_replace_task): when an agentic component replaces a placeholder, the placeholder's identity and run policy always win, including the ADF timeout and retries. These are task_key, name, depends_on, timeout_seconds, max_retries and min_retry_interval_millis. The component also keeps the placeholder's source metadata (notifications, compute_mode, description, data_reads, data_writes, raw_definition and the like) for each field the agent leaves out; a value the agent sets wins. NotebookActivity merging is unchanged.
• preparer/activity_preparers/agentic_component.py:
  • Written as given: the fragment goes into the bundle unchanged. flowx sets only FLOWX_OWNED_TASK_FIELDS and rejects a fragment that sets any of them.
  • Named wiring passes: a few passes that make the bundle work apply to authored tasks just as they do to generated ones. They are listed in the docstring and in skills/flowx-convert/sources/adf.md.
  • Plumbing: where a value is absent, flowx adds only the plumbing the task needs (the ForEach item, collapsed notifications, a cluster for a notebook that needs one). It never overrides a value the fragment set.
  • Refusals: package refuses any of these:
    • a fragment without exactly one *_task payload, or that sets a flowx-owned field, at top level or in a for_each_task body (the body keeps its own task_key);
    • a for_each_task without a body task carrying a task_key;
    • a Python, wheel, JAR or dbt task that names no compute, or a spark_submit_task without new_cluster or a job_cluster_key;
    • a compute key set to null or empty, or more than one compute key on one task;
    • a job_cluster_key that isn't one of the clusters flowx defines;
    • a notebook_task.notebook_path that isn't a non-empty string, or base_parameters / job_parameters that aren't mappings;
    • libraries or existing_cluster_id set on the component instead of in the fragment;
    • a file path that escapes src, including Windows drive-only or root-only paths, or that uses backslashes;
    • malformed files, resources or environments, including a file without exactly one of content or binary_content;
    • unknown fields on a file entry, a resource entry or the component itself;
    • binary content that isn't valid base64.
• bundler/dab_writer.py:
  • Collision checks: packaging stops before writing any file when an authored file shares its path with a different file below src (another component's, a generated notebook or a setup notebook). It also stops when two authored paths differ only in letter case or one file path is another file's directory, and when a pipeline resource or environment key is reused with a different definition, or a pipeline key clashes with a job key. Identical duplicates are written once.
  • Environments: they are written at job level for the jobs whose tasks reference them, and every environment_key an authored task names must be defined.
  • Authored files: a plain DabNotebook.authored flag keeps them under src, and their text is written byte for byte, with no added newline and no platform line endings.
• preparer/activity_preparers/{if_condition,switch,for_each}.py: forward their children's pipeline resources and environments, so a component nested in control flow keeps them.
• validate/bundle_invariants.py: reports dangling_pipeline_reference when a pipeline_task points at a pipeline the bundle doesn't declare.
• Carries feat: add enrich step that records agent-authored insights on the discover inventory #95's head 960849e (including feat(adf): save source graphs with dataset lineage at ADF discover #94 and refactor: add source-neutral discovery graph contract and shared lineage core #84, with feat: add enrich step that records agent-authored insights on the discover inventory #95's enrich lock fix and feat(adf): save source graphs with dataset lineage at ADF discover #94's round-four fix) through clean --no-ff merges, with no conflicts. Nothing in feat: add AgenticComponentActivity for agent-authored bundle components #96 refers to agentic_insights.json, authored_against or the inventory rebuild, so they needed no changes. uv.lock is identical to feat: add enrich step that records agent-authored insights on the discover inventory #95's.

How it was tested

• Unit tests: tests/unit/test_agentic_component.py and test_bundle_invariants.py cover:

  • the round trip, and packaging serverless spark_python_task and python_wheel_task components with their environments;
  • every refusal and collision rule;
  • D4 through the real merge path;
  • each wiring and plumbing site;
  • components nested in control flow.

  The full unit suite (2136 passed, 9 skipped), ruff and mypy pass.

• Deterministic output: ADF discover, convert and package on the fixture pipelines produce bundles, metadata and translation reports byte-identical to feat: add enrich step that records agent-authored insights on the discover inventory #95's head 960849e (366 files).

• End to end: the no-mistakes test step drove the component through the real CLI (discover, convert, merge-agentic, package). The bundle validated with 0 errors against the DAB JSON schema.

• Codex cross-review, read-only, three times. Every fix has a regression test that fails on the previous code.

  • Round 1 found three feat: add AgenticComponentActivity for agent-authored bundle components #96 issues, all fixed in 9e70c54, with 63d12ad from the no-mistakes review:

    1. Two authored files at one path with different content: the second was renamed, but its task kept the original path.
    2. Windows root-relative and drive-relative paths escaped src.
    3. An empty or conflicting task fragment packaged and failed only at deploy.

  • Round 2 found three feat: add AgenticComponentActivity for agent-authored bundle components #96 issues, all fixed in 1b5a763:

    1. An authored file could still collide with a generated or setup notebook.
    2. A component inside IfCondition, Switch or ForEach lost its pipeline resources.
    3. dab_writer imported a private helper, now public as content_signature.

  • Deferred: Round 2's fourth finding, namespacing agentic resources in shared multi-DAG Airflow bundles.

  • Round 3 (on 839b2c6) found five. Three were feat: add AgenticComponentActivity for agent-authored bundle components #96 issues, all fixed in 3586b9d:

    1. A Python, wheel or JAR task with no compute packaged and failed only at deploy. It is now refused.
    2. A file entry with neither content nor binary_content packaged as an empty file. It is now refused.
    3. Authored text gained a trailing newline, and would have gained Windows line endings. It is now written byte for byte.

    The no-mistakes review then tightened the compute rule, in 1151a0e and 07eae42:

    • a job_cluster_key must name a cluster flowx defines;
    • spark_submit_task and dbt_task need compute too;
    • a spark_submit_task runs only on a new cluster.

    Round 3's other two findings join the deferred Airflow namespacing item: environment and pipeline paths that aren't rewritten, and paths that can be rewritten twice.

• Round 4, a review in the style of Greg's refactor: add source-neutral discovery graph contract and shared lineage core #84 and feat(adf): save source graphs with dataset lineage at ADF discover #94 reviews (fail closed, no unknown keys riding along, no false links, docs that match the code), plus an independent Codex pass with the same checklist, found seven feat: add AgenticComponentActivity for agent-authored bundle components #96 issues, all fixed test first in 24aa295 and 6ec268d:

  1. A null, empty or second compute key counted as naming compute.
  2. A for_each_task with no body, or a body without a task_key, packaged.
  3. Unknown fields on a file entry were dropped (an encoding flag wrote base64 text).
  4. Authored paths that differ only in case, or a file that is also a directory, clashed on disk.
  5. A generated file's added newline could hide an authored file at the same path.
  6. A merged component dropped the placeholder's notifications, compute mode and lineage.
  7. A for_each_task body inside a component was cleaned up as if generated.

  The no-mistakes review of that round then refused flowx-owned fields in a loop body, backslash paths, mistyped notebook and parameter fields, unknown resource and component keys, and libraries or existing_cluster_id on the component (6701a13, f2c8df7, 7e0222d, ca3d7c7). Outside feat: add AgenticComponentActivity for agent-authored bundle components #96 and left as follow-ups: an unknown IR type reloads as a placeholder, and main's generic merge drops source metadata for other replacement types.

• An audit against the same criteria, 26 Oct, found three more feat: add AgenticComponentActivity for agent-authored bundle components #96 issues. Each was fixed test first and then put through two Codex rounds with Greg's checklist, which added 28b758c and 92a7045:

  1. 543c67f: each ${var.X} an authored task or component environment uses is now declared in databricks.yml, and package refuses an authored file, library or environment path that names no file in the bundle. Generated tasks aren't checked, so their output is unchanged.
  2. 9567c25: on reload, a component's task must be a mapping and files, resources and environments must be lists; nothing is coerced.
  3. 39b19d6: a component's parameter_approximations, its own or kept from the placeholder, are listed in SETUP.md, and a malformed entry is refused.

  Follow-up for main, not this PR: IfCondition, ForEach and Switch drop every child's parameter_approximations, generated notebook tasks included, so a nested task's substitutions are missing from SETUP.md.

• The no-mistakes review then tightened the contract:

  • D4 and "written as given": 38531a4, 9234665, b99d98b and 1236e5c.
  • Payload validation and identical-duplicate pipeline resources: 38531a4 and bb488ec.
  • Component-carried job environments for serverless tasks: bf32de8, the option chosen in review.
  • Docs matched to the code: a9ab36f, ddbba41 and 540efe3.

Merge order

Merge #84 first, then #94 retargeted to main, then #95, then this PR (retargeted to main once #95 has merged), then #97.

This pull request and its description were written by Isaac.

[ghanse]

The task-ownership rules, collision checks and byte-for-byte writes look solid. Fix the parameter rewrite on nested agent-written loops before merging.

[ghanse]

Blocking: normalize_inner_task_params still rewrites the body of an agent-written for_each_task, so {{input.x}} becomes {{job.parameters.x}}, which points at the outer loop's parameters. Skip bodies marked _authored here and at line 191.

# authored:  {"table": "{{input.table_name}}"}
# after:     {"table": "{{job.parameters.table_name}}"}  # wrong loop

[matthewmoorcroft]

Fixed in 07a36c7, ccfb051 and 415309d: normalize_inner_task_params now leaves a for_each_task body marked _authored alone, which covers both call sites, so {{input.table_name}} stays as written. Each {{job.parameters.X}} the authored loop names is still forwarded from the enclosing job.

[ghanse]

Environments are written as-is, but the multi-DAG rename only updates paths inside tasks. So ../src/libraries/foo.whl in an environment points at the old location after combine. The PR description already defers this, so a TODO or follow-up issue is enough.

[matthewmoorcroft]

Added a TODO in 07a36c7 where environments are collected (_collect_environments); it stays with the deferred Airflow multi-DAG namespacing follow-up, as the description says.

[matthewmoorcroft]

Thanks Greg, both addressed in 3406913:

• Nested agent-written loops (blocking, r4321169995): fixed in 07a36c7, ccfb051 and 415309d. Inside a ForEach that runs as its own job, normalize_inner_task_params now skips a for_each_task body marked _authored, so your example {"table": "{{input.table_name}}"} stays as written, and {{input.x}} is no longer forwarded from the outer loop. That covers both call sites in for_each.py. A {{job.parameters.X}} the authored loop's inputs or body names is still forwarded from the enclosing job. An authored loop that is a ForEach's only child now runs in the inner job too, instead of nesting one For each inside another. Regression test: test_authored_loop_body_inside_a_for_each_job_keeps_its_own_input_references, covering all three ForEach shapes with a notebook and a serverless Python body; it fails before the fix.
• Environment paths after the multi-DAG rename (r4321170002): added a TODO in 07a36c7 where environments are collected (_collect_environments). It stays with the deferred Airflow multi-DAG namespacing follow-up, as the description says.