route those APIs

2026-05-25 00:23:55 +07:00
parent 2e252f6252
commit 7a3555e7bc
8 changed files with 155 additions and 18 deletions
@@ -55,8 +55,11 @@ implementation state.
  quiescent interrupt behavior. Remaining work is polish and future reuse of
  its barrier/lineage machinery by native subgraphs and fork/gather. Current
  lineage progress includes ordered `StateWrite` records, `LineageStateView`,
-  foreach item `lineage_id`s, and nested foreach lineage identity. Full
-  `RuntimeScope` / `LineageState` storage is still future work.
+  foreach item `lineage_id`s, nested foreach lineage identity, root
+  `RuntimeScope` / `LineageState` storage, scope-aware reads, and non-root write
+  buffering. Current direct commits are still root-frame-only via an explicit
+  helper; native subgraph completion should replace that shortcut with an
+  explicit scope/lineage commit target.
 - **Persistent run history**: add a run store before adding stable `run_id`,
  `inspect_run`, or `read_run_trace(run_id, range)` APIs. Current traces are
  returned directly from immediate run responses.
@@ -27,11 +27,19 @@ the compatibility subset needed before native subgraphs:
  `parent_lineage_id`.
 - Concurrent foreach child frames receive deterministic, opaque lineage ids,
  including nested foreach frames.
+- `RunState` has root scope/lineage storage, scope-aware state views, and
+  generic non-root node writes buffer into `RunState.lineages`.
+
+Direct commits currently go through `is_root_lineage_frame(frame)`, which is the
+migration shortcut for root scope/root lineage. The eventual better shape is an
+explicit scope/lineage commit target, feasible once native subgraph completion
+can declare whether child writes commit to child scope, parent lineage, or only
+through boundary output bindings.

 Remaining work should avoid jumping straight to native subgraphs. The next
-small slice is to centralize "which writes are visible to this frame" behind a
-helper, then later decide whether to add full `RunState.scopes` /
-`RunState.lineages`.
+small slice is to migrate foreach pending patch storage from
+`ForeachBarrierState` into `RunState.lineages`, or defer that and start native
+subgraph scaffolding using the current scope/lineage primitives.

 ---

@@ -530,8 +538,7 @@ For root scope/root lineage it may return `run.state` directly as an optimizatio

 In `_finalize_node_execution(...)`:

- if `frame.scope_id == "root"` and `frame.lineage_id == "root"`, commit patch
-  to `run.state`
+- if `is_root_lineage_frame(frame)`, commit patch to `run.state`
 - otherwise append `patch.writes` to the frame lineage and return empty committed
  `state_changes`

@@ -25,12 +25,19 @@ The first compatibility slices are implemented:
 - Foreach pending result metadata persists write records and `lineage_id`.
 - Frames and runtime context carry `scope_id`, `lineage_id`, and
  `parent_lineage_id`.
+- Generic non-root frame writes are buffered into `RunState.lineages`.

 The full `RuntimeScope` / `LineageState` store is not implemented yet.
 Currently, foreach still owns pending write storage through
 `ForeachBarrierState`; the lineage ids are identity and diagnostics, not yet the
 primary storage key.

+Direct node commits currently use the explicit root-frame helper
+`is_root_lineage_frame(frame)`. That helper still means "root scope plus root
+lineage" during migration. The better long-term shape becomes feasible when
+native subgraph completion exists: direct commits should be decided by an
+explicit scope/lineage commit target, not by root ids.
+
 ## Problem

 `RunState` currently owns too many meanings:
@@ -7,6 +7,10 @@ from typing import Any
 from wf_core.models.reducers import ReducerRef
 from wf_core.paths import StatePath

+ROOT_SCOPE_ID = "root"
+ROOT_LINEAGE_ID = "root"
+ROOT_FRAME_ID = "root"
+

 class RunStatus(StrEnum):
    PENDING = "pending"
@@ -70,8 +74,8 @@ class ExecutionFrame:
    node_id: str
    status: FrameStatus = FrameStatus.PENDING
    parent_frame_id: str | None = None
-    scope_id: str = "root"
-    lineage_id: str = "root"
+    scope_id: str = ROOT_SCOPE_ID
+    lineage_id: str = ROOT_LINEAGE_ID
    parent_lineage_id: str | None = None
    prior_outcome: str | None = None
    activated_incoming_edge: str | None = None
@@ -82,9 +86,9 @@ class ExecutionFrame:
@dataclass(slots=True)
 class RuntimeContext:
    current_node_id: str
-    frame_id: str = "root"
-    scope_id: str = "root"
-    lineage_id: str = "root"
+    frame_id: str = ROOT_FRAME_ID
+    scope_id: str = ROOT_SCOPE_ID
+    lineage_id: str = ROOT_LINEAGE_ID
    parent_lineage_id: str | None = None
    retry_count: int = 0
    prior_outcome: str | None = None
@@ -6,6 +6,7 @@ from dataclasses import dataclass
 from typing import Any

 from wf_core.run_state import ExecutionFrame, LineageState, RunState, StateWrite
+from wf_core.run_state import ROOT_LINEAGE_ID, ROOT_SCOPE_ID
 from wf_core.runtime.foreach_state import ForeachBarrierState, item_frame_owner
 from wf_core.runtime.ops.state import StatePatch
 from wf_core.runtime.ops.state import safe_set_nested_value
@@ -69,6 +70,16 @@ def lineage_writes_for_frame(
    return pending.patch.writes


+def is_root_lineage_frame(frame: ExecutionFrame) -> bool:
+    """Return whether a frame currently commits directly to root run state.
+
+    This is a migration shortcut, not the final commit policy. Once native
+    subgraphs can complete, direct commits should be decided by an explicit
+    scope/lineage commit target rather than only by root ids.
+    """
+    return frame.scope_id == ROOT_SCOPE_ID and frame.lineage_id == ROOT_LINEAGE_ID
+
+
 def scope_state_for_frame(run: RunState, frame: ExecutionFrame) -> dict[str, Any]:
    """Return the committed state root for the frame's runtime scope."""
    scope = run.scopes.get(frame.scope_id)
@@ -18,6 +18,7 @@ from wf_core.run_state import (
    StepExecutionResult,
 )
 from wf_core.runtime.foreach_state import ForeachBarrierState, item_frame_owner
+from wf_core.runtime.lineage import append_lineage_writes, is_root_lineage_frame
 from wf_core.runtime.ops.frames import frame_context_values
 from wf_core.runtime.ops.merges import ReducerDefinition
 from wf_core.runtime.ops.overlays import state_view_for_frame
@@ -120,7 +121,16 @@ def _finalize_node_execution(
    )
    owner = item_frame_owner(frame)
    if owner is None:
-        state_changes = commit_state_patch(run.state, patch)
+        if is_root_lineage_frame(frame):
+            state_changes = commit_state_patch(run.state, patch)
+        else:
+            append_lineage_writes(
+                run,
+                scope_id=frame.scope_id,
+                lineage_id=frame.lineage_id,
+                writes=patch.writes,
+            )
+            state_changes = {}
    else:
        parent_frame_id, foreach_node_id, item_index = owner
        parent_frame = run.frames[parent_frame_id]
@@ -8,6 +8,9 @@ from wf_core.run_state import (
    ExecutionFrame,
    FrameStatus,
    LineageState,
+    ROOT_FRAME_ID,
+    ROOT_LINEAGE_ID,
+    ROOT_SCOPE_ID,
    RunState,
    RunStatus,
    RuntimeScope,
@@ -27,20 +30,25 @@ def create_run_state(workflow: Workflow, workflow_input: dict[str, object]) -> R
        workflow_input=dict(workflow_input),
        state=state,
        scopes={
-            "root": RuntimeScope(
-                id="root",
+            ROOT_SCOPE_ID: RuntimeScope(
+                id=ROOT_SCOPE_ID,
                workflow_name=workflow.name,
                committed_state=state,
            )
        },
-        lineages={"root": LineageState(id="root", scope_id="root")},
-        current_frame_id="root",
+        lineages={
+            ROOT_LINEAGE_ID: LineageState(
+                id=ROOT_LINEAGE_ID,
+                scope_id=ROOT_SCOPE_ID,
+            )
+        },
+        current_frame_id=ROOT_FRAME_ID,
        current_node_id=workflow.start,
    )
    add_frame(
        run,
        ExecutionFrame(
-            id="root",
+            id=ROOT_FRAME_ID,
            kind="workflow",
            node_id=workflow.start,
            status=FrameStatus.PENDING,
@@ -21,6 +21,7 @@ from wf_core.runtime.lineage import (
    lineage_state_view,
    scope_state_for_frame,
 )
+from wf_core.runtime.ops.nodes import execute_node_use
 from wf_core.runtime.ops.overlays import state_view_for_frame
 from wf_core.runtime.ops.runs import create_run_state

@@ -157,6 +158,47 @@ def test_state_view_for_frame_overlays_writes_onto_frame_scope_state() -> None:
    assert run.state["value"] == "root"


+def test_non_root_frame_node_writes_are_buffered_in_lineage() -> None:
+    workflow = _write_value_workflow()
+    run = create_run_state(workflow, {"value": "root"})
+    run.scopes["child"] = RuntimeScope(
+        id="child",
+        workflow_name="child_workflow",
+        committed_state={"value": "child"},
+    )
+    run.lineages["child/root"] = LineageState(id="child/root", scope_id="child")
+    add_lineage(
+        run,
+        scope_id="child",
+        lineage_id="child/branch",
+        parent_id="child/root",
+    )
+    frame = ExecutionFrame(
+        id="child-frame",
+        kind="workflow",
+        node_id="write_value",
+        scope_id="child",
+        lineage_id="child/branch",
+        parent_lineage_id="child/root",
+    )
+    run.frames[frame.id] = frame
+    run.current_frame_id = frame.id
+
+    result = execute_node_use(
+        workflow,
+        run,
+        workflow.nodes[0],  # type: ignore[arg-type]
+        workflow.node_defs[0],
+        {"write_value": lambda payload, _ctx: {"value": f"{payload['value']}-next"}},
+    )
+
+    assert result.state_changes == {}
+    assert run.scopes["child"].committed_state["value"] == "child"
+    assert run.state["value"] == "root"
+    assert run.lineages["child/branch"].writes[0].incoming_value == "child-next"
+    assert state_view_for_frame(run, frame)["value"] == "child-next"
+
+
 def _minimal_workflow() -> Workflow:
    return Workflow(
        name="lineage_root",
@@ -182,3 +224,48 @@ def _minimal_workflow() -> Workflow:
        ],
        edges=[Edge.model_validate({"from": "finish", "outcome": "ok", "to": END})],
    )
+
+
+def _write_value_workflow() -> Workflow:
+    return Workflow(
+        name="lineage_write",
+        input_schema=SchemaRef(
+            type="object",
+            properties={"value": {"type": "string"}},
+        ),
+        state_schema=StateSchema.from_field_map(
+            {"value": StateField(type="string", default="default")}
+        ),
+        output_schema=SchemaRef(type="object", properties={}),
+        node_defs=[
+            NodeDef(
+                name="write_value",
+                input_schema=SchemaRef(
+                    type="object",
+                    properties={"value": {"type": "string"}},
+                    required=["value"],
+                ),
+                output_schema=SchemaRef(
+                    type="object",
+                    properties={"value": {"type": "string"}},
+                    required=["value"],
+                ),
+                outcomes=["ok"],
+            )
+        ],
+        start="write_value",
+        nodes=[
+            NodeUse.model_validate(
+                {
+                    "id": "write_value",
+                    "type": "node",
+                    "node": "write_value",
+                    "input": [{"target": "value", "path": "state.value"}],
+                    "output": [{"source": "value", "target": "state.value"}],
+                }
+            )
+        ],
+        edges=[
+            Edge.model_validate({"from": "write_value", "outcome": "ok", "to": END})
+        ],
+    )