lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement saved native subgraph resolution through deployments

Browse Source
This commit is contained in:
lda
2026-05-25 18:04:07 +07:00 Verified
parent 6e3a6cac48
commit 1a5efe6cda
12 changed files with 1099 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_mcp/workflow_surface/handlers.py
+23 -27
View File
@@ -9,7 +9,6 @@ from wf_artifacts import (
AvailableCapability,
AvailableSource,
DependencyDiagnostic,
DiagnosticSeverity,
DraftWorkspaceStore,
RequiredCapability,
WorkflowArtifact,
@@ -52,6 +51,11 @@ from .constants import (
)
from .models import TraceRange
from .refs import parse_workflow_surface_capability_id
from .saved_subgraphs import (
interrupting_artifact_diagnostic,
resolve_saved_subgraph_tree,
validate_saved_subgraph_tree,
)
from .wrapper_hints import wrapper_hints_for_capability
if TYPE_CHECKING:
@@ -279,13 +283,13 @@ class WorkflowSurfaceHandlers:
deployment_id: str | None,
) -> dict[str, Any]:
"""Execute a saved wrapper artifact through the workflow runner."""
unsupported = _unsupported_interrupt_diagnostic(artifact)
unsupported = interrupting_artifact_diagnostic(artifact)
if unsupported is not None:
raise ValueError(unsupported.message)
# For now only wrapper artifacts are honest node capabilities here.
# Full saved workflows stay on `run_deployment` until core supports
# graph-as-node semantics instead of us faking subgraphs at this layer.
# Direct capability calls remain wrapper-only. Full saved workflows run
# through deployments, where native subgraph dependencies and bindings
# are prepared before core execution.
plan = _raw_plan_from_artifact(artifact)
deployment = None
if deployment_id is not None:
@@ -882,7 +886,7 @@ class WorkflowSurfaceHandlers:
diagnostics=diagnostics,
)
unsupported = _unsupported_interrupt_diagnostic(artifact)
unsupported = interrupting_artifact_diagnostic(artifact)
if unsupported is not None:
return _run_payload(
deployment=deployment,
@@ -933,10 +937,22 @@ class WorkflowSurfaceHandlers:
deployment.artifact_id,
deployment.artifact_version,
)
available_sources = _available_sources(self.service)
diagnostics = validate_deployment_dependencies(
artifact=artifact,
deployment=deployment,
sources=_available_sources(self.service),
sources=available_sources,
)
tree = resolve_saved_subgraph_tree(
root_artifact=artifact,
artifact_store=self.service.artifact_store,
)
diagnostics.extend(
validate_saved_subgraph_tree(
tree=tree,
deployment=deployment,
sources=available_sources,
)
)
return deployment, artifact, diagnostics
@@ -1212,26 +1228,6 @@ def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
) from exc
def _unsupported_interrupt_diagnostic(
artifact: WorkflowArtifact,
) -> DependencyDiagnostic | None:
if not any(node.get("type") == "interrupt" for node in _plan_nodes(artifact)):
return None
return DependencyDiagnostic(
severity=DiagnosticSeverity.ERROR,
code="interrupting_artifact_unsupported",
logical_ref=f"workflow.{artifact.id}.v{artifact.version}",
message=(
"Running saved workflow artifacts with interrupt nodes is unsupported "
"until nested run-state resume is implemented."
),
repair_hint=(
"Run this workflow as a top-level core workflow or remove interrupt "
"nodes before saving it as a runnable deployment."
),
)
def _plan_nodes(artifact: WorkflowArtifact) -> list[dict[str, Any]]:
nodes = artifact.plan.get("nodes", [])
return [node for node in nodes if isinstance(node, dict)]
src/wf_mcp/workflow_surface/saved_subgraphs.py
+219
View File
@@ -0,0 +1,219 @@
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from pydantic import TypeAdapter
from wf_artifacts import (
AvailableSource,
DependencyDiagnostic,
DiagnosticSeverity,
WorkflowArtifact,
WorkflowArtifactStore,
WorkflowDeployment,
validate_deployment_dependencies,
)
from wf_core import (
AsyncNodeHandler,
InterruptNode,
NodeUse,
PreparedSubgraph,
SubgraphNode,
Workflow,
)
from wf_core.models.steps import Step
from wf_core.models.workflow_refs import WorkflowRef
from wf_platform import CapabilitySource
from ..models import RawWorkflowPlan
from .runtime_dependencies import resolve_runtime_dependencies
_STEPS_ADAPTER = TypeAdapter(list[Step])
@dataclass(frozen=True, slots=True)
class SavedSubgraphTree:
"""Saved descendant artifacts prepared from one root artifact boundary."""
artifacts_by_ref: dict[str, WorkflowArtifact]
diagnostics: list[DependencyDiagnostic]
def resolve_saved_subgraph_tree(
*,
root_artifact: WorkflowArtifact,
artifact_store: WorkflowArtifactStore,
) -> SavedSubgraphTree:
"""Load exact saved descendants and report missing refs or recursion cycles.
A saved subgraph ref identifies an immutable artifact version. This loader
intentionally does not resolve capabilities or deployment bindings; it
identifies the artifact tree that later platform validation/preparation
will operate on.
"""
artifacts_by_ref: dict[str, WorkflowArtifact] = {}
diagnostics: list[DependencyDiagnostic] = []
_visit_saved_children(
artifact=root_artifact,
artifact_store=artifact_store,
active={(root_artifact.id, root_artifact.version)},
artifacts_by_ref=artifacts_by_ref,
diagnostics=diagnostics,
)
return SavedSubgraphTree(
artifacts_by_ref=artifacts_by_ref,
diagnostics=diagnostics,
)
def validate_saved_subgraph_tree(
*,
tree: SavedSubgraphTree,
deployment: WorkflowDeployment,
sources: list[AvailableSource],
) -> list[DependencyDiagnostic]:
"""Validate saved descendants under the parent deployment environment."""
diagnostics = list(tree.diagnostics)
for child in tree.artifacts_by_ref.values():
diagnostics.extend(
validate_deployment_dependencies(
artifact=child,
deployment=deployment,
sources=sources,
)
)
interrupt_diagnostic = interrupting_artifact_diagnostic(child)
if interrupt_diagnostic is not None:
diagnostics.append(interrupt_diagnostic)
return diagnostics
def prepare_saved_subgraphs(
*,
tree: SavedSubgraphTree,
deployment: WorkflowDeployment | None,
sources: dict[str, CapabilitySource],
compile_plan: Callable[[RawWorkflowPlan, dict[str, str] | None], Workflow],
) -> dict[str, PreparedSubgraph[AsyncNodeHandler]]:
"""Compile saved descendants using one inherited deployment environment.
The tree has already fixed exact artifact versions. Binding resolution is
deliberately shared with the root deployment; per-use-site deployment
overrides are a future platform feature rather than an implicit fallback.
"""
if tree.diagnostics:
messages = "; ".join(diagnostic.message for diagnostic in tree.diagnostics)
raise ValueError(f"cannot prepare invalid saved subgraph tree: {messages}")
prepared: dict[str, PreparedSubgraph[AsyncNodeHandler]] = {}
for ref_display, child in tree.artifacts_by_ref.items():
plan = RawWorkflowPlan.model_validate(child.plan)
dependencies = resolve_runtime_dependencies(
artifact=child,
deployment=deployment,
sources=sources,
plan_node_names=[
node.node for node in plan.nodes if isinstance(node, NodeUse)
],
)
prepared[ref_display] = PreparedSubgraph(
workflow=compile_plan(plan, dependencies.node_name_bindings),
registry=dependencies.node_registry,
reducers=dependencies.reducers,
)
return prepared
def interrupting_artifact_diagnostic(
artifact: WorkflowArtifact,
) -> DependencyDiagnostic | None:
"""Reject saved interrupt workflows until the platform exposes resume."""
if not any(isinstance(node, InterruptNode) for node in _artifact_steps(artifact)):
return None
return DependencyDiagnostic(
severity=DiagnosticSeverity.ERROR,
code="interrupting_artifact_unsupported",
logical_ref=f"workflow.{artifact.id}.v{artifact.version}",
message=(
"Running saved workflow artifacts with interrupt nodes is unsupported "
"until nested run-state resume is implemented."
),
repair_hint=(
"Run this workflow as a top-level core workflow or remove interrupt "
"nodes before saving it as a runnable deployment."
),
)
def _visit_saved_children(
*,
artifact: WorkflowArtifact,
artifact_store: WorkflowArtifactStore,
active: set[tuple[str, int]],
artifacts_by_ref: dict[str, WorkflowArtifact],
diagnostics: list[DependencyDiagnostic],
) -> None:
for ref in _saved_child_refs(artifact):
identity = _saved_identity(ref)
if identity in active:
diagnostics.append(_cycle_diagnostic(ref))
continue
if ref.display in artifacts_by_ref:
continue
try:
child = artifact_store.get_artifact(*identity)
except KeyError:
diagnostics.append(_missing_diagnostic(ref))
continue
artifacts_by_ref[ref.display] = child
_visit_saved_children(
artifact=child,
artifact_store=artifact_store,
active=active | {identity},
artifacts_by_ref=artifacts_by_ref,
diagnostics=diagnostics,
)
def _saved_child_refs(artifact: WorkflowArtifact) -> list[WorkflowRef]:
return [
node.workflow
for node in _artifact_steps(artifact)
if isinstance(node, SubgraphNode) and node.workflow.artifact_id is not None
]
def _artifact_steps(artifact: WorkflowArtifact) -> list[Step]:
"""Validate only step payloads needed for dependency discovery."""
raw_nodes = artifact.plan.get("nodes", [])
return _STEPS_ADAPTER.validate_python(raw_nodes)
def _saved_identity(ref: WorkflowRef) -> tuple[str, int]:
"""Return the required saved-ref fields after structural model validation."""
if ref.artifact_id is None or ref.version is None:
raise ValueError(f"workflow ref {ref.display!r} is not a saved artifact ref")
return ref.artifact_id, ref.version
def _missing_diagnostic(ref: WorkflowRef) -> DependencyDiagnostic:
return DependencyDiagnostic(
severity=DiagnosticSeverity.ERROR,
code="workflow_dependency_missing",
logical_ref=ref.display,
message=f"Saved child workflow {ref.display!r} is unavailable.",
repair_hint=(
"Save the referenced artifact version or update the parent graph."
),
)
def _cycle_diagnostic(ref: WorkflowRef) -> DependencyDiagnostic:
return DependencyDiagnostic(
severity=DiagnosticSeverity.ERROR,
code="workflow_dependency_cycle",
logical_ref=ref.display,
message=f"Saved child workflow {ref.display!r} creates a dependency cycle.",
repair_hint="Remove the recursive saved subgraph reference.",
)