fix(containers): preserve declared output types

ARCHITECTURE.md

@@ -1,5 +1,5 @@
 <!-- knowledge
-last_checked: "2026-10-29T01:02:46Z"
+last_checked: "2026-10-30T16:06:28Z"
 -->
 # Architecture
 
@@ -161,14 +161,34 @@ compatibility details:
 | File adapter binding and isolation | [File runtime](src/graphon/file/runtime.py), [engine](src/graphon/engine/engine.py) | [Execution file isolation](tests/engine/test_file_runtime.py), [file scopes](tests/file/test_runtime.py) |
 | Container ownership and factory preflight | [Graph and NodeFactory](src/graphon/graph/graph.py), [scoping](src/graphon/graph/scoping.py), [validation](src/graphon/graph/validation.py) | [Graph scoping and validation](tests/graph/test_graph_scoping.py) |
 | Branch joins and readiness | [Scheduler](src/graphon/engine/scheduler.py) | [Scheduling](tests/engine/test_scheduler.py) |
-| Container suspension and variable propagation | [Container effects](src/graphon/nodes/container_effects.py), [handlers](src/graphon/engine/container_handler/builtin/) | [Cooperative execution](tests/engine/test_cooperative_container_execution.py), [container variable scoping](tests/workflows/test_full_engine_events.py), [Iteration variable ownership](tests/engine/test_iteration_variable_scopes.py) |
+| Container suspension and variable propagation | [Container effects](src/graphon/nodes/container_effects.py), [handlers](src/graphon/engine/container_handler/builtin/) | [Container values](tests/nodes/test_container_effects.py), [cooperative execution](tests/engine/test_cooperative_container_execution.py), [container variable scoping](tests/workflows/test_full_engine_events.py), [Iteration variable ownership](tests/engine/test_iteration_variable_scopes.py) |
 | Pause, resume, and snapshot quiescence | [Dispatcher](src/graphon/engine/dispatcher.py), [execution guard](src/graphon/runtime/execution.py) | [Serialization](tests/engine/test_runtime_state_serialization.py) |
 | Snapshot versions and migrations | [Snapshot loading](src/graphon/runtime/runtime_state/snapshot.py) | [Runtime snapshots](tests/runtime/test_runtime_state.py), [version isolation](tests/test_snapshot_version_isolation.py) |
 
+[Concrete segment models](src/graphon/variables/segments.py) own fixed `value_type` tags and value validation; serialized values select their model by that tag.
+[Factories](src/graphon/variables/factory.py) infer types for raw values or validate explicit declarations, preserving declared empty arrays and `array[any]` values.
+See [factory tests](tests/variables/test_factory.py) for declared types and serialized tag selection.
+`array[any]` restores serialized [marked file values](src/graphon/file/constants.py) recursively inside arrays and objects using [File validation](src/graphon/file/models.py), preserving file values in container task and variable pool snapshots.
+Unmarked objects retain their data, and invalid marked files fail validation; see [container tests](tests/nodes/test_container_effects.py) and [workflow tests](tests/workflows/test_full_engine_events.py).
+[VariableBase.copy_with_value](src/graphon/variables/variables.py) validates replacements for assignment nodes, retaining identity and metadata and promoting integer variables when the new value is a float.
+The [v1 assignment tests](tests/nodes/variable_assigner/test_v1_node.py) and [v2 assignment tests](tests/nodes/variable_assigner/test_v2_node.py) cover invalid updates, fractional results, and clearing arrays while preserving their declared type, including `array[file]`.
+
+[Container values](src/graphon/nodes/container_effects.py) preserve declared segment types through construction, snapshots, and completion, including secret tags.
+Conversion removes source variable identity so the parent pool binds completed outputs to the container's selectors.
+Pool conversion also assigns a new identity when storing an existing variable under a different selector; see [destination binding tests](tests/runtime/test_variable_pool.py).
+Stored outputs remain typed; container node-result inputs and Loop/Iteration presentation events expose plain values.
+The [runtime JSON converter](src/graphon/workflow_type_encoder.py) masks typed secrets using their log representation; hosts must apply redaction separately to plain values in logs and presentation.
+The [container secret tests](tests/nodes/test_container_effects.py) cover this boundary; [Loop workflow tests](tests/workflows/test_full_engine_events.py) also cover v2 `SET`, `OVER_WRITE`, and `CLEAR` preserving the secret tag and masking.
+
+[Iteration](src/graphon/nodes/iteration/iteration_node.py) accepts array sequences, including tuples.
+When its [handler](src/graphon/engine/container_handler/builtin/iteration.py) flattens outputs, matching declared array types survive in both serial and parallel runs; mixed array types become `array[any]`.
+Empty iterations preserve the input array type and bind a separate output variable; see [workflow tests](tests/workflows/test_full_engine_events.py).
+
 For host rendering, bind the engine's adapter with
 [use_workflow_file_runtime](src/graphon/file/runtime.py); use
 [EngineEventFilterContext.from_engine](src/graphon/engine/filter/protocol.py) to
 wire response filters. Supply host adapters again when rebuilding from snapshots.
+Runtime JSON conversion renders tuple file arrays as JSON lists within the same adapter scope; see [host file conversion tests](tests/variables/test_workflow_type_encoder.py).
 
 Persist after fully consuming the run iterator or from a quiescent `on_graph_end`
 hook. Hosts must also coordinate direct mutation of exposed runtime objects: the

docs/development.md

@@ -1,5 +1,5 @@
 <!-- knowledge
-last_checked: "2026-10-29T01:01:57Z"
+last_checked: "2026-10-30T16:06:28Z"
 -->
 # Development navigation
 
@@ -16,15 +16,19 @@ Choose boundaries around state ownership and callers. Extract a responsibility
 when it has independent ownership or repeated churn; file length alone does not
 justify splitting a module.
 
+For runtime values, let concrete types own their tags, validation, and value replacement.
+Preserve typed values across state transfers and convert to plain values at presentation boundaries, where redaction is a separate responsibility.
+Follow the [variable and container contracts](../ARCHITECTURE.md#state-and-execution-invariants) when changing those boundaries.
+
 | Task | Start in source | Relevant checks |
 | --- | --- | --- |
 | Change package dependencies or queue ownership | [Import boundaries](../ARCHITECTURE.md#import-boundaries), [runtime queues](../src/graphon/runtime/ready_queue/) | `just imports`, [runtime isolation](../tests/runtime/test_runtime_imports.py), [public contract imports](../tests/test_protocols_exports.py) |
 | Parse or validate a graph | [graph](../src/graphon/graph/), [graph config](../src/graphon/entities/graph_config.py) | [graph tests](../tests/graph/), especially validation and scoping |
 | Import Dify DSL or wire node dependencies | [importer](../src/graphon/dsl/importer.py), [node factory](../src/graphon/dsl/node_factory.py) | [DSL tests](../tests/dsl/), especially importer, node factory, and app bootstrap |
 | Change a built-in node | [nodes](../src/graphon/nodes/), [base node](../src/graphon/nodes/base/node.py) | Matching [node tests](../tests/nodes/); [workflow events](../tests/workflows/test_full_engine_events.py) for graph-visible changes |
-| Change dispatch, scheduling, or containers | [engine](../src/graphon/engine/), [container effects](../src/graphon/nodes/container_effects.py) | [engine tests](../tests/engine/), [container dispatch](../tests/nodes/test_container_dispatch.py), [workflow events](../tests/workflows/test_full_engine_events.py) |
+| Change dispatch, scheduling, or containers | [engine](../src/graphon/engine/), [container effects](../src/graphon/nodes/container_effects.py) | [engine tests](../tests/engine/), [container values](../tests/nodes/test_container_effects.py), [container dispatch](../tests/nodes/test_container_dispatch.py), [workflow events](../tests/workflows/test_full_engine_events.py) |
 | Change event payloads or response streaming | [raw frame observations](../ARCHITECTURE.md#raw-frame-observations), [node events](../src/graphon/node_events/), [engine events](../src/graphon/engine_events/), [filters](../src/graphon/engine/filter/) | [frame observation tests](../tests/engine/test_frame_observations.py), [frame failure tests](../tests/engine/test_frame_failures.py), [node event tests](../tests/node_events/), [engine event tests](../tests/engine_events/), [response filter tests](../tests/engine/test_response_stream_filter.py) |
-| Change variables or secrets | [variable pool](../src/graphon/runtime/variable_pool.py), [variables](../src/graphon/variables/) | [variable pool tests](../tests/runtime/test_variable_pool.py), [Iteration variable ownership](../tests/engine/test_iteration_variable_scopes.py), [variable tests](../tests/variables/), [HTTP secret masking](../tests/http/test_executor_secret_masking.py) |
+| Change variables or secrets | [variable pool](../src/graphon/runtime/variable_pool.py), [segment models](../src/graphon/variables/segments.py), [variable models](../src/graphon/variables/variables.py), [conversion](../src/graphon/variables/factory.py), [variable assignment](../src/graphon/nodes/variable_assigner/) | [variable pool tests](../tests/runtime/test_variable_pool.py), [Iteration variable ownership](../tests/engine/test_iteration_variable_scopes.py), [variable tests](../tests/variables/), [assignment tests](../tests/nodes/variable_assigner/), [HTTP secret masking](../tests/http/test_executor_secret_masking.py) |
 | Change persistence or resume | [runtime state](../src/graphon/runtime/runtime_state/), [container state](../src/graphon/runtime/container_state.py) | [runtime tests](../tests/runtime/), [engine serialization](../tests/engine/test_runtime_state_serialization.py), [frame failure restoration](../tests/engine/test_frame_failures.py), [version isolation](../tests/test_snapshot_version_isolation.py) |
 | Change model invocation | [graph-facing LLM protocol](../src/graphon/nodes/llm/runtime_protocols.py), [Slim adapter](../src/graphon/dsl/slim/llm.py), [model runtime](../src/graphon/model_runtime/) | [LLM node tests](../tests/nodes/llm/), [Slim tests](../tests/dsl/test_slim_llm.py), [model dispatch](../tests/model_runtime/test_model_dispatch.py) |
 | Change file or HTTP integration | [file](../src/graphon/file/), [engine adapter binding](../src/graphon/engine/engine.py), [HTTP](../src/graphon/http/), [node ports](../src/graphon/nodes/protocols.py) | [file tests](../tests/file/), [execution file isolation](../tests/engine/test_file_runtime.py), [HTTP tests](../tests/http/), [HTTP node tests](../tests/nodes/http_request/), [extractor tests](../tests/nodes/document_extractor/) |

src/graphon/engine/container_handler/builtin/iteration.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from datetime import UTC, datetime
-from typing import cast, final
+from typing import final
 
 from graphon.engine_events.base import NodeEvent
 from graphon.engine_events.node import NodeRunFailedEvent, NodeRunVariableUpdatedEvent
@@ -29,7 +29,7 @@
 from graphon.runtime.execution import ROOT_FRAME_ID
 from graphon.runtime.ready_queue import ResumeTask
 from graphon.runtime.runtime_state import RuntimeState
-from graphon.variables.segments import NoneSegment, SerializableSegment
+from graphon.variables.segments import ArrayAnySegment, ArraySegment, NoneSegment
 from graphon.variables.variables import Variable
 
 from ...frame import ExecutionFrame, FrameRegistry
@@ -206,13 +206,12 @@ def _complete_ready_iteration_frame(
         result = frame.state.variable_pool.get(
             run_state.output_selector,
         )
-        output = NoneSegment() if result is None else cast(SerializableSegment, result)
         run_state = self._complete_iteration_step(
             frame=frame,
             frame_state=frame_state,
             parent_frame=parent_frame,
             run_state=run_state,
-            output=output,
+            output=build_container_value(result),
             store_output=True,
         )
         self._continue_or_complete_iteration(
@@ -266,7 +265,7 @@ def _complete_iteration_step(
         frame_state: IterationFrameState,
         parent_frame: ExecutionFrame,
         run_state: IterationRunState,
-        output: SerializableSegment,
+        output: ContainerValue,
         store_output: bool,
     ) -> IterationRunState:
         duration_map = dict(run_state.duration_map)
@@ -455,14 +454,7 @@ def _build_iteration_result(
         status: WorkflowNodeExecutionStatus,
         error: str = "",
     ) -> ContainerExecutionResult:
-        outputs: dict[str, ContainerValue] = {
-            "output": build_container_value(
-                self._flatten_outputs_if_needed(
-                    self._ordered_iteration_outputs(run_state),
-                    flatten_output=run_state.flatten_output,
-                ),
-            ),
-        }
+        outputs = {"output": self._collect_iteration_outputs(run_state)}
         metadata = self._iteration_metadata(run_state)
         return ContainerExecutionResult(
             metadata=self._event_metadata(metadata),
@@ -492,36 +484,29 @@ def _iteration_metadata(
             ),
         }
 
-    def _flatten_outputs_if_needed(
-        self,
-        outputs: list[object],
-        *,
-        flatten_output: bool,
-    ) -> list[object]:
-        if not flatten_output:
-            return outputs
-        non_empty_outputs = [output for output in outputs if output is not None]
-        if not non_empty_outputs:
-            return outputs
-        if not all(isinstance(output, list) for output in non_empty_outputs):
-            return outputs
-
-        flattened: list[object] = []
-        for output in outputs:
-            if isinstance(output, list):
-                flattened.extend(output)
-            elif output is not None:
-                flattened.append(output)
-        return flattened
-
-    def _ordered_iteration_outputs(
+    def _collect_iteration_outputs(
         self,
         run_state: IterationRunState,
-    ) -> list[object]:
-        return [
-            run_state.outputs[key].to_object()
-            for key in sorted(run_state.outputs, key=int)
-        ]
+    ) -> ContainerValue:
+        outputs = [run_state.outputs[key] for key in sorted(run_state.outputs, key=int)]
+        arrays = [output for output in outputs if isinstance(output, ArraySegment)]
+        if (
+            run_state.flatten_output
+            and arrays
+            and all(
+                isinstance(output, (ArraySegment, NoneSegment)) for output in outputs
+            )
+        ):
+            flattened = [item for output in arrays for item in output.value]
+            if len({output.value_type for output in arrays}) == 1:
+                return build_container_value(arrays[0]).model_copy(
+                    update={"value": flattened}
+                )
+            return ArrayAnySegment(value=flattened)
+        values = [output.to_object() for output in outputs]
+        return (
+            ArrayAnySegment(value=values) if arrays else build_container_value(values)
+        )
 
     def _enqueue_container_result(
         self,

src/graphon/engine/event/processor.py

@@ -46,6 +46,7 @@
 )
 from graphon.runtime.execution import ROOT_FRAME_ID, GraphExecution
 from graphon.runtime.ready_queue import ResumeTask, StartTask
+from graphon.variables.segments import Segment
 
 from ..container_handler import ContainerHandler
 from ..frame import ExecutionFrame, FrameRegistry
@@ -203,11 +204,15 @@ def _(
         frame: ExecutionFrame,
     ) -> None:
         if isinstance(event, NodeRunLoopSucceededEvent):
-            event.outputs = self._current_loop_outputs(
+            outputs = self._current_loop_outputs(
                 frame=frame,
                 node_id=event.node_id,
                 outputs=event.outputs,
             )
+            event.outputs = {
+                key: value.to_object() if isinstance(value, Segment) else value
+                for key, value in outputs.items()
+            }
         self._collect(frame=frame, event=event)
 
     @_dispatch.register
@@ -403,7 +408,7 @@ def _current_loop_outputs(
         for key in outputs:
             value = frame.state.variable_pool.get((node_id, key))
             if value is not None:
-                current_outputs[key] = value.to_object()
+                current_outputs[key] = value
         return current_outputs
 
     def _store_node_outputs(

src/graphon/nodes/container_effects.py

@@ -1,33 +1,32 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from typing import Annotated, Literal, cast
+from typing import Annotated, Literal
 
-from pydantic import BaseModel, BeforeValidator, ConfigDict, Field
+from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, TypeAdapter
 
 from graphon.enums import ErrorHandleMode
 from graphon.node_events.base import NodeRunResult
 from graphon.variables.factory import build_segment
 from graphon.variables.segments import Segment, SerializableSegment
 
 
-def _to_serializable_segment(value: object) -> object:
-    if isinstance(value, Segment):
-        return build_segment(value.to_object())
+def _parse_container_value(value: object) -> Segment | Mapping[str, object]:
     if isinstance(value, Mapping) and "value_type" in value:
         return value
     return build_segment(value)
 
 
 type ContainerValue = Annotated[
     SerializableSegment,
-    BeforeValidator(_to_serializable_segment),
+    BeforeValidator(_parse_container_value),
 ]
 
+_CONTAINER_VALUE_ADAPTER = TypeAdapter(ContainerValue)
+
 
 def build_container_value(value: object) -> ContainerValue:
-    raw_value = value.to_object() if isinstance(value, Segment) else value
-    return cast(ContainerValue, build_segment(raw_value))
+    return _CONTAINER_VALUE_ADAPTER.validate_python(build_segment(value))
 
 
 class ContainerNodeRunResult(NodeRunResult):
@@ -38,7 +37,7 @@ def to_node_run_result(self) -> NodeRunResult:
         return NodeRunResult(
             status=self.status,
             inputs={key: value.to_object() for key, value in self.inputs.items()},
-            outputs={key: value.to_object() for key, value in self.outputs.items()},
+            outputs=self.outputs,
             metadata=self.metadata,
             llm_usage=self.llm_usage,
             error=self.error,