From 0f1ed5687614dc7838708970dc912b0435f426ef Mon Sep 17 00:00:00 2001 From: lda Date: Fri, 14 Aug 2026 15:17:06 +0700 Subject: [PATCH] fix: harden runtime context analysis --- .../task-2-report.md | 55 +++++++++ src/wf_core/analysis/context_scopes.py | 104 ++++++++++++++---- src/wf_core/context_contracts.py | 9 +- src/wf_core/runtime/ops/frames.py | 7 +- tests/core/test_context_scopes.py | 82 ++++++++++++++ 5 files changed, 236 insertions(+), 21 deletions(-) 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 index b57a64ff..d61a9eff 100644 --- 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 @@ -70,3 +70,58 @@ uv run basedpyright --level error src/wf_core/context_contracts.py src/wf_core/a - The authoring projector accepts an optional workflow for automatic context projection, while existing callers can continue supplying Task 1 payloads explicitly. + +## Round 1 Fix + +The review identified three Important findings and the fixes were tested at +their affected seams. + +### 1. Bounded local `$ref` item schemas + +Added `test_foreach_item_schema_resolves_bounded_local_array_reference` before +the production change. Its targeted RED run failed because `loop_item` had no +`type` after a declared `state.items` array was selected through +`#/$defs/Items`. + +The analyzer now resolves bounded local `$defs`/`definitions` references for +both the selected array source and its `items` schema. Cyclic, unsupported, or +unresolved references remain conservative and produce `{}`. + +### 2. Reserved context aliases + +Added `test_all_standard_context_names_are_reserved_from_foreach_aliases` +before the production change. Its targeted RED run failed because +`prior_outcome` was accepted as a foreach alias. + +The shared `RESERVED_CONTEXT_KEYS` registry now covers every standard context +name plus `loop_item` and `loop_index`. Both contract generation and +`frame_context_values` use it, so runtime values and authoring inventory cannot +disagree on a colliding alias. + +### 3. Scoped-cycle regression + +Added `test_scoped_cycle_terminates_and_preserves_scoped_field_availability`, +which enters a foreach body, routes back to the foreach node under the child +scope, and asserts the body/owner availability results. The test passed before +the round-1 production changes because Task 2 already memoized +`(node_id, active_foreach_id)` correctly; this finding required regression +coverage but did not require a production change. + +Round 1 verification: + +```text +uv run pytest tests/core/test_context_scopes.py tests/core/test_scheduler.py tests/wf_api/test_authoring_contracts.py -q +32 passed + +uv run pytest tests/core -q +299 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 ruff format --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 +8 files already formatted + +uv run basedpyright --level error src/wf_core/context_contracts.py src/wf_core/analysis src/wf_core/runtime/ops/frames.py src/wf_api/authoring_contracts.py +0 errors, 0 warnings, 0 notes +``` diff --git a/src/wf_core/analysis/context_scopes.py b/src/wf_core/analysis/context_scopes.py index 2fa69a6f..3dd3fc59 100644 --- a/src/wf_core/analysis/context_scopes.py +++ b/src/wf_core/analysis/context_scopes.py @@ -19,6 +19,8 @@ from wf_core.tokens import END type ContextAvailability = Literal["available", "conditional"] type FrameScope = str | None +_MAX_LOCAL_SCHEMA_REFERENCE_DEPTH = 32 + @dataclass(frozen=True, slots=True) class ContextFieldAvailability: @@ -206,7 +208,15 @@ def _foreach_item_schema( 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 {} + if not is_array or not isinstance(items, Mapping): + return {} + try: + resolved_items = _resolve_local_reference( + _schema_document(workflow, foreach.over.root), items + ) + except ValueError: + return {} + return deepcopy(dict(resolved_items)) def _schema_at_path( @@ -216,15 +226,45 @@ def _schema_at_path( 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 + try: + schema_document = _schema_document( + workflow, + root, + active_scope=active_scope, + foreach_nodes=foreach_nodes, ) - 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: + current: object = schema_document + for part in parts: + if not isinstance(current, Mapping): + return None + resolved = _resolve_local_reference(schema_document, current) + properties = resolved.get("properties") + if not isinstance(properties, Mapping): + return None + current = properties.get(part) + if not isinstance(current, Mapping): + return None + return _resolve_local_reference(schema_document, current) + except ValueError: + return None + + +def _schema_document( + workflow: Workflow, + root: str, + *, + active_scope: FrameScope = None, + foreach_nodes: Mapping[str, ForeachNode] | None = None, +) -> Mapping[str, object]: + if root == "input": + return workflow.input_schema.model_dump(mode="json", exclude_none=True) + if root == "state": + return workflow.state_schema.model_dump(mode="json", exclude_none=True) + if root == "context": + current: dict[str, object] = { + field.name: field.schema for field in STANDARD_CONTEXT_FIELDS + } + if active_scope is not None and foreach_nodes is not None: foreach = foreach_nodes.get(active_scope) if foreach is not None: current.update( @@ -241,14 +281,40 @@ def _schema_at_path( ) } ) - else: - return None + return {"type": "object", "properties": current} + return {} - 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 + +def _resolve_local_reference( + root_schema: Mapping[str, object], + candidate: Mapping[str, object], +) -> Mapping[str, object]: + """Resolve bounded repository-local refs without becoming a full resolver.""" + current = candidate + seen: set[str] = set() + while "$ref" in current: + reference = current["$ref"] + if not isinstance(reference, str): + raise ValueError("schema reference must be a string") + if reference in seen: + raise ValueError(f"cyclic schema reference {reference!r}") + if len(seen) >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH: + raise ValueError( + f"local schema reference depth exceeds " + f"{_MAX_LOCAL_SCHEMA_REFERENCE_DEPTH}" + ) + if not ( + reference.startswith("#/$defs/") or reference.startswith("#/definitions/") + ): + raise ValueError(f"unsupported schema reference {reference!r}") + seen.add(reference) + resolved: object = root_schema + for raw_part in reference.removeprefix("#/").split("/"): + part = raw_part.replace("~1", "/").replace("~0", "~") + if not isinstance(resolved, Mapping) or part not in resolved: + raise ValueError(f"unresolved schema reference {reference!r}") + resolved = resolved[part] + if not isinstance(resolved, Mapping): + raise ValueError(f"schema reference {reference!r} is not an object") + current = resolved + return current diff --git a/src/wf_core/context_contracts.py b/src/wf_core/context_contracts.py index 0b0092f9..62845819 100644 --- a/src/wf_core/context_contracts.py +++ b/src/wf_core/context_contracts.py @@ -51,6 +51,13 @@ STANDARD_CONTEXT_FIELDS = ( "Parent lineage id", ), ) +STANDARD_CONTEXT_FIELD_NAMES = frozenset( + field.name for field in STANDARD_CONTEXT_FIELDS +) +RESERVED_CONTEXT_KEYS = STANDARD_CONTEXT_FIELD_NAMES | { + LOOP_ITEM_CONTEXT_KEY, + LOOP_INDEX_CONTEXT_KEY, +} def foreach_context_fields( @@ -69,7 +76,7 @@ def foreach_context_fields( "Current foreach item index", ) fields = [item_contract, index_contract] - if alias and alias not in {LOOP_ITEM_CONTEXT_KEY, LOOP_INDEX_CONTEXT_KEY}: + if alias and alias not in RESERVED_CONTEXT_KEYS: fields.append( ContextFieldContract( alias, diff --git a/src/wf_core/runtime/ops/frames.py b/src/wf_core/runtime/ops/frames.py index 0cf61c3e..22422927 100644 --- a/src/wf_core/runtime/ops/frames.py +++ b/src/wf_core/runtime/ops/frames.py @@ -7,6 +7,7 @@ from wf_core.context_contracts import ( LOOP_ITEM_CONTEXT_KEY, PARENT_LINEAGE_ID_CONTEXT_KEY, PRIOR_OUTCOME_CONTEXT_KEY, + RESERVED_CONTEXT_KEYS, SCOPE_ID_CONTEXT_KEY, ) from wf_core.run_state import ExecutionFrame @@ -26,6 +27,10 @@ def frame_context_values(frame: ExecutionFrame) -> dict[str, object | None]: loop_alias = frame.metadata.get("loop_alias") context[LOOP_ITEM_CONTEXT_KEY] = loop_item context[LOOP_INDEX_CONTEXT_KEY] = loop_index - if isinstance(loop_alias, str) and loop_alias: + if ( + isinstance(loop_alias, str) + and loop_alias + and loop_alias not in RESERVED_CONTEXT_KEYS + ): context[loop_alias] = loop_item return context diff --git a/tests/core/test_context_scopes.py b/tests/core/test_context_scopes.py index fc1e1849..8285ebe2 100644 --- a/tests/core/test_context_scopes.py +++ b/tests/core/test_context_scopes.py @@ -101,6 +101,37 @@ def test_context_contracts_deduplicate_aliases_that_are_standard_loop_keys() -> ] +def test_all_standard_context_names_are_reserved_from_foreach_aliases() -> None: + expected_values = { + "prior_outcome": "ok", + "activated_incoming_edge": "start", + "scope_id": "scope", + "lineage_id": "lineage", + "parent_lineage_id": "parent", + } + standard_names = {field.name for field in STANDARD_CONTEXT_FIELDS} + + for name in standard_names: + foreach_names = { + field.name for field in foreach_context_fields(name, {"type": "string"}) + } + assert name not in foreach_names + context = frame_context_values( + ExecutionFrame( + id="child", + kind="foreach_iteration", + node_id="body", + scope_id="scope", + lineage_id="lineage", + parent_lineage_id="parent", + prior_outcome="ok", + activated_incoming_edge="start", + metadata={"loop_item": "item", "loop_index": 0, "loop_alias": name}, + ) + ) + assert context[name] == expected_values[name] + + def test_serial_and_concurrent_foreach_expose_the_same_scoped_context() -> None: for mode in ("serial", "concurrent"): workflow = _workflow( @@ -142,6 +173,40 @@ def test_foreach_item_schema_and_configured_alias_are_reported() -> None: assert fields["loop_index"].contract.schema == {"type": "integer"} +def test_foreach_item_schema_resolves_bounded_local_array_reference() -> 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}, + ], + state_schema={ + "type": "object", + "properties": {"items": {"$ref": "#/$defs/Items"}}, + "$defs": { + "Items": { + "type": "array", + "items": {"$ref": "#/$defs/Item"}, + }, + "Item": { + "type": "object", + "properties": {"id": {"type": "string"}}, + "required": ["id"], + }, + }, + }, + ) + + fields = _field_map(workflow, "body") + assert fields["loop_item"].contract.schema["type"] == "object" + assert fields["loop_item"].contract.schema["properties"] == { + "id": {"type": "string"} + } + assert fields["record"].contract.schema["properties"] == {"id": {"type": "string"}} + + def test_only_foreach_reachable_node_has_available_context() -> None: workflow = _workflow( start="start", @@ -224,3 +289,20 @@ def test_cyclic_graph_analysis_memoizes_node_and_frame_scope() -> None: assert set(fields) == {"a", "b"} assert fields["a"] assert fields["b"] + + +def test_scoped_cycle_terminates_and_preserves_scoped_field_availability() -> None: + workflow = _workflow( + start="each", + nodes=[_foreach("each", alias="item"), _node("body")], + edges=[ + {"from": "each", "outcome": "loop", "to": "body"}, + {"from": "body", "outcome": "ok", "to": "each"}, + {"from": "each", "outcome": "done", "to": END}, + ], + ) + + fields = context_fields_by_node(workflow) + assert fields["body"] + assert _field_map(workflow, "body")["item"].availability == "available" + assert _field_map(workflow, "each")["item"].availability == "conditional"