From ce085b14f8401fb700b5728a75597abb1b6180ef Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 14 Aug 2026 15:00:28 +0700 Subject: [PATCH] feat: inspect node runtime context --- .../task-2-report.md | 72 +++++ src/wf_api/authoring_contracts.py | 84 +++++- src/wf_core/analysis/__init__.py | 13 + src/wf_core/analysis/context_scopes.py | 254 ++++++++++++++++++ src/wf_core/context_contracts.py | 80 ++++++ src/wf_core/runtime/ops/frames.py | 23 +- tests/core/test_context_scopes.py | 226 ++++++++++++++++ tests/core/test_scheduler.py | 20 ++ tests/wf_api/test_authoring_contracts.py | 46 ++++ 9 files changed, 810 insertions(+), 8 deletions(-) create mode 100644 .superpowers/sdd/2026-08-14-workflow-console-contract-graph/task-2-report.md create mode 100644 src/wf_core/analysis/__init__.py create mode 100644 src/wf_core/analysis/context_scopes.py create mode 100644 src/wf_core/context_contracts.py create mode 100644 tests/core/test_context_scopes.py diff --git a/.superpowers/sdd/2026-08-14-workflow-console-contract-graph/task-2-report.md b/.superpowers/sdd/2026-08-14-workflow-console-contract-graph/task-2-report.md new file mode 100644 index 00000000..b57a64ff --- /dev/null +++ b/.superpowers/sdd/2026-08-14-workflow-console-contract-graph/task-2-report.md @@ -0,0 +1,72 @@ +# Task 2 Report: Analyze Node-Scoped Runtime Context + +## Status + +Implemented Task 2 of the workflow contract graph backend slice. + +## Changes + +- Added `wf_core.context_contracts` with the exact standard runtime context + field schemas and shared key constants. +- Added `foreach_context_fields`, including `loop_item`, `loop_index`, and a + configured alias with duplicate loop-key aliases removed. +- Updated `frame_context_values` to use the shared context key registry while + preserving its existing runtime values. +- Added `context_fields_by_node`, an abstract traversal that memoizes + `(node_id, active_foreach_id)` and distinguishes available from conditional + fields across reachable frame scopes. +- Added bounded graph warnings for missing route targets, missing loop routes, + invalid workflow starts, and invalid edge sources. +- Derived foreach item schemas from declared input/state array sources, falling + back to `{}` when the source is not declared as an array with an item schema. +- Added Task 1 authoring projection helpers for canonical `context.` paths. + Runtime context is offered only for `step_input`; workflow output projections + do not advertise `context.*`. +- Added coverage for ordinary frames, serial/concurrent foreach, conditional + reachability, nested scope replacement/restoration, malformed routes, cyclic + graphs, alias deduplication, and authoring projection behavior. + +## Test-First Evidence + +The required RED command was run after adding tests and before production +changes: + +```text +uv run pytest tests/core/test_context_scopes.py tests/core/test_scheduler.py tests/wf_api/test_authoring_contracts.py -q +``` + +It failed during collection for the expected missing production seams: + +```text +ModuleNotFoundError: No module named 'wf_core.analysis' +ImportError: cannot import name 'context_path_options' from 'wf_api.authoring_contracts' +14 passed, 2 errors +``` + +## Verification + +```text +uv run pytest tests/core/test_context_scopes.py tests/core/test_scheduler.py tests/wf_api/test_authoring_contracts.py -q +29 passed + +uv run pytest tests/core -q +296 passed + +uv run pytest tests/wf_api/test_authoring_contracts.py -q +7 passed + +uv run ruff check src/wf_core/context_contracts.py src/wf_core/analysis src/wf_core/runtime/ops/frames.py src/wf_api/authoring_contracts.py tests/core/test_context_scopes.py tests/core/test_scheduler.py tests/wf_api/test_authoring_contracts.py +All checks passed! + +uv run basedpyright --level error src/wf_core/context_contracts.py src/wf_core/analysis src/wf_core/runtime/ops/frames.py +0 errors, 0 warnings, 0 notes +``` + +## Concerns + +- Foreach item schema traversal intentionally handles declared direct object + properties and array `items`; complex external or deeply composed schema + references fall back conservatively to `{}` rather than inventing a type. +- The authoring projector accepts an optional workflow for automatic context + projection, while existing callers can continue supplying Task 1 payloads + explicitly. diff --git a/src/wf_api/authoring_contracts.py b/src/wf_api/authoring_contracts.py index 47fd9b29..f92bac01 100644 --- a/src/wf_api/authoring_contracts.py +++ b/src/wf_api/authoring_contracts.py @@ -4,6 +4,13 @@ from collections.abc import Mapping, Sequence from copy import deepcopy from typing import Any +from wf_core.analysis.context_scopes import ( + ContextFieldAvailability, + context_analysis_warnings, + context_fields_by_node, +) +from wf_core.models.workflow import Workflow + from .models.authoring_contracts import ( AuthoringContractInventoryPayload, AuthoringPathOptionPayload, @@ -73,6 +80,7 @@ def project_authoring_contract_inventory( entry_steps: Sequence[AuthoringStepContractPayload] = (), workflow_outcomes: Sequence[str] = (), warnings: Sequence[str] = (), + workflow: Workflow | None = None, ) -> AuthoringContractInventoryPayload: """Compose an inventory from caller-provided schemas and graph facts. @@ -80,6 +88,10 @@ def project_authoring_contract_inventory( service layer supplies the selected-step and runtime-context projections; this function only derives schema choices and copies those projections. """ + if workflow is not None and selected_step_id is not None and not context_entries: + context_entries = context_path_options_for_node(workflow, selected_step_id) + warnings = [*warnings, *context_analysis_warnings(workflow)] + input_sources = schema_path_options( input_schema, root="input", @@ -107,7 +119,7 @@ def project_authoring_contract_inventory( "selected_step_id": selected_step_id, "readable_sources": [ *input_sources, - *deepcopy(list(context_entries)), + *_context_entries_for_inventory(context_entries), *state_sources, ], "step_input_targets": deepcopy(list(step_input_targets)), @@ -120,6 +132,76 @@ def project_authoring_contract_inventory( } +def context_path_options( + fields: Sequence[ContextFieldAvailability | Mapping[str, Any]], +) -> list[AuthoringPathOptionPayload]: + """Project analyzed runtime context fields into Task 1 path payloads.""" + options: list[AuthoringPathOptionPayload] = [] + for field in fields: + if isinstance(field, ContextFieldAvailability): + name = field.name + schema = field.schema + description = field.description + availability = field.availability + reason = field.reason + else: + raw_name = field.get("name") + if not isinstance(raw_name, str) or not raw_name: + continue + name = raw_name + raw_schema = field.get("schema") + schema = raw_schema if isinstance(raw_schema, Mapping) else {} + raw_description = field.get("description") + description = raw_description if isinstance(raw_description, str) else name + raw_availability = field.get("availability") + availability = ( + raw_availability + if raw_availability in {"available", "conditional"} + else "available" + ) + raw_reason = field.get("reason") + reason = raw_reason if isinstance(raw_reason, str) else None + + option: AuthoringPathOptionPayload = { + "path": f"context.{name}", + "label": name.replace("_", " ").replace("-", " ").title(), + "origin": "runtime_context", + "schema": deepcopy(dict(schema)), + "required": False, + "availability": availability, + "uses": ["step_input"], + } + if description: + option["description"] = description + if reason is not None: + option["reason"] = reason + options.append(option) + return options + + +def context_path_options_for_node( + workflow: Workflow, + node_id: str, +) -> list[AuthoringPathOptionPayload]: + """Project the runtime context available at one workflow node.""" + return context_path_options(context_fields_by_node(workflow).get(node_id, ())) + + +def _context_entries_for_inventory( + entries: Sequence[AuthoringPathOptionPayload], +) -> list[AuthoringPathOptionPayload]: + """Keep runtime context readable only where execution has frame context.""" + result: list[AuthoringPathOptionPayload] = [] + for entry in entries: + copied = deepcopy(entry) + if copied["path"].startswith("context."): + copied["uses"] = [use for use in copied["uses"] if use == "step_input"] + if not copied["uses"]: + continue + result.append(copied) + return result + + def _append_schema_options( schema: JsonObject, *, diff --git a/src/wf_core/analysis/__init__.py b/src/wf_core/analysis/__init__.py new file mode 100644 index 00000000..44fab703 --- /dev/null +++ b/src/wf_core/analysis/__init__.py @@ -0,0 +1,13 @@ +"""Static analyses over workflow graph contracts.""" + +from .context_scopes import ( + ContextFieldAvailability, + context_analysis_warnings, + context_fields_by_node, +) + +__all__ = [ + "ContextFieldAvailability", + "context_analysis_warnings", + "context_fields_by_node", +] diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py new file mode 100644 index 00000000..2fa69a6f --- /dev/null +++ b/src/wf_core/analysis/context_scopes.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +from collections import deque +from collections.abc import Mapping +from copy import deepcopy +from dataclasses import dataclass +from typing import Literal + +from wf_core.context_contracts import ( + STANDARD_CONTEXT_FIELDS, + ContextFieldContract, + ContextSchema, + foreach_context_fields, +) +from wf_core.models.steps import ForeachNode +from wf_core.models.workflow import Edge, Workflow +from wf_core.tokens import END + +type ContextAvailability = Literal["available", "conditional"] +type FrameScope = str | None + + +@dataclass(frozen=True, slots=True) +class ContextFieldAvailability: + """One context contract plus whether it is guaranteed at a graph node.""" + + contract: ContextFieldContract + availability: ContextAvailability + reason: str | None = None + + @property + def name(self) -> str: + return self.contract.name + + @property + def schema(self) -> ContextSchema: + return self.contract.schema + + @property + def description(self) -> str: + return self.contract.description + + +@dataclass(slots=True) +class _ContextAnalysis: + fields_by_node: dict[str, tuple[ContextFieldAvailability, ...]] + warnings: tuple[str, ...] + + +class _Warnings: + def __init__(self) -> None: + self.values: list[str] = [] + self.seen: set[str] = set() + + def add(self, value: str) -> None: + if value not in self.seen: + self.seen.add(value) + self.values.append(value) + + +def context_fields_by_node( + workflow: Workflow, +) -> dict[str, tuple[ContextFieldAvailability, ...]]: + """Return runtime context contracts for every reachable graph node. + + This is an abstract execution-frame analysis rather than ordinary graph + reachability: the same node can execute in the root frame and in a + foreach child frame, and those frames expose different context keys. + The traversal memoizes both node id and active frame scope so cyclic + graphs terminate without granting aliases from an impossible scope. + """ + return _analyze(workflow).fields_by_node + + +def context_analysis_warnings(workflow: Workflow) -> tuple[str, ...]: + """Return bounded warnings found while analyzing workflow frame scopes.""" + return _analyze(workflow).warnings + + +def _analyze(workflow: Workflow) -> _ContextAnalysis: + nodes = {node.id: node for node in workflow.nodes} + foreach_nodes = { + node.id: node for node in workflow.nodes if isinstance(node, ForeachNode) + } + edges_by_node: dict[str, list[Edge]] = {} + warnings = _Warnings() + + for edge in workflow.edges: + edges_by_node.setdefault(edge.from_, []).append(edge) + if edge.from_ not in nodes: + warnings.add(f"edge source {edge.from_!r} is not a workflow node") + if edge.to != END and edge.to not in nodes: + warnings.add(f"edge from {edge.from_!r} targets missing node {edge.to!r}") + + for foreach in foreach_nodes.values(): + if not any( + edge.outcome == "loop" for edge in edges_by_node.get(foreach.id, []) + ): + warnings.add( + f"foreach node {foreach.id!r} has no loop route; " + "no scoped alias is guaranteed" + ) + + if workflow.start not in nodes: + warnings.add(f"workflow start targets missing node {workflow.start!r}") + return _ContextAnalysis({}, tuple(warnings.values)) + + scopes_by_node: dict[str, set[FrameScope]] = {} + pending: deque[tuple[str, FrameScope]] = deque([(workflow.start, None)]) + visited: set[tuple[str, FrameScope]] = set() + while pending: + node_id, active_scope = pending.popleft() + state = (node_id, active_scope) + if state in visited: + continue + visited.add(state) + node = nodes.get(node_id) + if node is None: + continue + scopes_by_node.setdefault(node_id, set()).add(active_scope) + + for edge in edges_by_node.get(node_id, []): + if edge.to == END or edge.to not in nodes: + continue + next_scope = active_scope + if isinstance(node, ForeachNode) and edge.outcome == "loop": + next_scope = node.id + pending.append((edge.to, next_scope)) + + fields_by_node: dict[str, tuple[ContextFieldAvailability, ...]] = {} + for node_id, scopes in scopes_by_node.items(): + fields_by_node[node_id] = _available_fields( + workflow, + foreach_nodes, + node_id, + scopes, + ) + return _ContextAnalysis(fields_by_node, tuple(warnings.values)) + + +def _available_fields( + workflow: Workflow, + foreach_nodes: Mapping[str, ForeachNode], + node_id: str, + scopes: set[FrameScope], +) -> tuple[ContextFieldAvailability, ...]: + del node_id + fields_by_name: dict[str, ContextFieldContract] = {} + scopes_by_field: dict[str, set[FrameScope]] = {} + for scope in sorted(scopes, key=lambda value: value or ""): + contracts = STANDARD_CONTEXT_FIELDS + if scope is not None: + foreach = foreach_nodes.get(scope) + if foreach is not None: + contracts = ( + *contracts, + *foreach_context_fields( + foreach.as_, + _foreach_item_schema(workflow, foreach, scope, foreach_nodes), + ), + ) + for contract in contracts: + fields_by_name.setdefault( + contract.name, + ContextFieldContract( + contract.name, + deepcopy(contract.schema), + contract.description, + ), + ) + scopes_by_field.setdefault(contract.name, set()).add(scope) + + field_count = len(scopes) + result: list[ContextFieldAvailability] = [] + for contract in fields_by_name.values(): + field_scopes = scopes_by_field[contract.name] + availability: ContextAvailability = ( + "available" if len(field_scopes) == field_count else "conditional" + ) + reason = None + if availability == "conditional": + reason = "Available only in some reachable execution frames." + result.append( + ContextFieldAvailability( + contract=contract, + availability=availability, + reason=reason, + ) + ) + return tuple(result) + + +def _foreach_item_schema( + workflow: Workflow, + foreach: ForeachNode, + active_scope: FrameScope, + foreach_nodes: Mapping[str, ForeachNode], +) -> ContextSchema: + source_schema = _schema_at_path( + workflow, foreach.over.root, foreach.over.parts, active_scope, foreach_nodes + ) + if not isinstance(source_schema, Mapping): + return {} + source_type = source_schema.get("type") + is_array = source_type == "array" or ( + isinstance(source_type, list) and "array" in source_type + ) + items = source_schema.get("items") + return deepcopy(dict(items)) if is_array and isinstance(items, Mapping) else {} + + +def _schema_at_path( + workflow: Workflow, + root: str, + parts: tuple[str, ...], + active_scope: FrameScope, + foreach_nodes: Mapping[str, ForeachNode], +) -> Mapping[str, object] | None: + if root == "input": + current: object = workflow.input_schema.model_dump( + mode="json", exclude_none=True + ) + elif root == "state": + current = workflow.state_schema.model_dump(mode="json", exclude_none=True) + elif root == "context": + current = {field.name: field.schema for field in STANDARD_CONTEXT_FIELDS} + if active_scope is not None: + foreach = foreach_nodes.get(active_scope) + if foreach is not None: + current.update( + { + field.name: field.schema + for field in foreach_context_fields( + foreach.as_, + _foreach_item_schema( + workflow, + foreach, + None, + foreach_nodes, + ), + ) + } + ) + else: + return None + + for part in parts: + if not isinstance(current, Mapping): + return None + properties = current.get("properties") + if not isinstance(properties, Mapping): + return None + current = properties.get(part) + return current if isinstance(current, Mapping) else None diff --git a/src/wf_core/context_contracts.py b/src/wf_core/context_contracts.py new file mode 100644 index 00000000..0b0092f9 --- /dev/null +++ b/src/wf_core/context_contracts.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass +from typing import Any + +type ContextSchema = dict[str, Any] + +PRIOR_OUTCOME_CONTEXT_KEY = "prior_outcome" +ACTIVATED_INCOMING_EDGE_CONTEXT_KEY = "activated_incoming_edge" +SCOPE_ID_CONTEXT_KEY = "scope_id" +LINEAGE_ID_CONTEXT_KEY = "lineage_id" +PARENT_LINEAGE_ID_CONTEXT_KEY = "parent_lineage_id" +LOOP_ITEM_CONTEXT_KEY = "loop_item" +LOOP_INDEX_CONTEXT_KEY = "loop_index" + + +@dataclass(frozen=True, slots=True) +class ContextFieldContract: + """Semantic contract for one key exposed by a runtime execution frame.""" + + name: str + schema: ContextSchema + description: str + + +STANDARD_CONTEXT_FIELDS = ( + ContextFieldContract( + PRIOR_OUTCOME_CONTEXT_KEY, + {"type": ["string", "null"]}, + "Prior route outcome", + ), + ContextFieldContract( + ACTIVATED_INCOMING_EDGE_CONTEXT_KEY, + {"type": ["string", "null"]}, + "Incoming step id", + ), + ContextFieldContract( + SCOPE_ID_CONTEXT_KEY, + {"type": "string"}, + "Execution scope id", + ), + ContextFieldContract( + LINEAGE_ID_CONTEXT_KEY, + {"type": "string"}, + "Execution lineage id", + ), + ContextFieldContract( + PARENT_LINEAGE_ID_CONTEXT_KEY, + {"type": ["string", "null"]}, + "Parent lineage id", + ), +) + + +def foreach_context_fields( + alias: str, + item_schema: ContextSchema, +) -> tuple[ContextFieldContract, ...]: + """Return the iteration keys, including a configured item alias once.""" + item_contract = ContextFieldContract( + LOOP_ITEM_CONTEXT_KEY, + deepcopy(item_schema), + "Current foreach item", + ) + index_contract = ContextFieldContract( + LOOP_INDEX_CONTEXT_KEY, + {"type": "integer"}, + "Current foreach item index", + ) + fields = [item_contract, index_contract] + if alias and alias not in {LOOP_ITEM_CONTEXT_KEY, LOOP_INDEX_CONTEXT_KEY}: + fields.append( + ContextFieldContract( + alias, + deepcopy(item_schema), + "Current foreach item", + ) + ) + return tuple(fields) diff --git a/src/wf_core/runtime/ops/frames.py b/src/wf_core/runtime/ops/frames.py index 30b16fb9..0cf61c3e 100644 --- a/src/wf_core/runtime/ops/frames.py +++ b/src/wf_core/runtime/ops/frames.py @@ -1,22 +1,31 @@ from __future__ import annotations +from wf_core.context_contracts import ( + ACTIVATED_INCOMING_EDGE_CONTEXT_KEY, + LINEAGE_ID_CONTEXT_KEY, + LOOP_INDEX_CONTEXT_KEY, + LOOP_ITEM_CONTEXT_KEY, + PARENT_LINEAGE_ID_CONTEXT_KEY, + PRIOR_OUTCOME_CONTEXT_KEY, + SCOPE_ID_CONTEXT_KEY, +) from wf_core.run_state import ExecutionFrame def frame_context_values(frame: ExecutionFrame) -> dict[str, object | None]: context: dict[str, object | None] = { - "prior_outcome": frame.prior_outcome, - "activated_incoming_edge": frame.activated_incoming_edge, - "scope_id": frame.scope_id, - "lineage_id": frame.lineage_id, - "parent_lineage_id": frame.parent_lineage_id, + PRIOR_OUTCOME_CONTEXT_KEY: frame.prior_outcome, + ACTIVATED_INCOMING_EDGE_CONTEXT_KEY: frame.activated_incoming_edge, + SCOPE_ID_CONTEXT_KEY: frame.scope_id, + LINEAGE_ID_CONTEXT_KEY: frame.lineage_id, + PARENT_LINEAGE_ID_CONTEXT_KEY: frame.parent_lineage_id, } if frame.kind == "foreach_iteration": loop_item = frame.metadata.get("loop_item") loop_index = frame.metadata.get("loop_index") loop_alias = frame.metadata.get("loop_alias") - context["loop_item"] = loop_item - context["loop_index"] = loop_index + context[LOOP_ITEM_CONTEXT_KEY] = loop_item + context[LOOP_INDEX_CONTEXT_KEY] = loop_index if isinstance(loop_alias, str) and loop_alias: context[loop_alias] = loop_item return context diff --git a/tests/core/test_context_scopes.py b/tests/core/test_context_scopes.py new file mode 100644 index 00000000..fc1e1849 --- /dev/null +++ b/tests/core/test_context_scopes.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +from wf_core import END, Edge, ForeachNode, NodeUse, SchemaRef, StateSchema, Workflow +from wf_core.analysis.context_scopes import ( + context_analysis_warnings, + context_fields_by_node, +) +from wf_core.context_contracts import STANDARD_CONTEXT_FIELDS, foreach_context_fields +from wf_core.run_state import ExecutionFrame +from wf_core.runtime.ops.frames import frame_context_values + + +def _node(node_id: str) -> NodeUse: + return NodeUse(id=node_id, type="node", node="noop") + + +def _foreach( + node_id: str, + *, + alias: str, + mode: str = "serial", + over: str = "state.items", +) -> ForeachNode: + data: dict[str, object] = { + "id": node_id, + "type": "foreach", + "over": over, + "as": alias, + "mode": mode, + } + if mode == "concurrent": + data["concurrent"] = {"max_active": 2, "max_outstanding": 2} + return ForeachNode.model_validate(data) + + +def _workflow( + *, + start: str, + nodes: list[object], + edges: list[dict[str, str]], + state_schema: dict[str, object] | None = None, +) -> Workflow: + return Workflow( + name="context-analysis", + input_schema=SchemaRef(type="object"), + state_schema=StateSchema.model_validate( + state_schema + or { + "type": "object", + "properties": {"items": {"type": "array", "items": {"type": "string"}}}, + } + ), + output_schema=SchemaRef(type="object"), + start=start, + nodes=nodes, + edges=[Edge.model_validate(edge) for edge in edges], + ) + + +def _field_map(workflow: Workflow, node_id: str) -> dict[str, object]: + return { + field.contract.name: field + for field in context_fields_by_node(workflow)[node_id] + } + + +def test_frame_context_values_uses_standard_and_foreach_contract_keys() -> None: + ordinary = frame_context_values( + ExecutionFrame( + id="root", + kind="root", + node_id="plain", + prior_outcome="ok", + activated_incoming_edge="start", + ) + ) + assert ordinary["prior_outcome"] == "ok" + assert ordinary["activated_incoming_edge"] == "start" + assert ordinary["scope_id"] == "root" + assert ordinary["lineage_id"] == "root" + assert ordinary["parent_lineage_id"] is None + assert "loop_item" not in ordinary + + iteration = ExecutionFrame( + id="root:each:0", + kind="foreach_iteration", + node_id="body", + metadata={"loop_item": "a", "loop_index": 0, "loop_alias": "item"}, + ) + context = frame_context_values(iteration) + assert context["loop_item"] == "a" + assert context["loop_index"] == 0 + assert context["item"] == "a" + + +def test_context_contracts_deduplicate_aliases_that_are_standard_loop_keys() -> None: + assert STANDARD_CONTEXT_FIELDS[0].schema == {"type": ["string", "null"]} + assert [field.name for field in foreach_context_fields("loop_item", {})] == [ + "loop_item", + "loop_index", + ] + + +def test_serial_and_concurrent_foreach_expose_the_same_scoped_context() -> None: + for mode in ("serial", "concurrent"): + workflow = _workflow( + start="each", + nodes=[ + _foreach("each", alias="item", mode=mode), + _node("body"), + _node("tail"), + ], + edges=[ + {"from": "each", "outcome": "loop", "to": "body"}, + {"from": "each", "outcome": "done", "to": "tail"}, + {"from": "body", "outcome": "ok", "to": END}, + {"from": "tail", "outcome": "ok", "to": END}, + ], + ) + + body = _field_map(workflow, "body") + assert body["loop_item"].availability == "available" + assert body["loop_index"].availability == "available" + assert body["item"].availability == "available" + assert "item" not in _field_map(workflow, "tail") + + +def test_foreach_item_schema_and_configured_alias_are_reported() -> None: + workflow = _workflow( + start="each", + nodes=[_foreach("each", alias="record"), _node("body")], + edges=[ + {"from": "each", "outcome": "loop", "to": "body"}, + {"from": "body", "outcome": "ok", "to": END}, + {"from": "each", "outcome": "done", "to": END}, + ], + ) + + fields = _field_map(workflow, "body") + assert fields["record"].contract.schema == {"type": "string"} + assert fields["loop_item"].contract.schema == {"type": "string"} + assert fields["loop_index"].contract.schema == {"type": "integer"} + + +def test_only_foreach_reachable_node_has_available_context() -> None: + workflow = _workflow( + start="start", + nodes=[_node("start"), _foreach("each", alias="item"), _node("body")], + edges=[ + {"from": "start", "outcome": "ok", "to": "body"}, + {"from": "start", "outcome": "loop", "to": "each"}, + {"from": "each", "outcome": "loop", "to": "body"}, + {"from": "each", "outcome": "done", "to": END}, + {"from": "body", "outcome": "ok", "to": END}, + ], + ) + + assert _field_map(workflow, "body")["item"].availability == "conditional" + assert _field_map(workflow, "body")["item"].reason + + +def test_nested_foreach_replaces_inner_scope_and_restores_outer_scope() -> None: + workflow = _workflow( + start="outer", + nodes=[ + _foreach("outer", alias="outer_item"), + _foreach("inner", alias="inner_item", over="state.inner_items"), + _node("inner_body"), + _node("after_inner"), + ], + edges=[ + {"from": "outer", "outcome": "loop", "to": "inner"}, + {"from": "inner", "outcome": "loop", "to": "inner_body"}, + {"from": "inner", "outcome": "done", "to": "after_inner"}, + {"from": "inner_body", "outcome": "ok", "to": END}, + {"from": "after_inner", "outcome": "ok", "to": END}, + {"from": "outer", "outcome": "done", "to": END}, + ], + state_schema={ + "type": "object", + "properties": { + "items": {"type": "array", "items": {"type": "string"}}, + "inner_items": {"type": "array", "items": {"type": "integer"}}, + }, + }, + ) + + inner = _field_map(workflow, "inner_body") + after_inner = _field_map(workflow, "after_inner") + assert "outer_item" not in inner + assert inner["inner_item"].availability == "available" + assert after_inner["outer_item"].availability == "available" + assert "inner_item" not in after_inner + + +def test_malformed_routes_warn_without_granting_a_scoped_alias() -> None: + workflow = _workflow( + start="each", + nodes=[_foreach("each", alias="item"), _node("body")], + edges=[ + {"from": "each", "outcome": "done", "to": "missing"}, + {"from": "each", "outcome": "ok", "to": "body"}, + {"from": "body", "outcome": "ok", "to": END}, + ], + ) + + assert "item" not in _field_map(workflow, "body") + warnings = context_analysis_warnings(workflow) + assert any("missing" in warning for warning in warnings) + assert any("loop" in warning for warning in warnings) + + +def test_cyclic_graph_analysis_memoizes_node_and_frame_scope() -> None: + workflow = _workflow( + start="a", + nodes=[_node("a"), _node("b")], + edges=[ + {"from": "a", "outcome": "ok", "to": "b"}, + {"from": "b", "outcome": "ok", "to": "a"}, + ], + ) + + fields = context_fields_by_node(workflow) + assert set(fields) == {"a", "b"} + assert fields["a"] + assert fields["b"] diff --git a/tests/core/test_scheduler.py b/tests/core/test_scheduler.py index 95d57e47..741d24e7 100644 --- a/tests/core/test_scheduler.py +++ b/tests/core/test_scheduler.py @@ -7,6 +7,7 @@ from wf_core.models.schemas import SchemaRef, StateSchema from wf_core.models.workflow import Workflow from wf_core.run_state import ExecutionFrame, FrameStatus, RunState, RunStatus from wf_core.runtime.ops.flow import advance_frame +from wf_core.runtime.ops.frames import frame_context_values from wf_core.runtime.ops.runs import create_run_state from wf_core.runtime.scheduler import ( add_frame, @@ -207,3 +208,22 @@ def test_deadlock_error_includes_ready_queue_and_frame_summary() -> None: assert "deadlocked" in message assert "ready_frame_ids=[]" in message assert "parent:blocked@foreach" in message + + +def test_frame_context_values_exposes_configured_foreach_alias() -> None: + context = frame_context_values( + ExecutionFrame( + id="child", + kind="foreach_iteration", + node_id="body", + metadata={ + "loop_item": {"id": "a"}, + "loop_index": 2, + "loop_alias": "record", + }, + ) + ) + + assert context["loop_item"] == {"id": "a"} + assert context["loop_index"] == 2 + assert context["record"] == {"id": "a"} diff --git a/tests/wf_api/test_authoring_contracts.py b/tests/wf_api/test_authoring_contracts.py index 7be42b7b..31056de7 100644 --- a/tests/wf_api/test_authoring_contracts.py +++ b/tests/wf_api/test_authoring_contracts.py @@ -1,6 +1,7 @@ from __future__ import annotations from wf_api.authoring_contracts import ( + context_path_options, project_authoring_contract_inventory, schema_path_options, ) @@ -328,3 +329,48 @@ def test_project_authoring_contract_inventory_composes_pure_inputs() -> None: assert inventory["entry_steps"] == [entry_step] assert inventory["workflow_outcomes"] == ["ok", "error"] assert inventory["warnings"] == ["selected step has conditional context"] + + +def test_context_path_options_are_step_input_only() -> None: + options = context_path_options( + [ + { + "name": "loop_item", + "schema": {"type": "string"}, + "description": "Current foreach item", + "availability": "conditional", + "reason": "Only available inside the foreach body.", + } + ] + ) + + assert options[0]["path"] == "context.loop_item" + assert options[0]["origin"] == "runtime_context" + assert options[0]["uses"] == ["step_input"] + assert options[0]["availability"] == "conditional" + assert options[0]["reason"] == "Only available inside the foreach body." + + +def test_project_inventory_does_not_offer_context_for_workflow_output() -> None: + context_entry = { + "path": "context.item", + "label": "Item", + "origin": "runtime_context", + "schema": {}, + "required": False, + "availability": "available", + "uses": ["step_input", "workflow_output"], + } + + inventory = project_authoring_contract_inventory( + workspace_id="workspace-1", + revision=1, + selected_step_id=None, + input_schema={"type": "object"}, + state_schema={"type": "object"}, + output_schema={"type": "object"}, + context_entries=[context_entry], + ) + + assert inventory["readable_sources"][0]["path"] == "context.item" + assert inventory["readable_sources"][0]["uses"] == ["step_input"]