diff --git a/.superpowers/sdd/2026-08-14-workflow-console-contract-graph/task-1-report.md b/.superpowers/sdd/2026-08-14-workflow-console-contract-graph/task-1-report.md new file mode 100644 index 00000000..c658ab57 --- /dev/null +++ b/.superpowers/sdd/2026-08-14-workflow-console-contract-graph/task-1-report.md @@ -0,0 +1,60 @@ +# Task 1 Report: Model Schema-Derived Authoring Choices + +## Status + +Implemented Task 1 of the workflow contract graph backend slice. + +## Changes + +- Added explicit transport payload types for path options, step contracts, and + revision-scoped authoring inventories. +- Added `schema_path_options`, which derives deterministic parent-before-child + choices from JSON Schema object properties. +- Preserved whole arrays as selectable paths without synthetic wildcard paths. +- Omitted invented child names for unconstrained or open-ended + `additionalProperties`. +- Reused `schema_fragment_at_location` and its bounded local-reference depth for + nested schema fragments and `$defs`/`definitions` references. +- Derived labels from schema titles or humanized path segments and copied only + string descriptions. +- Added pure inventory composition for input/state/context sources, selected + step targets and sources, state/output targets, entry steps, outcomes, and + warnings. +- Re-exported the new payload types through `wf_api.models`. + +## Test-First Evidence + +The required RED command was run before production implementation: + +```text +uv run pytest tests/wf_api/test_authoring_contracts.py -q +``` + +It failed during collection with: + +```text +ModuleNotFoundError: No module named 'wf_api.authoring_contracts' +``` + +After implementation, the focused authoring-contract tests passed. + +## Verification + +```text +uv run pytest tests/wf_api/test_authoring_contracts.py tests/wf_api/test_schema_projection.py -q +48 passed + +uv run basedpyright --level error src/wf_api/authoring_contracts.py src/wf_api/models/authoring_contracts.py +0 errors, 0 warnings, 0 notes + +uv run ruff check src/wf_api/authoring_contracts.py src/wf_api/models/authoring_contracts.py src/wf_api/models/__init__.py tests/wf_api/test_authoring_contracts.py +All checks passed! + +uv run ruff format --check src/wf_api/authoring_contracts.py src/wf_api/models/authoring_contracts.py src/wf_api/models/__init__.py tests/wf_api/test_authoring_contracts.py +4 files already formatted +``` + +## Concerns + +None for the Task 1 scope. Runtime-context analysis and persisted workspace or +capability loading remain intentionally deferred to Tasks 2 and 3. diff --git a/src/wf_api/authoring_contracts.py b/src/wf_api/authoring_contracts.py new file mode 100644 index 00000000..61d7aa6e --- /dev/null +++ b/src/wf_api/authoring_contracts.py @@ -0,0 +1,231 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from copy import deepcopy +from typing import Any + +from .models.authoring_contracts import ( + AuthoringContractInventoryPayload, + AuthoringPathOptionPayload, + AuthoringPathOrigin, + AuthoringPathUse, + AuthoringStepContractPayload, +) +from .models.common import JsonObject +from .schema_projection import ( + _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH, + schema_fragment_at_location, +) + +_ROOTS: dict[str, tuple[str, AuthoringPathOrigin]] = { + "input": ("input", "workflow_input"), + "state": ("state", "workflow_state"), + "context": ("context", "runtime_context"), + "step_input": ("step_input", "step_input"), + "step_output": ("step_output", "step_output"), + "output": ("output", "workflow_output"), + "workflow_output": ("output", "workflow_output"), +} + + +def schema_path_options( + schema: JsonObject, + *, + root: str, + uses: Sequence[AuthoringPathUse], +) -> list[AuthoringPathOptionPayload]: + """Flatten declared object paths into deterministic authoring choices. + + Arrays are kept as whole values because an array item's path needs a real + runtime index. Open-ended ``additionalProperties`` likewise contributes no + invented child names. + """ + try: + prefix, origin = _ROOTS[root] + except KeyError as exc: + raise ValueError(f"unsupported authoring schema root {root!r}") from exc + + normalized_uses = list(uses) + options: list[AuthoringPathOptionPayload] = [] + _append_schema_options( + schema, + location=(), + prefix=prefix, + origin=origin, + uses=normalized_uses, + options=options, + ) + return options + + +def project_authoring_contract_inventory( + *, + workspace_id: str, + revision: int, + selected_step_id: str | None, + input_schema: JsonObject, + state_schema: JsonObject, + output_schema: JsonObject, + context_entries: Sequence[AuthoringPathOptionPayload] = (), + step_input_targets: Sequence[AuthoringPathOptionPayload] = (), + step_output_sources: Sequence[AuthoringPathOptionPayload] = (), + entry_steps: Sequence[AuthoringStepContractPayload] = (), + workflow_outcomes: Sequence[str] = (), + warnings: Sequence[str] = (), +) -> AuthoringContractInventoryPayload: + """Compose an inventory from caller-provided schemas and graph facts. + + This projector deliberately has no store or capability dependencies. The + service layer supplies the selected-step and runtime-context projections; + this function only derives schema choices and copies those projections. + """ + input_sources = schema_path_options( + input_schema, + root="input", + uses=["step_input", "workflow_output"], + ) + state_sources = schema_path_options( + state_schema, + root="state", + uses=["step_input", "step_output_source", "workflow_output"], + ) + state_targets = schema_path_options( + state_schema, + root="state", + uses=["state_target"], + ) + workflow_output_targets = schema_path_options( + output_schema, + root="output", + uses=["workflow_output"], + ) + + return { + "workspace_id": workspace_id, + "revision": revision, + "selected_step_id": selected_step_id, + "readable_sources": [ + *input_sources, + *deepcopy(list(context_entries)), + *state_sources, + ], + "step_input_targets": deepcopy(list(step_input_targets)), + "step_output_sources": deepcopy(list(step_output_sources)), + "state_targets": state_targets, + "workflow_output_targets": workflow_output_targets, + "entry_steps": deepcopy(list(entry_steps)), + "workflow_outcomes": list(workflow_outcomes), + "warnings": list(warnings), + } + + +def _append_schema_options( + schema: JsonObject, + *, + location: tuple[str, ...], + prefix: str, + origin: AuthoringPathOrigin, + uses: list[AuthoringPathUse], + options: list[AuthoringPathOptionPayload], +) -> None: + fragment = schema_fragment_at_location(schema, location) + resolved = _resolve_local_reference(schema, fragment) + properties = resolved.get("properties") + if not isinstance(properties, Mapping): + return + + required_values = resolved.get("required") + required = ( + set(required_values) + if isinstance(required_values, list) + and all(isinstance(value, str) for value in required_values) + else set() + ) + for name, property_schema in properties.items(): + if not isinstance(name, str) or not isinstance(property_schema, Mapping): + continue + child_location = (*location, name) + child_fragment = schema_fragment_at_location(schema, child_location) + resolved_child = _resolve_local_reference(schema, child_fragment) + path = f"{prefix}.{name}" + option: AuthoringPathOptionPayload = { + "path": path, + "label": _label_for(name, child_fragment, resolved_child), + "origin": origin, + "schema": child_fragment, + "required": name in required, + "availability": "available", + "uses": list(uses), + } + description = _description_for(child_fragment, resolved_child) + if description is not None: + option["description"] = description + options.append(option) + + if _is_array_schema(resolved_child): + continue + _append_schema_options( + schema, + location=child_location, + prefix=path, + origin=origin, + uses=uses, + options=options, + ) + + +def _resolve_local_reference( + root_schema: Mapping[str, Any], + fragment: Mapping[str, Any], +) -> Mapping[str, Any]: + current = fragment + seen: set[str] = set() + while "$ref" in current: + reference = current["$ref"] + if not isinstance(reference, str): + return current + if reference in seen: + raise ValueError(f"cyclic reference {reference!r}") + if len(seen) >= _MAX_LOCAL_SCHEMA_REFERENCE_DEPTH: + raise ValueError( + f"local reference depth exceeds {_MAX_LOCAL_SCHEMA_REFERENCE_DEPTH}" + ) + if not ( + reference.startswith("#/$defs/") or reference.startswith("#/definitions/") + ): + return current + 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 reference {reference!r}") + resolved = resolved[part] + if not isinstance(resolved, Mapping): + raise ValueError(f"reference {reference!r} is not a schema object") + current = resolved + return current + + +def _label_for( + name: str, + fragment: Mapping[str, Any], + resolved: Mapping[str, Any], +) -> str: + title = resolved.get("title", fragment.get("title")) + if isinstance(title, str) and title: + return title + return name.replace("_", " ").replace("-", " ").title() + + +def _description_for( + fragment: Mapping[str, Any], + resolved: Mapping[str, Any], +) -> str | None: + description = resolved.get("description", fragment.get("description")) + return description if isinstance(description, str) else None + + +def _is_array_schema(schema: Mapping[str, Any]) -> bool: + schema_type = schema.get("type") + return schema_type == "array" diff --git a/src/wf_api/models/__init__.py b/src/wf_api/models/__init__.py index 210c4a53..ac5c62c2 100644 --- a/src/wf_api/models/__init__.py +++ b/src/wf_api/models/__init__.py @@ -22,6 +22,14 @@ from .artifacts import ( SaveArtifactResult, WorkflowArtifactPayload, ) +from .authoring_contracts import ( + AuthoringContractInventoryPayload, + AuthoringPathAvailability, + AuthoringPathOptionPayload, + AuthoringPathOrigin, + AuthoringPathUse, + AuthoringStepContractPayload, +) from .capabilities import ( CapabilityCallResult, CapabilitySummary, @@ -122,6 +130,12 @@ __all__ = [ "AdminEventPayload", "ApplyRegistryChangesResult", "ArtifactVersionPayload", + "AuthoringContractInventoryPayload", + "AuthoringPathAvailability", + "AuthoringPathOptionPayload", + "AuthoringPathOrigin", + "AuthoringPathUse", + "AuthoringStepContractPayload", "AuthRecordSummaryPayload", "ArtifactCatalogEntryPayload", "ArtifactKindPayload", diff --git a/src/wf_api/models/authoring_contracts.py b/src/wf_api/models/authoring_contracts.py new file mode 100644 index 00000000..33cb2443 --- /dev/null +++ b/src/wf_api/models/authoring_contracts.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from typing import Literal, NotRequired, TypedDict + +from .common import JsonObject + +type AuthoringPathOrigin = Literal[ + "workflow_input", + "workflow_state", + "runtime_context", + "step_input", + "step_output", + "workflow_output", +] +type AuthoringPathAvailability = Literal["available", "conditional"] +type AuthoringPathUse = Literal[ + "step_input", + "step_output_source", + "state_target", + "workflow_output", +] + + +class AuthoringPathOptionPayload(TypedDict): + """One schema-derived source or target available to an author.""" + + path: str + label: str + origin: AuthoringPathOrigin + schema: JsonObject + required: bool + availability: AuthoringPathAvailability + uses: list[AuthoringPathUse] + description: NotRequired[str] + reason: NotRequired[str] + + +class AuthoringStepContractPayload(TypedDict): + """Compact executable-step choice used by authoring inventories.""" + + step_id: str + label: str + description: NotRequired[str] + input_targets: NotRequired[list[AuthoringPathOptionPayload]] + output_sources: NotRequired[list[AuthoringPathOptionPayload]] + outcomes: NotRequired[list[str]] + + +class AuthoringContractInventoryPayload(TypedDict): + """Revision-scoped readable sources and writable authoring targets.""" + + workspace_id: str + revision: int + selected_step_id: str | None + readable_sources: list[AuthoringPathOptionPayload] + step_input_targets: list[AuthoringPathOptionPayload] + step_output_sources: list[AuthoringPathOptionPayload] + state_targets: list[AuthoringPathOptionPayload] + workflow_output_targets: list[AuthoringPathOptionPayload] + entry_steps: list[AuthoringStepContractPayload] + workflow_outcomes: list[str] + warnings: list[str] diff --git a/tests/wf_api/test_authoring_contracts.py b/tests/wf_api/test_authoring_contracts.py new file mode 100644 index 00000000..814d5af6 --- /dev/null +++ b/tests/wf_api/test_authoring_contracts.py @@ -0,0 +1,308 @@ +from __future__ import annotations + +from wf_api.authoring_contracts import ( + project_authoring_contract_inventory, + schema_path_options, +) + + +def test_schema_path_options_is_parent_first_and_schema_derived() -> None: + schema = { + "type": "object", + "properties": { + "request": { + "title": "Request", + "description": "The incoming request.", + "type": "object", + "properties": { + "id": {"type": "string"}, + "metadata": { + "type": "object", + "additionalProperties": True, + }, + }, + "required": ["id"], + }, + "items": { + "type": "array", + "items": { + "type": "object", + "properties": {"name": {"type": "string"}}, + }, + }, + "headers": { + "type": "object", + "additionalProperties": {"type": "string"}, + }, + }, + "required": ["request", "items"], + } + + options = schema_path_options( + schema, + root="input", + uses=["step_input", "workflow_output"], + ) + + assert options == [ + { + "path": "input.request", + "label": "Request", + "origin": "workflow_input", + "schema": { + "title": "Request", + "description": "The incoming request.", + "type": "object", + "properties": { + "id": {"type": "string"}, + "metadata": {"type": "object", "additionalProperties": True}, + }, + "required": ["id"], + }, + "required": True, + "availability": "available", + "uses": ["step_input", "workflow_output"], + "description": "The incoming request.", + }, + { + "path": "input.request.id", + "label": "Id", + "origin": "workflow_input", + "schema": {"type": "string"}, + "required": True, + "availability": "available", + "uses": ["step_input", "workflow_output"], + }, + { + "path": "input.request.metadata", + "label": "Metadata", + "origin": "workflow_input", + "schema": {"type": "object", "additionalProperties": True}, + "required": False, + "availability": "available", + "uses": ["step_input", "workflow_output"], + }, + { + "path": "input.items", + "label": "Items", + "origin": "workflow_input", + "schema": { + "type": "array", + "items": { + "type": "object", + "properties": {"name": {"type": "string"}}, + }, + }, + "required": True, + "availability": "available", + "uses": ["step_input", "workflow_output"], + }, + { + "path": "input.headers", + "label": "Headers", + "origin": "workflow_input", + "schema": { + "type": "object", + "additionalProperties": {"type": "string"}, + }, + "required": False, + "availability": "available", + "uses": ["step_input", "workflow_output"], + }, + ] + assert not any("*" in option["path"] for option in options) + assert not any(option["path"].startswith("input.metadata.") for option in options) + assert options[0] == { + "path": "input.request", + "label": "Request", + "origin": "workflow_input", + "schema": { + "title": "Request", + "description": "The incoming request.", + "type": "object", + "properties": { + "id": {"type": "string"}, + "metadata": {"type": "object", "additionalProperties": True}, + }, + "required": ["id"], + }, + "required": True, + "availability": "available", + "uses": ["step_input", "workflow_output"], + "description": "The incoming request.", + } + + +def test_schema_path_options_resolves_local_definition_metadata() -> None: + options = schema_path_options( + { + "type": "object", + "properties": {"snapshot": {"$ref": "#/$defs/Snapshot"}}, + "required": ["snapshot"], + "$defs": { + "Snapshot": { + "title": "Saved Snapshot", + "type": "object", + "properties": {"version": {"type": "integer"}}, + "required": ["version"], + } + }, + }, + root="input", + uses=["step_input"], + ) + + assert options == [ + { + "path": "input.snapshot", + "label": "Saved Snapshot", + "origin": "workflow_input", + "schema": { + "$ref": "#/$defs/Snapshot", + "$defs": { + "Snapshot": { + "title": "Saved Snapshot", + "type": "object", + "properties": {"version": {"type": "integer"}}, + "required": ["version"], + } + }, + }, + "required": True, + "availability": "available", + "uses": ["step_input"], + }, + { + "path": "input.snapshot.version", + "label": "Version", + "origin": "workflow_input", + "schema": { + "type": "integer", + "$defs": { + "Snapshot": { + "title": "Saved Snapshot", + "type": "object", + "properties": {"version": {"type": "integer"}}, + "required": ["version"], + } + }, + }, + "required": True, + "availability": "available", + "uses": ["step_input"], + }, + ] + + +def test_schema_path_options_returns_empty_schema_for_unconstrained_property() -> None: + options = schema_path_options( + { + "type": "object", + "properties": {"value": {}}, + }, + root="state", + uses=["state_target"], + ) + + assert options == [ + { + "path": "state.value", + "label": "Value", + "origin": "workflow_state", + "schema": {}, + "required": False, + "availability": "available", + "uses": ["state_target"], + } + ] + + +def test_project_authoring_contract_inventory_composes_pure_inputs() -> None: + context_entry = { + "path": "context.loop_item", + "label": "Loop Item", + "origin": "runtime_context", + "schema": {"type": "string"}, + "required": False, + "availability": "conditional", + "uses": ["step_input"], + "reason": "Only available inside the foreach body.", + } + step_input_target = { + "path": "step.input.query", + "label": "Query", + "origin": "step_input", + "schema": {"type": "string"}, + "required": True, + "availability": "available", + "uses": ["step_input"], + } + step_output_source = { + "path": "step.output.answer", + "label": "Answer", + "origin": "step_output", + "schema": {"type": "string"}, + "required": False, + "availability": "available", + "uses": ["step_output_source", "workflow_output"], + } + entry_step = { + "step_id": "fetch", + "label": "Fetch", + } + + inventory = project_authoring_contract_inventory( + workspace_id="workspace-1", + revision=4, + selected_step_id="fetch", + input_schema={ + "type": "object", + "properties": {"request": {"type": "string"}}, + }, + state_schema={ + "type": "object", + "properties": {"answer": {"type": "string"}}, + }, + output_schema={ + "type": "object", + "properties": {"result": {"type": "string"}}, + }, + context_entries=[context_entry], + step_input_targets=[step_input_target], + step_output_sources=[step_output_source], + entry_steps=[entry_step], + workflow_outcomes=["ok", "error"], + warnings=["selected step has conditional context"], + ) + + assert inventory["workspace_id"] == "workspace-1" + assert inventory["revision"] == 4 + assert inventory["selected_step_id"] == "fetch" + assert inventory["readable_sources"] == [ + { + "path": "input.request", + "label": "Request", + "origin": "workflow_input", + "schema": {"type": "string"}, + "required": False, + "availability": "available", + "uses": ["step_input", "workflow_output"], + }, + context_entry, + { + "path": "state.answer", + "label": "Answer", + "origin": "workflow_state", + "schema": {"type": "string"}, + "required": False, + "availability": "available", + "uses": ["step_input", "step_output_source", "workflow_output"], + }, + ] + assert inventory["step_input_targets"] == [step_input_target] + assert inventory["step_output_sources"] == [step_output_source] + assert inventory["state_targets"][0]["path"] == "state.answer" + assert inventory["workflow_output_targets"][0]["path"] == "output.result" + assert inventory["entry_steps"] == [entry_step] + assert inventory["workflow_outcomes"] == ["ok", "error"] + assert inventory["warnings"] == ["selected step has conditional context"]