Phase 1: In-engine agentic escape hatch (passthrough + post-convert IR alteration)

[matthewmoorcroft]

Summary: Add an in-engine passthrough IR node and a user-approved post-convert pipeline replacement path for routed agentic components; this Phase 1 escape hatch is not built.

Depends on: #84, the main re-land of #79-#81, and routing #77.

Blocks: Phase 1 coarse agentic conversion; provides a validated bridge until decision-fed conversion in #85/#86.

Current status

This is an open issue with no implementation yet.

The package seams already exist:

• src/flowx/preparer/workflow_preparer.py:PreparedActivity/PreparedWorkflow/merge_prepared_artifacts carry task dictionaries, DabNotebook files, and pipeline_resources.
• src/flowx/bundler/notebook_writer.py:write_notebooks writes text or binary DabNotebook payloads.
• src/flowx/bundler/dab_writer.py:write_bundle/_collect_pipeline_resources emits prepared files and pipeline resources.
• src/flowx/bundler/dab_writer.py:_reconstruct_ir rehydrates known task types from translation_report.json, but unknown types fall through to PlaceholderActivity. A passthrough type would therefore be silently stubbed today.
• src/flowx/validate/bundle_invariants.py:check_job/check_bundle_dir already validates duplicate keys, dangling depends_on, cycles, and dangling run_job_task references, but does not validate pipeline_task resource references.

The existing agentic contract is intentionally too narrow for this path. src/flowx/agentic.py:GapEnvelope/_validate_replacement accepts one generated leaf file and notebook/sql/spark_python replacement only; _assert_task_invariants/_graph_hash require unchanged task count, enclosing structure, identity, dependencies, and policy. That contract cannot add/combine/rename tasks or replace a whole routed pipeline.

What needs to change

Implement a generic IR node, for example AgenticComponentActivity, carrying:

• generated files (safe relative path, text/binary content, language/metadata);
• arbitrary declared bundle resources (kind, resource_key, definition), initially including pipelines;
• the DAB task wiring that invokes those files/resources;
• retained raw source definition plus an honesty manifest of consumed source names/hashes and model/provider provenance.

Add a separate, opt-in post-convert accept path. Convert runs once unchanged to produce the deterministic baseline; for components routed agentic by #77, the accepted result may replace pipeline IR structurally (add/combine/rename tasks and pipelines). Deterministic components remain untouched. All authored components must land in translation_report.json and flow through normal package validation/writers—never “convert outside.”

How to approach

• Add the passthrough dataclass in models/ir.py and forward serialization in ir_serde.py:activity_to_dict/activity_extra_fields.
• Add an explicit branch in bundler/dab_writer.py:_reconstruct_ir before the PlaceholderActivity fallback, including all file/resource/task fields.
• Add a small preparer module and dispatch entry in preparer/workflow_preparer.py:prepare_activity that copies files into PreparedActivity.notebooks (or a generalized file channel), resources into pipeline_resources, and authored wiring into task.
• Reuse bundler/notebook_writer.py:write_notebooks and dab_writer.py:write_bundle/_collect_pipeline_resources rather than adding an external writer.
• Extend validate/bundle_invariants.py:check_bundle_dir with pipeline_task-to-declared-pipeline-resource validation mirroring the existing dangling_run_job_reference check.
• Add a new coarse acceptance contract beside agentic.py, fingerprint-bound to inventory, routing plan, baseline report, and consumed source. Replace leaf graph-hash equality with structural validation for routed pipelines: unique names/task keys, valid dependencies, acyclic graphs, valid resource references, and component closure.
• Mark coverage/report evidence as agent-authored/unverified and retain the honesty manifest/raw_definition.

Acceptance/verification:

• Serde/rehydration round-trip preserves every passthrough file, resource, task, hash, and provenance field.
• Golden package tests produce exact expected notebooks/binary files, job tasks, and pipeline YAML using existing writers.
• Structural tests reject duplicate keys, dangling dependencies, cycles, unsafe paths, unknown resource kinds, undeclared pipeline resources, and alteration outside routed components.
• Tests prove add/combine/rename is allowed inside a routed component and deterministic pipelines remain byte-identical.
• With no routed agentic component, convert and package output are unchanged.
• Airflow's existing discover/convert/package and leaf resolver remain untouched unless Pete explicitly adopts this coarse path.

This is additive and opt-in; conversion itself stays unchanged in Phase 1.