Workflow variable types and explicit boxes¶
define_variable[T](name) captures a Python annotation and gives .quote the
static type T. It does not turn a Frame into a schema validator. A quote is a
definition marker; use the actual value only after argument materialization.
Choose a declaration¶
| Type family | Recommended declaration | Runtime behavior |
|---|---|---|
| Scalars and application classes | define_variable[int], define_variable[Path] |
Preserve the object; no coercion. |
| Containers and abstract collections | define_variable[list[str]], define_variable[Collection[str]] |
Preserve containers and elements by reference. |
| Optional and union values | define_variable[str \| None] |
Preserve the chosen value, including None. |
| Callbacks | define_variable[Callable[[str], int]] |
Retain the callable; invocation is explicit. |
| Protocol values | define_variable[Reader] |
No implicit structural validation. |
| Concrete or generic dataclasses | define_variable[Options], define_variable[Batch[str]] |
Whole mappings require an exactly matching record annotation; scopes construct that record. |
| Open values | define_variable[object] or define_variable[Any] |
Prefer a narrower type when known. |
| Null-only values | define_variable[None] |
None is a value, not a missing binding. |
Container elements, Literal constraints and structural typing rules are not
recursively checked. Validate external data in application code.
from collections.abc import Callable, Collection
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Protocol, assert_type
from lclang.workflow import define_variable
class Reader(Protocol):
def read(self) -> str: ...
@dataclass
class Batch[T]:
items: list[T]
assert_type(define_variable[Path]("path").quote, Path)
assert_type(define_variable[Collection[str]]("items").quote, Collection[str])
assert_type(define_variable[str | None]("optional").quote, str | None)
assert_type(define_variable[Callable[[str], int]]("callback").quote, Callable[[str], int])
assert_type(define_variable[Reader]("reader").quote, Reader)
assert_type(define_variable[object]("opaque").quote, object)
assert_type(define_variable[Any]("legacy").quote, Any)
assert_type(define_variable[None]("null").quote, None)
batch = define_variable[Batch[str]]("batch")
assert_type(batch.quote, Batch[str])
assert batch.value_type == Batch[str]
Use direct declarations for independent inputs. Generic subscription accepts
more annotation forms than .field(name, field_type). Avoid wrappers when a
direct variable already expresses the data flow.
Project concrete wrapper types¶
.field() keeps type[T] and zero runtime dependencies. It verifies the field
name and exact declared annotation. Ordinary classes and suitable parameterized
concrete types work directly. Bare Protocol classes and union expressions are
not a portable strict-typing interface for this parameter.
ValueBox[T] gives an optional value, collection or Protocol a concrete wrapper.
CallableBox[P, R] forwards a callback with its regular callable signature.
Both expose only one data field, value, and are available from lclang.utils.
Declare the wrapper in the source record as well as the projection.
import asyncio
import logging
from dataclasses import dataclass
from datetime import date
from lclang.lang import define_frame
from lclang.utils import CallableBox, ValueBox
from lclang.workflow import (
ExecutionStatusManager, TaskContext, WorkflowExecutionContext,
define_task, define_variable, define_workflow,
)
@dataclass
class Settings:
rename: ValueBox[str | None]
parse: CallableBox[[str], int]
async def use_settings(
context: TaskContext, args: Settings, status_mgr: ExecutionStatusManager,
) -> Settings:
assert args.rename.value is None
assert args.parse("17") == 17
return args
async def main() -> None:
source = define_variable[Settings]("settings")
output = define_variable[ValueBox[str | None]]("result")
mapping = Settings(
source.field("rename", ValueBox[str | None]).quote,
source.field("parse", CallableBox[[str], int]).quote,
)
task = define_task(
"use", "Use settings", task_action=use_settings, args_mapping=mapping,
outputs_mapping=Settings(output.quote, CallableBox(int)),
)
async with define_frame(preset={"settings.rename": None, "settings.parse": int}) as frame:
result = await define_workflow("Settings", task).execute(WorkflowExecutionContext(
False, date(2026, 9, 23), False, logging.getLogger("boxes-example"), frame,
))
assert await frame.get("result") is None
assert await frame.get("settings.parse") is int
assert result.task_outputs["use"].rename.value is None
asyncio.run(main())
Selected projections box raw scope values. Only result is an output quote,
so it publishes its payload None. The literal callback in the output template
is unmapped. The returned action record retains its boxes.
Preserve Protocol contracts explicitly¶
A box never checks isinstance(payload, Protocol). Hold a non-callable Protocol
in ValueBox[ProtocolType] and use .value. For callable Protocols with overloads,
keywords or additional members, use a named subclass with explicit delegation.
from typing import Protocol
from lclang.utils import CallableBox, ValueBox
class Rule(Protocol):
def __call__(self, text: str, *, upper: bool = False) -> str: ...
class RuleBox(ValueBox[Rule]):
def __call__(self, text: str, *, upper: bool = False) -> str:
return self.value(text, upper=upper)
def format_text(text: str, *, upper: bool = False) -> str:
return text.upper() if upper else text
assert RuleBox(format_text)("hello", upper=True) == "HELLO"
assert CallableBox(format_text)("hello", upper=True) == "HELLO"
shared = ["a", "b"]
assert ValueBox(shared).value is shared
ParamSpec preserves the ordinary callback's keyword signature. The named subclass also gives the Protocol a concrete identity. Add overloads and other members explicitly when needed; no universal Protocol proxy is generated.
Conversion cost and unsupported uses¶
- Reading one explicitly boxed variable or projection normally allocates one wrapper holding one reference: constant work and storage relative to payload size. Re-reading may allocate another wrapper; the payload remains shared.
- Matching boxes are reused. Incompatible boxes are rejected, not double-boxed. Only the outer wrapper type is checked; payload annotations are not validators.
- Publishing an explicitly boxed output reads
.valueonce. There is no deep copy, container walk or mutation. Returned task outputs keep their wrappers. - Callable boxes add one ordinary Python call, with no task, implicit await, resource ownership or exception translation.
- Ordinary records, whole-record contents and container elements are not searched for wrappers. For raw scoped fields requiring boxes, use explicit projections as above. Whole-record quotes do not recursively box their fields.
Non-dataclass whole mappings, output projections, projections from union or
non-record roots, mismatched field annotations and cyclic templates are
unsupported. A bare non-runtime-checkable Protocol projection can fail the
existing outer-shape check even if static errors are bypassed; use boxes.
TypedDict, forward-reference strings and special typing forms are not a
general runtime validation language. Resolve record annotations normally and
validate external data explicitly. See mapping best practices.
See errors and diagnostic codes for exception fields, specific
codes, cause chains, and executable troubleshooting examples. Ordinary library
validation uses LclValidationError; native callback and operation failures
retain their original exception in __cause__.