Load a TaskEnv from a config file with TaskEnv.from_config (v0.1.17)

README.md

@@ -614,6 +614,61 @@ def score(...):
     ...
 ```
 
+Declare the environment in a config file instead with `TaskEnv.from_config(path)`. A relative `path` resolves against the **calling pipeline module**, never the working directory, so the same script selects the same config wherever `tangle` runs from:
+
+```python
+EVAL = TaskEnv.from_config("envs.yaml")  # relative to THIS file
+```
+
+```yaml
+# envs.yaml
+image: python:3.12
+dependencies_from: pyproject.toml  # relative to THIS config file
+```
+
+The file is read with the same loader `--config` uses, so the `_select` directive works here identically — the case is chosen exactly and case-sensitively by an environment variable, and an unset or unmatched variable is a hard error unless an explicit `default` branch is authored (no implicit default, no implicit production):
+
+```yaml
+_select:
+  env: DEPLOY_ENVIRONMENT
+  cases:
+    production:
+      image: registry.example/scoring:prod
+    staging:
+      image: registry.example/scoring:staging
+  default:
+    image: python:3.12
+```
+
+The document must resolve to exactly one object, and its keys must be named parameters of the generated `__init__` of the class `from_config` was called on — `InitVar` pseudo-fields included, `ClassVar`s and `field(init=False)` excluded — so these files stay environment-only: pipeline concerns such as file paths, versioning, annotations, schedules, or subscriptions have no field to land in and are rejected with the allowed field names. `from_config` is generic — any `TaskEnv` dataclass subclass inherits it unchanged, returns its own type, and is validated against its own fields:
+
+```python
+from tangle_cli.python_pipeline import TaskEnv
+
+@dataclass(frozen=True)
+class GpuEnv(TaskEnv):
+    accelerator: str = ""
+
+    def __post_init__(self):
+        super().__post_init__()
+        if self.accelerator not in ("gpu", "tpu"):
+            raise ValueError("GpuEnv.accelerator must be one of: gpu, tpu")
+
+GPU = GpuEnv.from_config("envs.yaml")  # accepts image, dependencies_from, accelerator
+```
+
+Failures raise `CompileError` naming the resolved config path. A config file is untrusted input, so **no diagnostic echoes a config value**. Keys are rendered through a capped, control-character-scrubbing renderer, and a constructor's own validation text is never quoted — it could embed a value directly, nested inside a structure, or transformed (lower cased, sliced, re-encoded). The rejected exception is also kept off `__cause__`/`__context__`, since `traceback.format_exception` would otherwise print it into the same CI log. The message names the class and the fields present, and points at constructing the class directly to see the validation error:
+
+```
+GpuEnv.from_config: /repo/envs.yaml case is not a valid GpuEnv (fields present:
+accelerator, image). Its validation message is withheld because it can contain
+config values; construct GpuEnv(...) directly to see it.
+```
+
+Calling `GpuEnv(...)` directly is unaffected and raises the ordinary `ValueError` with its full message.
+
+See `examples/python_pipeline/task_env_from_config/` for a runnable example.
+
 Use `@task(image_id="eval-slim")` when source should carry a logical image name instead of a concrete registry reference. Downstream code can register defaults with `register_image_id(...)`, and callers can override at compile time with repeatable `--image ID=REF`:
 
 ```bash

examples/python_pipeline/task_env_from_config/envs.yaml

@@ -0,0 +1,25 @@
+# Deployment environments for the tasks in ``pipeline.py``.
+#
+# ENVIRONMENT-ONLY on purpose: the keys here are exactly the ``TaskEnv``
+# constructor fields (``image``, ``dependencies_from``). Pipeline concerns —
+# file paths, versioning, annotations, schedules, subscriptions — have no
+# field to land in and are rejected at load time.
+#
+# ``_select`` is the same directive the CLI's ``--config`` understands: the
+# case is chosen by an environment variable, exactly and case sensitively.
+# The explicit ``default`` branch is what makes local runs work; without one,
+# an unset or unmatched ``DEPLOY_ENVIRONMENT`` is a hard error (there is no
+# implicit default and no implicit production).
+_select:
+  env: DEPLOY_ENVIRONMENT
+  cases:
+    production:
+      image: registry.example/scoring:prod
+      # Relative to THIS file, not to the pipeline module and not to the cwd.
+      dependencies_from: pyproject.toml
+    staging:
+      image: registry.example/scoring:staging
+      dependencies_from: pyproject.toml
+  default:
+    image: python:3.12
+    dependencies_from: pyproject.toml

examples/python_pipeline/task_env_from_config/pipeline.py

@@ -0,0 +1,45 @@
+"""Runnable example: load a ``TaskEnv`` from a config file.
+
+``TaskEnv.from_config`` declares the execution environment once, in
+``envs.yaml``, instead of repeating an image string across tasks. The path is
+relative to THIS file, so the same script picks the same config wherever
+``tangle`` is run from.
+
+Compile from the repository root with::
+
+    uv run tangle sdk pipelines compile \
+      examples/python_pipeline/task_env_from_config/pipeline.py \
+      --pipeline scoring_pipeline \
+      --output /tmp/tangle-task-env-from-config-demo/pipeline.yaml
+
+The config uses ``_select`` over ``DEPLOY_ENVIRONMENT``; with the variable
+unset the ``default`` case applies, so the command above works as written::
+
+    DEPLOY_ENVIRONMENT=production uv run tangle sdk pipelines compile ...
+"""
+
+from tangle_cli.python_pipeline import Out, TaskEnv, pipeline, task
+
+# One declaration, reused by every task below. Nothing about this is specific
+# to a kind of environment: any ``TaskEnv`` dataclass subclass inherits
+# ``from_config`` and is validated against its own fields.
+SCORING = TaskEnv.from_config("envs.yaml")
+
+
+@task(env=SCORING)
+def load_queries(count: str = "3") -> str:
+    """Produce the queries to score."""
+    return ",".join(f"query-{index}" for index in range(int(count)))
+
+
+@task(env=SCORING)
+def score_queries(queries: str) -> str:
+    """Score the queries on the same image, without repeating it."""
+    return ";".join(f"{query}=1.0" for query in queries.split(","))
+
+
+@pipeline("TaskEnv from config demo")
+def scoring_pipeline() -> Out[str]:
+    queries = load_queries(count="3")
+    scored = score_queries(queries=queries.output)
+    return scored.output

examples/python_pipeline/task_env_from_config/pyproject.toml

@@ -0,0 +1,5 @@
+# Pip dependencies for the generated components of this example.
+[project]
+name = "task-env-from-config-example"
+version = "0.0.0"
+dependencies = []

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

@@ -14,6 +14,6 @@
 try:
     __version__ = metadata_version("tangle-cli")
 except PackageNotFoundError:
-    __version__ = "0.1.16"
+    __version__ = "0.1.17"
 
 __all__ = ["TangleDynamicDiscoveryClient", "__version__"]

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

@@ -41,10 +41,17 @@ class ConfigFileError(Exception):
     """Raised when there is an error loading or resolving a config file."""
 
 
-def _render_case_key(key: str) -> str:
-    """Render an already-validated configured case key for diagnostics."""
+def _render_config_key(key: Any) -> str:
+    """Render a config-document key safely for a diagnostic message.
 
-    rendered = "".join(char if char.isprintable() else "?" for char in key)
+    Config keys come from user files, so a diagnostic must never echo them
+    verbatim: non-printable characters are replaced and the result is length
+    capped before being quoted. Accepts non-string keys (YAML permits them) so
+    callers can report a bad key without first proving it is a string.
+    """
+
+    text = key if isinstance(key, str) else str(key)
+    rendered = "".join(char if char.isprintable() else "?" for char in text)
     if len(rendered) > _MAX_RENDERED_CASE_KEY_LENGTH:
         rendered = rendered[: _MAX_RENDERED_CASE_KEY_LENGTH - 3] + "..."
     return repr(rendered)
@@ -172,8 +179,14 @@ def _validate_selector_node(
             if seen is not None and depth <= seen[0]:
                 return seen[2]
 
+        # Every key reaching a diagnostic goes through the ONE capped/scrubbed
+        # renderer: these keys are user input, and an unbounded repr() would
+        # let a hostile document paste control characters or a wall of text
+        # into a compile/CI log.
         siblings = sorted(
-            repr(key) for key in node if not (isinstance(key, str) and key.startswith("_"))
+            _render_config_key(key)
+            for key in node
+            if not (isinstance(key, str) and key.startswith("_"))
         )
         if siblings:
             raise ConfigFileError(
@@ -189,7 +202,9 @@ def _validate_selector_node(
         selector_dict = cast(dict[Any, Any], selector)
 
         unexpected = sorted(
-            repr(key) for key in selector_dict if key not in ("env", "cases", "default")
+            _render_config_key(key)
+            for key in selector_dict
+            if key not in ("env", "cases", "default")
         )
         if unexpected:
             raise ConfigFileError(
@@ -231,7 +246,7 @@ def _validate_selector_node(
             # Every branch document is shape-checked, not just the selected one,
             # so a malformed selector fails identically in every environment.
             ArgsContainer._validate_branch_document(
-                case_value, f"{SELECT_KEY} case {_render_case_key(case_key)}", depth, memo
+                case_value, f"{SELECT_KEY} case {_render_config_key(case_key)}", depth, memo
             )
 
         default_branch: Any = None
@@ -241,7 +256,7 @@ def _validate_selector_node(
                 default_branch, f"{SELECT_KEY}.default", depth, memo
             )
 
-        allowed = ", ".join(_render_case_key(key) for key in sorted(cases_dict))
+        allowed = ", ".join(_render_config_key(key) for key in sorted(cases_dict))
         summary: _SelectorSummary = (env_name, cases_dict, default_branch, allowed)
         if memo is not None:
             previous = memo.get(node_id)