scope / lineage in frame

more structs to prepare for subgraph

src/wf_core/run_state.py

@@ -40,6 +40,29 @@ class StateWrite:
     reducer: ReducerRef
 
 
+@dataclass(slots=True)
+class RuntimeScope:
+    """Committed workflow state root for one workflow activation.
+
+    During migration, the root scope intentionally shares `RunState.state` so
+    existing runtime code can keep using the compatibility state dict.
+    """
+
+    id: str
+    workflow_name: str
+    committed_state: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass(slots=True)
+class LineageState:
+    """Ordered pending writes owned by one lineage inside a runtime scope."""
+
+    id: str
+    scope_id: str
+    parent_id: str | None = None
+    writes: list[StateWrite] = field(default_factory=list)
+
+
 @dataclass(slots=True)
 class ExecutionFrame:
     id: str
@@ -108,6 +131,8 @@ class RunState:
     output: dict[str, Any] = field(default_factory=dict)
     trace: list[TraceEntry] = field(default_factory=list)
     frames: dict[str, ExecutionFrame] = field(default_factory=dict)
+    scopes: dict[str, RuntimeScope] = field(default_factory=dict)
+    lineages: dict[str, LineageState] = field(default_factory=dict)
     ready_frame_ids: list[str] = field(default_factory=list)
     current_frame_id: str | None = None
     current_node_id: str | None = None

src/wf_core/runtime/ops/runs.py

@@ -4,7 +4,14 @@ from copy import deepcopy
 
 from wf_core.models.workflow import Workflow
 from wf_core.paths import set_nested_value
-from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, RunStatus
+from wf_core.run_state import (
+    ExecutionFrame,
+    FrameStatus,
+    LineageState,
+    RunState,
+    RunStatus,
+    RuntimeScope,
+)
 from wf_core.runtime.scheduler import add_frame
 
 
@@ -19,6 +26,14 @@ def create_run_state(workflow: Workflow, workflow_input: dict[str, object]) -> R
         status=RunStatus.PENDING,
         workflow_input=dict(workflow_input),
         state=state,
+        scopes={
+            "root": RuntimeScope(
+                id="root",
+                workflow_name=workflow.name,
+                committed_state=state,
+            )
+        },
+        lineages={"root": LineageState(id="root", scope_id="root")},
         current_frame_id="root",
         current_node_id=workflow.start,
     )

tests/core/test_lineage_state.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from wf_core import (
+    END,
+    Edge,
+    NodeDef,
+    NodeUse,
+    SchemaRef,
+    StateField,
+    StateSchema,
+    Workflow,
+)
+from wf_core.runtime.ops.runs import create_run_state
+
+
+def test_create_run_state_initializes_root_scope_and_lineage() -> None:
+    workflow = _minimal_workflow()
+
+    run = create_run_state(workflow, {"value": "seed"})
+
+    assert run.scopes["root"].id == "root"
+    assert run.scopes["root"].workflow_name == workflow.name
+    assert run.scopes["root"].committed_state is run.state
+    assert run.scopes["root"].committed_state["value"] == "seed"
+    assert run.lineages["root"].id == "root"
+    assert run.lineages["root"].scope_id == "root"
+    assert run.lineages["root"].parent_id is None
+    assert run.lineages["root"].writes == []
+    assert run.frames["root"].scope_id == "root"
+    assert run.frames["root"].lineage_id == "root"
+    assert run.frames["root"].parent_lineage_id is None
+
+
+def _minimal_workflow() -> Workflow:
+    return Workflow(
+        name="lineage_root",
+        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="finish",
+                input_schema=SchemaRef(type="object", properties={}),
+                output_schema=SchemaRef(type="object", properties={}),
+                outcomes=["ok"],
+            )
+        ],
+        start="finish",
+        nodes=[
+            NodeUse.model_validate({"id": "finish", "type": "node", "node": "finish"})
+        ],
+        edges=[Edge.model_validate({"from": "finish", "outcome": "ok", "to": END})],
+    )