TangleML · Volv-G · Sep 25, 2026 · Sep 25, 2026
@@ -603,6 +603,37 @@ The rules live in one place, `tangle_cli.schema_validation`: `check_annotations(
 
 A distribution that reads these annotations from its own config file should call `check_annotations(mapping, policy=CALLER_ANNOTATION_POLICY, error_cls=...)` at config-parse time — passing its own error type and adding the config path and key to the message — so one user mistake produces one diagnostic instead of two competing ones. The compiler's own call is then the backstop for anything arriving by another route.
 
+##### Editor layout (`editor.*`)
+
+The pipeline editor stores graph layout in ordinary annotations, so layout is authorable from Python. `.with_position(x, y)` on a task or subpipeline handle writes `editor.position`, and `@pipeline(flow_direction=...)` writes the root `editor.flow-direction`:
+
+```python
+@pipeline("Daily Pulse", flow_direction="left-to-right")
+def daily_pulse() -> Out[str]:
+    scrape = SCRAPE.named("Scrape").with_position(0, 0)()
+    judge = JUDGE.named("Judge").with_position(300, 0)(rows=scrape.rows)
+    return judge.report
+```
+
+Both are pure sugar over annotations you can still write by hand, and they compile to byte-identical output:
+
+```python
+# Equivalent, and what the sugar emits:
+SCRAPE.with_annotations({"editor.position": '{"x": 0, "y": 0}'})
+@pipeline("Daily Pulse", annotations={"editor.flow-direction": "left-to-right"})
+```
+
+- **Value format is the editor's, not ours.** `editor.position` is a JSON object *string* (`'{"x": 300, "y": 120}'`) because the editor writes `JSON.stringify` and reads `JSON.parse`. Coordinates may be negative. Optional `width=` / `height=` keywords add the node-size fields the editor also reads; they are omitted entirely when not given.
+- **`flow_direction`** accepts `"left-to-right"` (what the editor writes, and what it renders today) and `"top-to-bottom"` (legacy, accepted so existing documents stay expressible). There is no default: omitting the keyword emits no annotation.
+- **Last write wins**, one rule for both spellings. `.with_position(...)` *is* a `.with_annotations({"editor.position": ...})` call, so whichever runs last sets the value, and unrelated annotations are untouched. Within a single `@pipeline(...)`, the typed `flow_direction` keyword is applied after the `annotations` mapping and therefore wins on that key.
+- **Validated without echoing values.** A non-numeric, `bool`, `NaN` or infinite coordinate, or an unknown flow direction, raises `InvalidEditorLayoutError` (a `CompileError`) at the call, before anything is written. Diagnostics name the coordinate or list the allowed directions.
+- **Positions are descriptive.** They are task annotations, not part of any `componentRef`, so they never change component digests, compile identity, or cache behaviour. On a subpipeline handle the position applies to the parent task; the child sidecar is byte-identical either way.
+- **Auto-layout interaction.** At submit time the runner lays a graph out only when no task carries a *non-zero* position, so an explicit `.with_position(...)` suppresses auto-layout and `--force-layout` overrides that. A graph positioned entirely at `(0, 0)` still counts as unpositioned and is laid out. `tangle sdk pipelines layout` writes the same canonical value this sugar writes.
+
+Graph **inputs** and **outputs** carry `editor.position` in the same way, and the editor reads it there, but there is no Python authoring surface for a graph input/output object yet — inputs come from the `In[T]` signature and outputs from the return annotation, neither of which has a place to hang layout. That is follow-up work tied to a public `graph_input()` / `graph_output()` API; until then, position inputs by editing the YAML or by using the editor.
+
+The key names, the serialized format, and the validation live in `tangle_cli.editor_layout` (`POSITION_ANNOTATION`, `FLOW_DIRECTION_ANNOTATION`, `FLOW_DIRECTIONS`, `position_annotation_value`, `validate_flow_direction`), which the CLI's own `pipelines layout` and the runner's auto-layout gate share, so a tool emitting layout cannot drift from what the editor reads. That module is stdlib-only and lives outside `python_pipeline` on purpose, so layout and submit paths do not pull in the authoring/codegen stack; it raises whatever `error_cls` the caller injects, and the authoring surfaces inject `InvalidEditorLayoutError`.
+
 ##### Conditional task execution
 
 Pipeline inputs used as conditions are ordinary `In[str]` values; there is no special conditional input annotation. Pass the value through the reserved task-call metadata keyword `is_enabled=`:

@@ -14,6 +14,6 @@
 try:
     __version__ = metadata_version("tangle-cli")
 except PackageNotFoundError:
-    __version__ = "0.1.21"
+    __version__ = "0.1.22"
 
 __all__ = ["TangleDynamicDiscoveryClient", "__version__"]
@@ -0,0 +1,83 @@
+"""Canonical ``editor.*`` layout keys, value formats and validation.
+
+Layout is descriptive: it never affects execution, component digests or the
+cache. Stdlib-only and outside :mod:`tangle_cli.python_pipeline` so the
+layout command and the submit-time gate import it without the authoring
+stack; callers inject their own ``error_cls``.
+"""
+from __future__ import annotations
+
+import json
+import math
+from typing import Any
+
+#: Task / graph-input / graph-output coordinates. The value is a JSON object
+#: STRING (the editor stringifies and parses it), ``{"x", "y"}`` plus optional
+#: ``"width"``/``"height"``; coordinates may be negative.
+POSITION_ANNOTATION = "editor.position"
+
+#: Root ``metadata.annotations`` rendering direction.
+FLOW_DIRECTION_ANNOTATION = "editor.flow-direction"
+
+#: Accepted ``editor.flow-direction`` values; ``top-to-bottom`` is legacy.
+FLOW_DIRECTIONS: tuple[str, ...] = ("left-to-right", "top-to-bottom")
+
+
+def position_annotation_value(
+    x: float,
+    y: float,
+    *,
+    width: float | None = None,
+    height: float | None = None,
+    error_cls: type[Exception] = ValueError,
+) -> str:
+    """Return the canonical ``editor.position`` value for ``x`` / ``y``.
+
+    Key order is ``x, y, width, height``; absent dimensions are omitted
+    rather than written as null.
+
+    Raises:
+        error_cls: If a coordinate is not a finite real number. Messages name
+            the coordinate and its type, never the value.
+    """
+
+    position: dict[str, float] = {
+        "x": _finite_number("x", x, error_cls),
+        "y": _finite_number("y", y, error_cls),
+    }
+    if width is not None:
+        position["width"] = _finite_number("width", width, error_cls)
+    if height is not None:
+        position["height"] = _finite_number("height", height, error_cls)
+    return json.dumps(position)
+
+
+def validate_flow_direction(
+    value: Any, *, error_cls: type[Exception] = ValueError
+) -> str:
+    """Return ``value`` if it is a supported ``editor.flow-direction``.
+
+    Raises:
+        error_cls: If it is not one of :data:`FLOW_DIRECTIONS`.
+    """
+
+    if value in FLOW_DIRECTIONS:
+        return str(value)
+    allowed = ", ".join(repr(direction) for direction in FLOW_DIRECTIONS)
+    raise error_cls(
+        f"flow_direction must be one of {allowed}; got "
+        f"{type(value).__name__}."
+    )
+
+
+def _finite_number(name: str, value: Any, error_cls: type[Exception]) -> float:
+    # bool is an int subclass, and True would silently serialize as 1.
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise error_cls(
+            f"position coordinate {name!r} must be a real number; got "
+            f"{type(value).__name__}."
+        )
+    if not math.isfinite(value):
+        # json.dumps emits bare NaN / Infinity, which JSON.parse rejects.
+        raise error_cls(f"position coordinate {name!r} must be finite.")
+    return value
@@ -16,6 +16,7 @@
 from pathlib import Path
 from typing import Any, Mapping
 
+from .editor_layout import POSITION_ANNOTATION
 from .pipeline_run_manager import (
     PipelineRunContext,
     PipelineRunError,
@@ -124,7 +125,7 @@ def has_layout(self, pipeline_spec: Mapping[str, Any]) -> bool:
             if not isinstance(task, Mapping):
                 continue
             annotations = task.get("annotations", {})
-            position = annotations.get("editor.position") if isinstance(annotations, Mapping) else None
+            position = annotations.get(POSITION_ANNOTATION) if isinstance(annotations, Mapping) else None
             if isinstance(position, str):
                 try:
                     import json

@@ -17,6 +17,7 @@
 
 import yaml
 
+from .editor_layout import POSITION_ANNOTATION
 from .pipeline_spec_utils import _extract_task_output_refs
 from .pipeline_validation import (
     PipelineValidationError,
@@ -31,8 +32,6 @@
 if TYPE_CHECKING:
     from .pipeline_compiler import CompileResult
 
-POSITION_ANNOTATION = "editor.position"
-
 __all__ = [
     "HydrateResult",
     "LayoutResult",

@@ -29,6 +29,15 @@ class InvalidArgumentTypeError(CompileError):
     """Raised on an argument value with no supported emit dispatch."""
 
 
+class InvalidEditorLayoutError(CompileError):
+    """Raised on a malformed value passed to the typed layout sugar.
+
+    Covers ``.with_position`` coordinates and
+    ``@pipeline(flow_direction=...)`` only; hand-written
+    ``.with_annotations`` values still pass through unchecked.
+    """
+
+
 class InvalidPipelineAnnotationsError(CompileError):
     """Raised on a malformed caller-supplied ``pipeline_annotations`` mapping.
 

@@ -13,7 +13,13 @@
 from pathlib import Path
 from typing import Any, Callable
 
+from tangle_cli.editor_layout import (
+    FLOW_DIRECTION_ANNOTATION,
+    validate_flow_direction,
+)
+
 from . import emit
+from .errors import InvalidEditorLayoutError
 from .graph import GraphBuilder
 
 
@@ -130,6 +136,7 @@ def pipeline(
     config: str | None = None,
     annotations: dict[str, Any] | None = None,
     task_annotations: dict[str, Any] | None = None,
+    flow_direction: str | None = None,
     output_name: str = "wait_for_output",
     propagate_config: bool = False,
 ) -> Callable[[Callable[..., Any]], PipelineFn]:
@@ -149,6 +156,10 @@ def pipeline(
             time so ``--override key=value`` pairs can merge in.
         annotations: ``metadata.annotations`` block (e.g. ``version``,
             ``author``).
+        flow_direction: Editor rendering direction, written as the
+            ``editor.flow-direction`` root annotation
+            (``"left-to-right"`` / ``"top-to-bottom"``). Sugar over
+            ``annotations``; applied last, so it wins on that key.
         task_annotations: Per-task default annotations applied to every
             task in the pipeline. Accepted for API completeness but not
             wired through in MVP — the PoC sets per-task annotations
@@ -168,12 +179,20 @@ def decorator(fn: Callable[..., Any]) -> PipelineFn:
             # Tests inside generated modules may not have a real file.
             caller_dir = None
 
+        merged_annotations = dict(annotations or {})
+        if flow_direction is not None:
+            # Assigned after the mapping: the typed keyword wins, and an
+            # existing key keeps its position in key order.
+            merged_annotations[FLOW_DIRECTION_ANNOTATION] = validate_flow_direction(
+                flow_direction, error_cls=InvalidEditorLayoutError
+            )
+
         return PipelineFn(
             fn=fn,
             name=name,
             description=description,
             config_path=config,
-            annotations=dict(annotations or {}),
+            annotations=merged_annotations,
             task_annotations=dict(task_annotations or {}),
             caller_dir=caller_dir,
             output_name=output_name,

@@ -15,7 +15,9 @@
 from pathlib import Path
 from typing import Any
 
-from .errors import CompileError
+from tangle_cli.editor_layout import POSITION_ANNOTATION, position_annotation_value
+
+from .errors import CompileError, InvalidEditorLayoutError
 from .graph import EXECUTION_OPTIONS_UNSET, IS_ENABLED_UNSET
 
 _UNWRAPPED_KEY_RE = re.compile(r"^[A-Za-z0-9_-]+$")
@@ -222,6 +224,36 @@ def with_annotations(self, ann: dict[str, Any]) -> "CallableRef":
             merged[k] = v  # type: ignore[assignment]
         return self._replace(annotations=merged)
 
+    def with_position(
+        self,
+        x: float,
+        y: float,
+        *,
+        width: float | None = None,
+        height: float | None = None,
+    ) -> "CallableRef":
+        """Return a new CallableRef carrying the ``editor.position``
+        annotation for ``x``/``y``, with optional node ``width``/``height``.
+
+        Sugar over :meth:`with_annotations`, which is what makes the last
+        write to ``editor.position`` win regardless of spelling.
+
+        Raises:
+            InvalidEditorLayoutError: On a non-numeric or non-finite
+                coordinate.
+        """
+        return self.with_annotations(
+            {
+                POSITION_ANNOTATION: position_annotation_value(
+                    x,
+                    y,
+                    width=width,
+                    height=height,
+                    error_cls=InvalidEditorLayoutError,
+                )
+            }
+        )
+
     # ------------------------------------------------------------------
     # @task codegen — materialize() writes the component YAML.
 

@@ -36,7 +36,9 @@
 from dataclasses import dataclass, field, replace
 from typing import TYPE_CHECKING, Any
 
-from .errors import CompileError
+from tangle_cli.editor_layout import POSITION_ANNOTATION, position_annotation_value
+
+from .errors import CompileError, InvalidEditorLayoutError
 
 if TYPE_CHECKING:  # pragma: no cover
     from .pipeline import PipelineFn
@@ -90,6 +92,32 @@ def with_annotations(self, ann: dict[str, Any]) -> "SubpipelineRef":
             merged[k] = v  # type: ignore[assignment]
         return replace(self, annotations=merged)
 
+    def with_position(
+        self,
+        x: float,
+        y: float,
+        *,
+        width: float | None = None,
+        height: float | None = None,
+    ) -> "SubpipelineRef":
+        """Return a new handle carrying ``editor.position`` for ``x``/``y``.
+
+        Positions the PARENT task in the enclosing graph — the same scope
+        :meth:`with_annotations` writes to; child layout is authored inside
+        the child pipeline. Otherwise as :meth:`CallableRef.with_position`.
+        """
+        return self.with_annotations(
+            {
+                POSITION_ANNOTATION: position_annotation_value(
+                    x,
+                    y,
+                    width=width,
+                    height=height,
+                    error_cls=InvalidEditorLayoutError,
+                )
+            }
+        )
+
     def override_config(self, **kwargs: Any) -> "SubpipelineRef":
         """Return a new handle with compile-time cfg overrides for the direct
         child merged in (later calls win on conflict).

@@ -1,6 +1,6 @@
 [project]
 name = "tangle-cli"
-version = "0.1.21"
+version = "0.1.22"
 description = "CLI for Tangle, the open-source ML pipeline orchestration platform"
 readme = "README.md"
 authors = [