Add handling for global parameters (#15)

## Changes
<!-- Summary of your changes that are easy to understand. Add
screenshots when necessary -->
Add handler methods to resolve global parameters from ADF resources. 

Parameter is controlled by the user via prompting. The choice is applied
to an entire ADF resource during conversion. Users can choose to:
- Inject resolved parameter values directly into the code or
configuration during conversion
- Create bundle variables and parameterize code or configuration with
bundle variable references

### Linked issues
<!-- DOC: Link issue with a keyword: close, closes, closed, fix, fixes,
fixed, resolve, resolves, resolved. See
https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword
-->
N/A

### Tests
<!-- How is this tested? Please see the checklist below and also
describe any other relevant tests -->

- [x] manually tested
- [x] added unit tests
- [ ] added integration tests

Author: ghanse

AGENTS.md

@@ -113,6 +113,17 @@ ExecuteDataFlow, SqlServerStoredProcedure, AzureFunction, WebHook, Custom, Execu
 - Leaf types return Activity only, control-flow returns (Activity, TranslationContext)
 - Use `parse_expression()` for ADF expression translation, return None for unsupported
 
+### Naming, docstrings, and comments
+
+- Spell names out: use unabbreviated variable, parameter, function, and class names
+  (`parameter_values` not `params`, `whole_reference` not `whole`). Short loop indices and
+  regex match binders (`match`, `item`) are fine.
+- Write docstrings in plain, conversational language aimed at both users and maintainers.
+  Say what the function does and why in everyday terms; skip jargon and marketing tone.
+- Prefer self-documenting code over inline comments. Reserve comments for the non-obvious
+  *why* (a workaround, a spec quirk, a subtle ordering constraint) -- not for restating what
+  the code already says. Delete comments that narrate self-evident lines.
+
 ## Adding a New Deterministic Translator
 
 1. Add IR dataclass to `src/flowx/models/ir.py`

docs/content/docs/options.mdx

@@ -83,6 +83,30 @@ The following are required to consolidate into a single ingestion pipeline:
 - The number of metadata rows or objects must be less than 250
 </Callout>
 
+## global_parameter_resolution
+
+Controls how global parameters (`@pipeline().globalParameters.X`) are translated. This option is
+applied uniformly to every pipeline in the input resource.
+
+| Value             | Default | Behavior                                                                                                                                                 |
+|-------------------|---------|----------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `literal`         | True    | Writes each global variable into the translated configuration or notebook code.                                                                          |
+| `bundle_variable` | False   | Uses a `${var.X}` reference and declares the global variable as a [DAB bundle variable](https://docs.databricks.com/aws/en/dev-tools/bundles/variables). |
+
+Under `bundle_variable`, each referenced global becomes a variable in `databricks.yml` with its
+factory value as the default, so a deploy with no overrides reproduces the original ADF behavior.
+You then change the value per target or at deploy time
+(`databricks bundle deploy -t <target> --var "X=<value>"`). Globals referenced inside generated
+notebook code are wired through the task's `base_parameters` so `${var.X}` still resolves at
+runtime.
+
+<Callout type="warn" title="Sensitive values ship in plaintext">
+The default values for global parameters are written into `databricks.yml` as plain text. For sensitive
+values (e.g. tokens, connection strings, URLs with embedded SAS signatures), clear the default and
+supply the value at deploy time using a `--var` argument, a per-target override, or a secret scope instead 
+of committing sensitive values to the bundle. See `SETUP.md` for a full list of converted global variables.
+</Callout>
+
 ## Discover complexity report
 
 The `discover` phase emits `<output_dir>/metadata/profile_report.csv` (default `./flowx_output/metadata/profile_report.csv`), one row per pipeline with the following columns:

skills/flowx-convert/SKILL.md

@@ -95,14 +95,22 @@ Execute the translation engine on all deterministic activities:
 "$PY" -m flowx.translator.engine \
   --source-dir <adf_source_dir> \
   --output-dir <output_dir> \
-  [--pipeline <pipeline_name>]
+  [--pipeline <pipeline_name>] \
+  [--global-parameter-resolution literal|bundle_variable]
 ```
 
 Where:
 - `<adf_source_dir>` is the original ADF JSON directory (the same `--source-dir` used by discover)
 - `<output_dir>` is the **shared migration output directory** (default: `./flowx_output`) — the
   same one discover used
 - `<pipeline_name>` (optional) — when provided, translates only the named pipeline. **Always pass `--pipeline` when the user has specified a specific pipeline to migrate**, matching the value passed to the discover phase.
+- `--global-parameter-resolution` (optional, default `literal`) — how `@pipeline().globalParameters.X`
+  references resolve, applied to every pipeline. `literal` bakes the factory value in as a literal;
+  `bundle_variable` emits `${var.X}` and declares the global as a DAB bundle variable whose default is
+  the factory value, so it can be set at deploy time (`--var X=…` or a per-target override) instead of
+  being hard-coded into pipeline/activity bodies. Globals referenced inside generated notebook code are
+  bridged through the task's `base_parameters` so `${var.X}` still resolves. See SETUP.md for the list
+  of hoisted variables and a plaintext-secret caveat.
 
 The translation report and intermediate IR are written to the **transient** `<output_dir>/.work/`
 folder (`translation_report.json`, per-pipeline IR, `gaps.json`). These are consumed by the steps

src/flowx/adapter/constants.py

@@ -31,6 +31,7 @@
 INPUT_ADF_RESOURCE_URL: Final[str] = "adf_resource_url"
 INPUT_OUTPUT_DIR: Final[str] = "output_dir"
 INPUT_INVENTORY_PATH: Final[str] = "inventory_path"
+INPUT_GLOBAL_PARAMETER_RESOLUTION: Final[str] = "global_parameter_resolution"
 INPUT_TRANSLATION_REPORT_PATH: Final[str] = "translation_report_path"
 INPUT_OUTPUT_BUNDLE_PATH: Final[str] = "output_bundle_path"
 INPUT_CATALOG: Final[str] = "catalog"

src/flowx/adapter/session.py

@@ -19,6 +19,7 @@
     INPUT_BUNDLE_NAME,
     INPUT_CATALOG,
     INPUT_DATABRICKS_PROFILE,
+    INPUT_GLOBAL_PARAMETER_RESOLUTION,
     INPUT_INSTALL_DASHBOARD,
     INPUT_INVENTORY_PATH,
     INPUT_OUTPUT_BUNDLE_PATH,
@@ -297,6 +298,17 @@ def _collect_motif_consolidations(self) -> dict[str, MotifConsolidate]:
         default="./flowx_output",
         required=False,
     ),
+    MigrationInputOption(
+        option_id=INPUT_GLOBAL_PARAMETER_RESOLUTION,
+        prompt="How should factory global parameters be resolved?",
+        description=(
+            "Applies to every pipeline. 'literal' bakes each @pipeline().globalParameters.X value in as a "
+            "literal; 'bundle_variable' emits ${var.X} and declares the global as a DAB bundle variable with "
+            "the factory value as its default, so it can be changed at deploy time."
+        ),
+        default="literal",
+        required=False,
+    ),
 )
 
 _PACKAGE_OPTIONS: tuple[MigrationInputOption, ...] = (

src/flowx/bundler/dab_writer.py

@@ -121,6 +121,8 @@ def write_bundle(
 
     pipeline_resources = _collect_pipeline_resources(workflow)
     pipeline_variable_declarations = _build_pipeline_variable_declarations(pipeline_resources, catalog, schema)
+    hoisted_global_variables = _collect_hoisted_global_variables(workflow)
+    extra_variable_declarations = {**pipeline_variable_declarations, **hoisted_global_variables}
 
     # 1. Write databricks.yml. When any task runs on classic compute, spark_version / node_type_id
     #    defaults come from the ADF linked-service configs; when every task is serverless, they're omitted.
@@ -133,7 +135,7 @@ def write_bundle(
         spark_version=inferred_spark_version,
         node_type_id=inferred_node_type_id,
         include_cluster_variables=bundle_uses_classic_cluster,
-        extra_variables=pipeline_variable_declarations,
+        extra_variables=extra_variable_declarations,
     )
     databricks_yml_path.write_text(
         yaml.dump(
@@ -156,7 +158,8 @@ def write_bundle(
     resources_dir = output_dir / "resources"
     resources_dir.mkdir(parents=True, exist_ok=True)
     job_yml_path = resources_dir / f"{resource_key}.yml"
-    job_resource = _build_job_resource(workflow, resource_key)
+    hoisted_global_names = set(hoisted_global_variables)
+    job_resource = _build_job_resource(workflow, resource_key, hoisted_globals=hoisted_global_names)
     job_yml_path.write_text(
         yaml.dump(
             job_resource, default_flow_style=False, sort_keys=False, allow_unicode=True, Dumper=_BundleYamlDumper
@@ -170,7 +173,9 @@ def write_bundle(
     for inner in workflow.inner_workflows:
         inner_key = normalize_task_key(inner.name)
         inner_yml_path = resources_dir / f"{inner_key}.yml"
-        inner_resource = _build_job_resource(inner, inner_key, extra_notebooks_for_augment=workflow.notebooks)
+        inner_resource = _build_job_resource(
+            inner, inner_key, extra_notebooks_for_augment=workflow.notebooks, hoisted_globals=hoisted_global_names
+        )
         inner_yml_path.write_text(
             yaml.dump(
                 inner_resource,
@@ -289,6 +294,7 @@ def write_bundle(
         manual_schedule_time_of_day=manual_schedule_time_of_day_configs,
         manual_credentials=manual_credential_configs,
         neutralized_conditions=list(_neutralized_conditions),
+        hoisted_global_variables=hoisted_global_variables,
     )
     setup_path = output_dir / "SETUP.md"
     setup_path.write_text(render_setup_md(prereqs, bundle_name=effective_name), encoding="utf-8")
@@ -797,6 +803,26 @@ def _collect_variable_references(value: Any) -> set[str]:
     return refs
 
 
+def _collect_hoisted_global_variables(workflow: PreparedWorkflow) -> dict[str, Any]:
+    """Returns the bundle-variable declarations for hoisted factory globals.
+
+    Merges ``bundle_variables`` from *workflow* and every inner workflow so a
+    global referenced only inside a nested (run_job_task) workflow is still
+    declared in the root ``variables:`` block.
+
+    Args:
+        workflow: The prepared workflow being written.
+
+    Returns:
+        Mapping of variable name to its DAB declaration dict.
+    """
+    declarations: dict[str, Any] = {}
+    for inner in workflow.inner_workflows:
+        declarations.update(inner.bundle_variables)
+    declarations.update(workflow.bundle_variables)
+    return declarations
+
+
 def _build_pipeline_variable_declarations(
     pipeline_resources: list[dict[str, Any]],
     catalog: str,
@@ -1204,13 +1230,20 @@ def _collect_all_task_keys(tasks: list[dict[str, Any]]) -> set[str]:
     return keys
 
 
-def _augment_base_parameters(tasks: list[dict[str, Any]], notebooks: list[DabNotebook]) -> None:
+def _augment_base_parameters(
+    tasks: list[dict[str, Any]], notebooks: list[DabNotebook], hoisted_globals: set[str] | None = None
+) -> None:
     """Ensure every widget a notebook reads is declared in its base_parameters.
 
     Args:
         tasks: Top-level task dicts (mutated in place).
         notebooks: Generated notebooks to scan.
+        hoisted_globals: Names of factory globals hoisted to bundle variables.
+            A widget matching one of these binds to ``${var.NAME}`` so the
+            deploy-time bundle variable flows into the notebook; other widgets
+            default to an empty string as before.
     """
+    hoisted = hoisted_globals or set()
     notebook_by_relpath = {notebook.relative_path: notebook for notebook in notebooks}
 
     def visit(task: dict[str, Any]) -> None:
@@ -1223,7 +1256,8 @@ def visit(task: dict[str, Any]) -> None:
                 widgets = set(_WIDGET_REFERENCE.findall(notebook.content))
                 base_parameters = notebook_task.setdefault("base_parameters", {})
                 for widget_name in sorted(widgets):
-                    base_parameters.setdefault(widget_name, "")
+                    fallback = "${var." + widget_name + "}" if widget_name in hoisted else ""
+                    base_parameters.setdefault(widget_name, fallback)
         for_each = task.get("for_each_task")
         if for_each and isinstance(for_each.get("task"), dict):
             visit(for_each["task"])
@@ -1238,6 +1272,7 @@ def _build_job_resource(
     *,
     attach_clusters: bool = True,
     extra_notebooks_for_augment: list[DabNotebook] | None = None,
+    hoisted_globals: set[str] | None = None,
 ) -> dict[str, Any]:
     """Builds a job resource dict for a single workflow.
 
@@ -1256,7 +1291,7 @@ def _build_job_resource(
     # For inner jobs (run_job_task), notebooks live in the parent workflow's list — pass them in so widget
     # auto-augment can still find the bound notebook and populate base_parameters.
     augment_scope = list(workflow.notebooks) + list(extra_notebooks_for_augment or [])
-    _augment_base_parameters(workflow.tasks, augment_scope)
+    _augment_base_parameters(workflow.tasks, augment_scope, hoisted_globals)
     # Task values don't cross run_job_task boundaries; such a reference resolves to an empty string at
     # runtime, so emit it now for SETUP.md §4. C-43: a blanked condition operand is always-true, so record
     # each neutralised condition for the SETUP.md re-wiring section.
@@ -1463,6 +1498,7 @@ def pipeline_dict_to_ir(pipeline_dict: dict[str, Any]) -> tuple[Pipeline, list[d
         parameters=parameters or None,
         translation_configuration=_reconstruct_configuration(pipeline_dict.get("translation_configuration")),
         schedule=pipeline_dict.get("schedule"),
+        bundle_variables=pipeline_dict.get("bundle_variables") or {},
     )
     return pipeline, parameters
 

src/flowx/bundler/prereqs_writer.py

@@ -131,6 +131,7 @@ class Prereqs:
     # C-43 (CF5-001 / CF5-002): condition_task operands blanked because they referenced a task in another
     # job ({task_key, field, original_ref}); a blanked operand is always-true, so the user must re-wire it.
     neutralized_conditions: list[dict[str, str]] = field(default_factory=list)
+    hoisted_global_variables: dict[str, dict[str, Any]] = field(default_factory=dict)
 
     def is_empty(self) -> bool:
         """Return ``True`` when nothing needs to happen before ``bundle run``."""
@@ -150,6 +151,7 @@ def is_empty(self) -> bool:
             and not self.manual_schedule_time_of_day
             and not self.manual_credentials
             and not self.neutralized_conditions
+            and not self.hoisted_global_variables
         )
 
 
@@ -361,6 +363,7 @@ def build_prereqs(
     manual_schedule_time_of_day: list[dict[str, Any]] | None = None,
     manual_credentials: list[dict[str, Any]] | None = None,
     neutralized_conditions: list[dict[str, str]] | None = None,
+    hoisted_global_variables: dict[str, dict[str, Any]] | None = None,
 ) -> Prereqs:
     """Assemble a :class:`Prereqs` from the bundle's generated artifacts.
 
@@ -408,6 +411,7 @@ def build_prereqs(
         manual_schedule_time_of_day=list(manual_schedule_time_of_day or []),
         manual_credentials=list(manual_credentials or []),
         neutralized_conditions=list(neutralized_conditions or []),
+        hoisted_global_variables=dict(hoisted_global_variables or {}),
     )
 
 
@@ -476,6 +480,29 @@ def render_setup_md(prereqs: Prereqs, *, bundle_name: str) -> str:
         )
         lines.append("")
 
+    if prereqs.hoisted_global_variables:
+        lines.append("## Factory global parameters (bundle variables)")
+        lines.append("")
+        lines.append(
+            "These ADF factory global parameters were hoisted to DAB bundle variables "
+            "(`global_parameter_resolution=bundle_variable`). Each is declared in `databricks.yml` "
+            "with its factory value as the default, so a deploy with no overrides reproduces the "
+            "original behaviour. Override per target in `databricks.yml` or at deploy time:"
+        )
+        lines.append("")
+        lines.append("```bash")
+        for name in sorted(prereqs.hoisted_global_variables):
+            lines.append(f'databricks bundle deploy -t <target> --var "{name}=<value>"')
+        lines.append("```")
+        lines.append("")
+        lines.append(
+            "> **Security note:** the factory-value defaults are stored in plaintext in "
+            "`databricks.yml`. For any sensitive value (tokens, connection strings, URLs with "
+            "embedded SAS signatures), clear the default and supply it at deploy time via `--var`, "
+            "a per-target override, or a secret scope instead of committing it to the bundle."
+        )
+        lines.append("")
+
     if prereqs.missing_notebooks:
         lines.append("## Notebooks to author")
         lines.append("")

src/flowx/models/ir.py

@@ -532,6 +532,8 @@ class Pipeline:
         tasks: Ordered list of translated activities.
         tags: System and user-defined tags.
         not_translatable: Entries describing properties that could not be translated.
+        bundle_variables: DAB bundle-variable declarations (name -> ``{"description", "default"}``)
+            for factory globals hoisted under the ``bundle_variable`` resolution policy.
     """
 
     name: str
@@ -541,6 +543,7 @@ class Pipeline:
     tags: dict[str, str] = field(default_factory=dict)
     not_translatable: list[dict[str, Any]] = field(default_factory=list)
     translation_configuration: TranslationConfiguration | None = None
+    bundle_variables: dict[str, dict[str, Any]] = field(default_factory=dict)
 
 
 @dataclass(frozen=True, slots=True)
@@ -562,6 +565,7 @@ class TranslationContext:
     variable_default_literals: MappingProxyType[str, str] = field(default_factory=lambda: MappingProxyType({}))
     global_parameters: MappingProxyType[str, Any] = field(default_factory=lambda: MappingProxyType({}))
     linked_service_parameters: MappingProxyType[str, Any] = field(default_factory=lambda: MappingProxyType({}))
+    global_parameter_resolution: str = "literal"
 
     def with_activity(self, name: str, activity: Activity) -> TranslationContext:
         """Return a new context with *activity* added to the cache.
@@ -582,6 +586,7 @@ def with_activity(self, name: str, activity: Activity) -> TranslationContext:
             variable_default_literals=self.variable_default_literals,
             global_parameters=self.global_parameters,
             linked_service_parameters=self.linked_service_parameters,
+            global_parameter_resolution=self.global_parameter_resolution,
         )
 
     def get_activity(self, activity_name: str) -> Activity | None:
@@ -627,6 +632,7 @@ def with_variable(
             variable_default_literals=self.variable_default_literals,
             global_parameters=self.global_parameters,
             linked_service_parameters=self.linked_service_parameters,
+            global_parameter_resolution=self.global_parameter_resolution,
         )
 
     def with_variable_types(
@@ -661,6 +667,7 @@ def with_variable_types(
             variable_default_literals=MappingProxyType({**self.variable_default_literals, **(default_literals or {})}),
             global_parameters=self.global_parameters,
             linked_service_parameters=self.linked_service_parameters,
+            global_parameter_resolution=self.global_parameter_resolution,
         )
 
     def get_variable_task_key(self, variable_name: str) -> str | None:
@@ -698,6 +705,7 @@ def with_linked_service_parameters(self, params: dict[str, Any]) -> TranslationC
             variable_default_literals=self.variable_default_literals,
             global_parameters=self.global_parameters,
             linked_service_parameters=MappingProxyType(dict(params)),
+            global_parameter_resolution=self.global_parameter_resolution,
         )
 
     def get_global_parameter(self, name: str) -> Any: