feat: add AgenticComponentActivity for agent-authored bundle components

[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 final head 6e29e8a (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) 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 (2005 passed, 8 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 6e29e8a (325 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.

• 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 captain's choice.
  • 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

[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]

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.