Support bundle mode for inline Python tasks

packages/tangle-cli/src/tangle_cli/pipeline_compiler.py

@@ -1432,7 +1432,7 @@ def _build_local_from_python_components(
     """Build the ``<stem>.components.yaml`` content for @task refs.
 
     Returns an ordered map ``{fragment: {name?, local_from_python:
-    {image?, function, dependencies_from?, file}}}``, DEDUPED by FUNCTION
+    {image?, function, mode?, resolve_root?, dependencies_from?, file}}}``, DEDUPED by FUNCTION
     (the fragment = hyphenated function name). The SAME @task function
     called from multiple task sites collapses to one entry; TWO DISTINCT
     @task functions defined in ONE file each get their own entry (they
@@ -1496,6 +1496,16 @@ def _build_local_from_python_components(
         # to the file stem and extracts the wrong symbol.
         assert ref._task_function_name is not None
         local_from_python["function"] = ref._task_function_name
+        if ref._task_mode is not None:
+            local_from_python["mode"] = ref._task_mode
+        if ref._task_resolve_root is not None:
+            resolve_root = ref._task_resolve_root
+            if not resolve_root.exists():
+                raise CompileError(
+                    f"@task resolve_root is unreachable: {resolve_root}. "
+                    "Point resolve_root at an existing directory or drop it."
+                )
+            local_from_python["resolve_root"] = _relpath_posix(resolve_root, components_yaml_dir)
         if ref._task_dependencies_from is not None:
             deps = ref._task_dependencies_from
             if not deps.exists():

packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py

@@ -50,6 +50,8 @@ class CallableRef:
     _task_function_name: str | None = None
     _task_image: str | None = None
     _task_dependencies_from: Path | None = None
+    _task_mode: str | None = None
+    _task_resolve_root: Path | None = None
     _task_custom_annotations: dict[str, str] | None = None
     # ``@registered`` metadata. ``None`` for ``ref()``/``@task`` refs;
     # populated by the ``@registered`` decorator. Drives the compile-time
@@ -154,6 +156,8 @@ def materialize(self, output_path: Path | None = None) -> Path:
                 function_name=self._task_function_name,
                 dependencies_from=deps,
                 custom_annotations=self._task_custom_annotations,
+                mode=self._task_mode or "inline",
+                resolve_root=self._task_resolve_root,
             )
         else:
             generator.regenerate_yaml(
@@ -162,6 +166,8 @@ def materialize(self, output_path: Path | None = None) -> Path:
                 function_name=self._task_function_name,
                 image=self._task_image,
                 dependencies_from=self._task_dependencies_from,
+                mode=self._task_mode or "inline",
+                resolve_root=self._task_resolve_root,
             )
         return output_path
 

packages/tangle-cli/src/tangle_cli/python_pipeline/task.py

@@ -1,7 +1,7 @@
 """``@task`` decorator.
 
 The decorator captures metadata about a Python function (source path,
-function name, container image, dependencies, custom name, annotations)
+function name, container image, dependencies, generation mode, resolve root, custom name, annotations)
 and returns a :class:`CallableRef` with NO componentRef URL set.
 
 At compile time the driver collects every ``@task`` ref that gets called
@@ -66,6 +66,8 @@ def task(
     env: TaskEnv | None = None,
     image: str | None = None,
     dependencies_from: str | Path | None = None,
+    mode: str | None = None,
+    resolve_root: str | Path | None = None,
     annotations: dict[str, Any] | None = None,
 ) -> Callable[[Callable[..., Any]], CallableRef]:
     """Decorator: turn a Python function into a Tangle component ref.
@@ -99,6 +101,14 @@ def task(
             file when given as a string. Emitted into
             ``components.yaml#local_from_python.dependencies_from``.
             Overrides ``env.dependencies_from`` when both are given.
+        mode: Optional local-from-python generation mode. ``None``
+            preserves the hydrator default (currently ``inline``).
+            Use ``"bundle"`` to ask hydrate-time codegen to embed
+            first-party imports using the existing module bundler.
+        resolve_root: Optional module resolution root for bundle mode.
+            Relative strings are resolved relative to the task source
+            file, then emitted into
+            ``components.yaml#local_from_python.resolve_root``.
         annotations: Extra annotations to merge into the emitted
             component's ``metadata.annotations`` block.
 
@@ -143,6 +153,9 @@ def task(
     # the normalisation below unchanged; an explicit relative string is
     # still resolved relative to the @task source file.
     raw_dependencies_from = effective_deps_raw
+    raw_resolve_root = resolve_root
+    if mode is not None and mode not in {"inline", "bundle"}:
+        raise ValueError("@task(mode=...) must be 'inline', 'bundle', or None")
 
     def decorator(fn: Callable[..., Any]) -> CallableRef:
         # Capture the absolute path of the source file the user wrote
@@ -161,9 +174,8 @@ def decorator(fn: Callable[..., Any]) -> CallableRef:
 
         function_name = fn.__name__
 
-        # Resolve dependencies_from relative to the source file when
-        # the user gave a string. Absolute paths and explicit Path
-        # objects pass through unchanged.
+        # Resolve local paths relative to the source file when the user
+        # gave a relative value. Absolute paths pass through unchanged.
         deps_path: Path | None
         if raw_dependencies_from is None:
             deps_path = None
@@ -172,6 +184,14 @@ def decorator(fn: Callable[..., Any]) -> CallableRef:
             if not deps_path.is_absolute():
                 deps_path = (source_path.parent / deps_path).resolve()
 
+        resolve_root_path: Path | None
+        if raw_resolve_root is None:
+            resolve_root_path = None
+        else:
+            resolve_root_path = Path(raw_resolve_root)
+            if not resolve_root_path.is_absolute():
+                resolve_root_path = (source_path.parent / resolve_root_path).resolve()
+
         # No URL set at decoration time -- the compile driver rewrites
         # componentRef.url for @task-derived refs to
         # ``resolve://./<out_stem>.components.yaml#<fragment>`` after
@@ -183,6 +203,8 @@ def decorator(fn: Callable[..., Any]) -> CallableRef:
             _task_function_name=function_name,
             _task_image=effective_image,
             _task_dependencies_from=deps_path,
+            _task_mode=mode,
+            _task_resolve_root=resolve_root_path,
             _task_custom_annotations=dict(annotations) if annotations else None,
         )
 

tests/test_pipeline_compiler.py

@@ -344,6 +344,37 @@ def test_compile_task_decorator_emits_sidecar(tmp_path):
     validate_dehydrated_data(yaml.safe_load(out.read_text()))
 
 
+def test_compile_task_decorator_emits_bundle_mode_and_resolve_root(tmp_path):
+    """@task(mode="bundle") is carried into the auto-emitted sidecar."""
+    project = tmp_path / "project"
+    src = project / "src"
+    pipeline_path = src / "pipeline.py"
+    src.mkdir(parents=True)
+    (src / "helpers.py").write_text("MESSAGE = 'hello'\n", encoding="utf-8")
+    pipeline_path.write_text(
+        "from tangle_cli.python_pipeline import Out, pipeline, task\n"
+        "from helpers import MESSAGE\n\n"
+        "@task(image='python:3.12', mode='bundle', resolve_root='.')\n"
+        "def bundled_task() -> str:\n"
+        "    return MESSAGE\n\n"
+        "@pipeline('Bundle Pipeline')\n"
+        "def bundle_pipeline() -> Out[str]:\n"
+        "    bundled = bundled_task()\n"
+        "    return bundled\n",
+        encoding="utf-8",
+    )
+
+    out = project / "compiled.yaml"
+    result = compile_pipeline(pipeline_path, out)
+
+    sidecar = yaml.safe_load(result.components_path.read_text())
+    local_from_python = sidecar["bundled-task"]["local_from_python"]
+    assert local_from_python["mode"] == "bundle"
+    assert local_from_python["resolve_root"] == "./src"
+    assert local_from_python["file"] == "./src/pipeline.py"
+    assert local_from_python["function"] == "bundled_task"
+
+
 # ---------------------------------------------------------------------------
 # Runnable argument-value emission (raw string constant / graphInput /
 # taskOutput). Dispatch is on the VALUE's type, never the argument KEY.

tests/test_pipelines_cli.py

@@ -958,12 +958,15 @@ def _write_local_from_python_pipeline(
     project_dir: Path,
     python_file: str,
     *,
+    mode: str | None = None,
     resolve_root: str | None = None,
 ) -> Path:
     gen_config = {
         "file": python_file,
         "output_folder": "./generated",
     }
+    if mode is not None:
+        gen_config["mode"] = mode
     if resolve_root is not None:
         gen_config["resolve_root"] = resolve_root
     _write_pipeline(
@@ -1020,6 +1023,44 @@ def fake_regenerate_yaml(**kwargs):
     assert regenerated == [python_file.resolve()]
 
 
+def test_pipelines_hydrate_local_from_python_forwards_bundle_mode_and_resolve_root(
+    monkeypatch,
+    tmp_path: Path,
+):
+    from tangle_cli import pipeline_hydrator as hydrator_module
+    from tangle_cli.pipelines import hydrate_pipeline_file
+
+    project_dir = tmp_path / "project"
+    project_dir.mkdir()
+    src_dir = project_dir / "src"
+    src_dir.mkdir()
+    python_file = src_dir / "component.py"
+    python_file.write_text("# trusted project component\n", encoding="utf-8")
+    pipeline_path = _write_local_from_python_pipeline(
+        project_dir,
+        "./src/component.py",
+        mode="bundle",
+        resolve_root="./src",
+    )
+    calls: list[dict[str, object]] = []
+
+    def fake_regenerate_yaml(**kwargs):
+        calls.append(kwargs)
+        kwargs["output_path"].write_text(
+            "name: Generated Component\nimplementation:\n  container:\n    image: busybox\n",
+            encoding="utf-8",
+        )
+        return True
+
+    monkeypatch.setattr(hydrator_module, "regenerate_yaml", fake_regenerate_yaml)
+
+    hydrate_pipeline_file(pipeline_path)
+
+    assert calls[0]["python_file"] == python_file.resolve()
+    assert calls[0]["mode"] == "bundle"
+    assert calls[0]["resolve_root"] == src_dir.resolve()
+
+
 def test_pipelines_hydrate_local_from_python_refuses_untrusted_absolute_path(
     monkeypatch,
     tmp_path: Path,