Declare graph inputs and outputs from the pipeline body (v0.1.23)

README.md

@@ -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.
 
+##### Declaring graph inputs and outputs from the body
+
+A pipeline's inputs normally come from its `In[T]` parameters and its outputs from the return annotation. That only covers graphs whose I/O is Python-shaped. `graph_input()` / `graph_output()` declare the same entries directly on the active pipeline, for the shapes a signature cannot express:
+
+```python
+from tangle_cli.python_pipeline import Out, graph_input, graph_output, pipeline
+
+@pipeline("Daily Pulse")
+def daily_pulse() -> Out[str]:
+    created = graph_input(
+        "Pipeline Creation Time", "String", default="Use as a cache busting mechanism"
+    )
+    limit = graph_input("query_limit", "Integer", optional=True, when=has_limit)
+    scrape = SCRAPE.named("Scrape").with_position(0, 0)(created=created, limit=limit)
+    graph_output("scrape_date", scrape.scrape_date, "String")
+    return scrape
+```
+
+- **A name is a string, not an identifier.** `"Pipeline Creation Time"` is a legal Tangle input name and an illegal Python parameter name; same for a type that is a Tangle type string (`"Json"`) rather than a Python type.
+- **`when=False` declares nothing** and returns `None`, so a conditional input stays a single assignment instead of an `if`/`else` around the call site.
+- **`default` implies `optional: true`** unless `optional=False` is passed. This matches what a defaulted `In[T]` parameter emits; the corpus also contains defaulted-but-required inputs, which is why the implication is overridable.
+- **Declaration order is signature parameters first, then body order**, and it is the order the entries appear in the document.
+- **A name is declared once.** Duplicates are rejected, including a collision with an `In[...]` parameter or with the name the return annotation contributes (`Out[T]`'s `output_name`, or an `Outputs` field).
+- **Outputs wire to a handle, never a constant** — a task output or a graph input, the same rule the `Outputs` return path enforces. Returning a value *and* declaring outputs is supported: the returned output is appended after the declared ones.
+- **Layout comes along**: `position=(x, y)` writes the same canonical `editor.position` annotation as `.with_position(...)`, which is how a graph input gets a position in the editor. `annotations={...}` is validated under the same caller policy as `pipeline_annotations`.
+- **Validated up front, without echoing values.** A non-string `default` (the schema's `InputSpec.default` is a string), a non-string name/type/description, a non-boolean `optional`, a constant output, or a call outside a `@pipeline` body raises `InvalidGraphIoError` (a `CompileError`) naming the field, never the value.
+
+The signature route remains the recommended default: it is typed, it is checked by your IDE, and it reads better. Reach for these when porting existing YAML or when the shape genuinely needs it.
+
+Porting an existing helper that appended to the builder by hand (for example relevance-tools' `_gin` / `_gout`): the emitted document is identical except in two places. An input declared with a `default` and no explicit `optional` gains `optional: true`, because that is what a defaulted `In[T]` parameter already emits and what the majority of the corpus carries; pass `optional=False` to keep it required. An input declaring both keys emits `default` before `optional`, which is the order the corpus uses 28 times against 3 for the reverse.
+
 ##### 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`:
@@ -630,7 +661,7 @@ SCRAPE.with_annotations({"editor.position": '{"x": 0, "y": 0}'})
 - **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.
+Graph **inputs** and **outputs** carry `editor.position` in the same way. An `In[T]` parameter has nowhere to hang layout, so positioning those means declaring them with `graph_input(..., position=(x, y))` instead — see the section above.
 
 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`.
 

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

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

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

@@ -21,6 +21,7 @@
 from __future__ import annotations
 
 from .dynamic_data import dynamic_secret
+from .graph_io import graph_input, graph_output
 from .pipeline import pipeline
 from .raw import raw
 from .ref import ref
@@ -33,6 +34,8 @@
 
 __all__ = [
     "pipeline",
+    "graph_input",
+    "graph_output",
     "task",
     "Publish",
     "registered",

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

@@ -38,6 +38,15 @@ class InvalidEditorLayoutError(CompileError):
     """
 
 
+class InvalidGraphIoError(CompileError):
+    """Raised on an unusable ``graph_input()`` / ``graph_output()`` call.
+
+    Covers declaration outside a trace, duplicate names, fields the pipeline
+    schema would reject, and constant graph outputs. Messages name the field,
+    never the value.
+    """
+
+
 class InvalidPipelineAnnotationsError(CompileError):
     """Raised on a malformed caller-supplied ``pipeline_annotations`` mapping.
 

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

@@ -0,0 +1,214 @@
+"""``graph_input()`` / ``graph_output()`` — declare graph I/O from the body.
+
+A pipeline's inputs normally come from its ``In[T]`` parameters and its
+outputs from the return annotation. That covers Python-shaped graphs only:
+it cannot express a name that is not an identifier (``"Pipeline Creation
+Time"``), an exact Tangle type string, an input declared conditionally, or
+a per-input ``editor.position``. These two functions declare the same
+entries directly on the active trace, so a YAML pipeline that uses those
+shapes stays expressible in Python.
+
+Entry key order follows the tracer's signature path and the corpus:
+``name, type, description, default, optional, annotations``.
+"""
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from tangle_cli.editor_layout import POSITION_ANNOTATION, position_annotation_value
+from tangle_cli.schema_validation import CALLER_ANNOTATION_POLICY, check_annotations
+
+from .errors import InvalidEditorLayoutError, InvalidGraphIoError
+from .graph import EdgeRef
+from .placeholders import GraphInputPlaceholder, TaskOutputProxy
+
+# Distinguishes "no default" from an explicit ``default=None``, which is a
+# value the schema does not accept and must be reported, not dropped.
+_UNSET: Any = object()
+
+__all__ = ["graph_input", "graph_output"]
+
+
+def graph_input(
+    name: str,
+    type: str | None = None,
+    *,
+    description: str | None = None,
+    default: Any = _UNSET,
+    optional: bool | None = None,
+    annotations: Mapping[str, str] | None = None,
+    position: tuple[float, float] | None = None,
+    when: bool = True,
+) -> GraphInputPlaceholder | None:
+    """Declare a graph input on the active pipeline and return its handle.
+
+    The handle is passed to task arguments exactly like an ``In[T]``
+    parameter. Supplying ``default`` implies ``optional: true`` unless
+    ``optional=False`` is passed explicitly.
+
+    ``when=False`` declares nothing and returns ``None``, so a conditional
+    input reads as one assignment instead of an ``if``/``else``.
+
+    Raises:
+        InvalidGraphIoError: Outside a ``@pipeline`` body, on a duplicate
+            name, or on an unusable field. Messages name the field, never
+            the value.
+    """
+    builder = _active_builder("graph_input")
+    if not when:
+        return None
+
+    _check_name(name, "graph_input")
+    if any(entry.get("name") == name for entry in builder.inputs):
+        # Also catches an ``In[...]`` parameter: signature inputs are
+        # appended before the body runs.
+        raise InvalidGraphIoError(
+            f"graph input {name!r} is already declared on pipeline "
+            f"{builder.name!r}. Input names must be unique."
+        )
+
+    entry: dict[str, Any] = {"name": name}
+    if type is not None:
+        entry["type"] = _check_str("type", type)
+    if description is not None:
+        entry["description"] = _check_str("description", description)
+    if default is not _UNSET:
+        # InputSpec.default is a string in the schema; a non-string default
+        # would only fail later, against a schema path instead of this call.
+        entry["default"] = _check_str("default", default, hint="pass str(value)")
+        entry["optional"] = True if optional is None else _check_bool(optional)
+    elif optional is not None:
+        entry["optional"] = _check_bool(optional)
+    merged = _layout_annotations(annotations, position)
+    if merged:
+        entry["annotations"] = merged
+
+    builder.inputs.append(entry)
+    return GraphInputPlaceholder(input_name=name)
+
+
+def graph_output(
+    name: str,
+    value: Any,
+    type: str | None = None,
+    *,
+    description: str | None = None,
+    annotations: Mapping[str, str] | None = None,
+    position: tuple[float, float] | None = None,
+    when: bool = True,
+) -> None:
+    """Declare a graph output on the active pipeline, wired to ``value``.
+
+    ``value`` is a task output handle or a graph-input handle; a constant is
+    refused, matching the ``Outputs`` return path.
+
+    Raises:
+        InvalidGraphIoError: Outside a ``@pipeline`` body, on a duplicate
+            name, on an unusable field, or on a constant value.
+    """
+    builder = _active_builder("graph_output")
+    if not when:
+        return
+
+    _check_name(name, "graph_output")
+    if any(entry.get("name") == name for entry in builder.outputs):
+        raise InvalidGraphIoError(
+            f"graph output {name!r} is already declared on pipeline "
+            f"{builder.name!r}. Output names must be unique."
+        )
+
+    entry: dict[str, Any] = {"name": name}
+    if type is not None:
+        entry["type"] = _check_str("type", type)
+    if description is not None:
+        entry["description"] = _check_str("description", description)
+    merged = _layout_annotations(annotations, position)
+    if merged:
+        entry["annotations"] = merged
+
+    builder.outputs.append(entry)
+    builder.output_values[name] = _edge(name, value)
+
+
+def _active_builder(caller: str) -> Any:
+    from .trace import current_builder
+
+    builder = current_builder()
+    if builder is None:
+        raise InvalidGraphIoError(
+            f"{caller}() must be called inside a @pipeline function body, "
+            "where a trace is active."
+        )
+    return builder
+
+
+def _edge(name: str, value: Any) -> EdgeRef:
+    if isinstance(value, TaskOutputProxy):
+        return EdgeRef(
+            kind="taskOutput",
+            task_id=value._task_id,
+            output=value._resolved_output_name(),
+        )
+    if isinstance(value, GraphInputPlaceholder):
+        return EdgeRef(kind="graphInput", input_name=value.input_name)
+    raise InvalidGraphIoError(
+        f"graph output {name!r} must be wired to a task output or a graph "
+        f"input; got {type(value).__name__}. Constant graph outputs are not "
+        "supported — emit the value from a task."
+    )
+
+
+def _layout_annotations(
+    annotations: Mapping[str, str] | None,
+    position: tuple[float, float] | None,
+) -> dict[str, str]:
+    merged: dict[str, str] = {}
+    if annotations is not None:
+        if not isinstance(annotations, Mapping):
+            raise InvalidGraphIoError(
+                f"annotations must be a mapping; got {type(annotations).__name__}."
+            )
+        check_annotations(
+            annotations,
+            policy=CALLER_ANNOTATION_POLICY,
+            error_cls=InvalidGraphIoError,
+        )
+        merged.update(annotations)
+    if position is not None:
+        # Applied after the mapping, matching ``@pipeline(flow_direction=...)``.
+        if not isinstance(position, tuple) or len(position) != 2:
+            raise InvalidEditorLayoutError(
+                "position must be an (x, y) tuple; got "
+                f"{type(position).__name__}."
+            )
+        merged[POSITION_ANNOTATION] = position_annotation_value(
+            *position, error_cls=InvalidEditorLayoutError
+        )
+    return merged
+
+
+def _check_name(name: Any, caller: str) -> str:
+    if not isinstance(name, str) or not name:
+        raise InvalidGraphIoError(
+            f"{caller}() name must be a non-empty string; got "
+            f"{type(name).__name__}."
+        )
+    return name
+
+
+def _check_str(field: str, value: Any, *, hint: str | None = None) -> str:
+    if not isinstance(value, str):
+        suffix = f". {hint.capitalize()}." if hint else "."
+        raise InvalidGraphIoError(
+            f"{field} must be a string; got {type(value).__name__}{suffix}"
+        )
+    return value
+
+
+def _check_bool(value: Any) -> bool:
+    if not isinstance(value, bool):
+        raise InvalidGraphIoError(
+            f"optional must be True or False; got {type(value).__name__}."
+        )
+    return value

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

@@ -285,13 +285,20 @@ def trace_pipeline(
         inner = _annotation_inner_type(return_anno)
         type_str = _python_type_to_tangle_type(inner)
         output_name = pipeline_fn.output_name
+        _ensure_no_duplicate_output(builder, output_name)
         builder.outputs.append({"name": output_name, "type": type_str})
         builder.output_name = output_name
-        builder.output_taskref = EdgeRef(
+        edge = EdgeRef(
             kind="taskOutput",
             task_id=result._task_id,
             output=result._resolved_output_name(),
         )
+        builder.output_taskref = edge
+        if builder.output_values:
+            # graph_output() already populated the multi-output map, which
+            # emit PREFERS; the returned output has to join it or it would be
+            # silently dropped. Declared last, after the body's own outputs.
+            builder.output_values[output_name] = edge
     elif return_anno is not inspect.Signature.empty and _is_outputs_class(return_anno):
         _trace_multi_output(builder, pipeline_fn, return_anno, result)
 
@@ -355,9 +362,22 @@ def _trace_multi_output(
                 "outputs."
             )
         type_str = _python_type_to_tangle_type(inner)
+        _ensure_no_duplicate_output(builder, field_name)
         builder.outputs.append({"name": field_name, "type": type_str})
         builder.output_values[field_name] = edge
 
 
+def _ensure_no_duplicate_output(builder: GraphBuilder, name: str) -> None:
+    """Reject a return-annotation output that ``graph_output()`` already
+    declared, instead of emitting the name twice."""
+    if any(entry.get("name") == name for entry in builder.outputs):
+        from .errors import InvalidGraphIoError
+
+        raise InvalidGraphIoError(
+            f"graph output {name!r} is declared by graph_output() and by the "
+            f"return annotation of pipeline {builder.name!r}. Declare it once."
+        )
+
+
 # Re-export commonly-used names.
 __all__ = ["current_builder", "trace_pipeline"]

pyproject.toml

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