refactor: extract NextActions into reusable workflow-surface models
Move wrapper-draft next_actions from handler-local dict helpers into typed models in workflow_surface/next_actions.py. Adds NextActionTool enum, NextActionPatchExample, and NextActions.from_wrapper_hints() constructor. Replaces private _wrapper_draft_next_actions and _wrapper_draft_patch_examples helpers. Adds can_continue as additive advisory field. Keeps backward-compatible aliases in models.py.
This commit is contained in:
@@ -59,6 +59,7 @@ from .constants import (
|
||||
RUNTIME_ERROR_CAPABILITY,
|
||||
)
|
||||
from .models import TraceRange
|
||||
from .next_actions import NextActions
|
||||
from .refs import parse_workflow_surface_capability_id
|
||||
from .saved_subgraphs import (
|
||||
SavedSubgraphTree,
|
||||
@@ -836,11 +837,11 @@ class WorkflowSurfaceHandlers:
|
||||
return {
|
||||
**result,
|
||||
"wrapper_hints": hints,
|
||||
"next_actions": _wrapper_draft_next_actions(
|
||||
"next_actions": NextActions.from_wrapper_hints(
|
||||
workspace_id=workspace_id,
|
||||
revision=int(result["revision"]),
|
||||
hints=hints,
|
||||
),
|
||||
).model_dump(mode="json"),
|
||||
}
|
||||
|
||||
async def create_artifact_from_workspace(
|
||||
@@ -1486,98 +1487,6 @@ def _draft_name_from_capability(capability_name: str) -> str:
|
||||
return capability_name.replace(".", "_").replace("-", "_")
|
||||
|
||||
|
||||
def _wrapper_draft_next_actions(
|
||||
*,
|
||||
workspace_id: str,
|
||||
revision: int,
|
||||
hints: dict[str, Any],
|
||||
) -> dict[str, Any]:
|
||||
"""Convert wrapper_hints into advisory next-tool guidance for MCP clients."""
|
||||
confidence = str(hints.get("confidence", "low"))
|
||||
missing_decisions = hints.get("missing_decisions")
|
||||
notes = [str(note) for note in hints.get("notes", []) if isinstance(note, str)]
|
||||
has_missing = isinstance(missing_decisions, list) and len(missing_decisions) > 0
|
||||
can_save_now = confidence == "high" and not has_missing
|
||||
if can_save_now:
|
||||
return {
|
||||
"can_save_now": True,
|
||||
"recommended_next_tool": "wf.workflow.validate_draft_workspace",
|
||||
"reason": "Wrapper hints are high confidence and have no missing decisions.",
|
||||
"patch_examples": [],
|
||||
"warnings": [],
|
||||
}
|
||||
|
||||
return {
|
||||
"can_save_now": False,
|
||||
"recommended_next_tool": "wf.workflow.patch_draft_workspace",
|
||||
"reason": "Review missing wrapper decisions before saving.",
|
||||
"patch_examples": _wrapper_draft_patch_examples(
|
||||
workspace_id=workspace_id,
|
||||
revision=revision,
|
||||
hints=hints,
|
||||
),
|
||||
"warnings": notes,
|
||||
}
|
||||
|
||||
|
||||
def _wrapper_draft_patch_examples(
|
||||
*,
|
||||
workspace_id: str,
|
||||
revision: int,
|
||||
hints: dict[str, Any],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Return conservative JSON Patch examples without guessing semantics."""
|
||||
examples: list[dict[str, Any]] = []
|
||||
missing_decisions = hints.get("missing_decisions")
|
||||
if not isinstance(missing_decisions, list):
|
||||
return examples
|
||||
decision_kinds = {
|
||||
str(decision.get("kind"))
|
||||
for decision in missing_decisions
|
||||
if isinstance(decision, dict)
|
||||
}
|
||||
if {
|
||||
"choose_output_fields",
|
||||
"review_nested_output",
|
||||
} & decision_kinds:
|
||||
examples.append(
|
||||
{
|
||||
"description": (
|
||||
"Replace output bindings after choosing which capability "
|
||||
"outputs should be written to workflow state."
|
||||
),
|
||||
"tool": "wf.workflow.patch_draft_workspace",
|
||||
"request": {
|
||||
"workspace_id": workspace_id,
|
||||
"revision": revision,
|
||||
"patch": [
|
||||
{
|
||||
"op": "replace",
|
||||
"path": "/draft/steps/call/output",
|
||||
"value": [],
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
if "confirm_boolean_outcomes" in decision_kinds:
|
||||
examples.append(
|
||||
{
|
||||
"description": (
|
||||
"Review boolean output candidates before adding routing; "
|
||||
"do not route on boolean fields automatically."
|
||||
),
|
||||
"tool": "wf.workflow.patch_draft_workspace",
|
||||
"request": {
|
||||
"workspace_id": workspace_id,
|
||||
"revision": revision,
|
||||
"patch": [],
|
||||
},
|
||||
}
|
||||
)
|
||||
return examples
|
||||
|
||||
|
||||
def _source_id_for_capability(
|
||||
sources: dict[str, CapabilitySource],
|
||||
qualified_name: str,
|
||||
|
||||
@@ -4,6 +4,7 @@ from typing import Annotated, Any, Literal
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from .next_actions import NextActionPatchExample, NextActions
|
||||
from wf_artifacts import ArtifactKind
|
||||
from wf_artifacts.draft_workspaces.models import WORKSPACE_ID_PATTERN
|
||||
from wf_core.models.steps import InputBinding, OutputBinding
|
||||
@@ -353,40 +354,10 @@ class CreateDraftWorkspaceFromCapabilityRequest(BaseModel):
|
||||
)
|
||||
|
||||
|
||||
class WrapperDraftPatchExample(BaseModel):
|
||||
"""Concrete patch-workspace example for a likely next authoring edit."""
|
||||
|
||||
description: str = Field(description="Human-readable reason for this patch.")
|
||||
tool: str = Field(description="MCP tool to call for this example.")
|
||||
request: dict[str, Any] = Field(
|
||||
description="JSON request payload to pass to the tool."
|
||||
)
|
||||
|
||||
|
||||
class WrapperDraftNextActions(BaseModel):
|
||||
"""Advisory continuation hints after bootstrapping a wrapper draft."""
|
||||
|
||||
can_save_now: bool = Field(
|
||||
description=(
|
||||
"Advisory only. False means the scaffold likely needs review before "
|
||||
"saving, but the server does not enforce this."
|
||||
)
|
||||
)
|
||||
recommended_next_tool: str = Field(
|
||||
description=(
|
||||
"Suggested next MCP tool, usually wf.workflow.validate_draft_workspace "
|
||||
"or wf.workflow.patch_draft_workspace."
|
||||
)
|
||||
)
|
||||
reason: str = Field(description="Short explanation for the recommendation.")
|
||||
patch_examples: list[WrapperDraftPatchExample] = Field(
|
||||
default_factory=list,
|
||||
description="Concrete JSON Patch examples for common missing decisions.",
|
||||
)
|
||||
warnings: list[str] = Field(
|
||||
default_factory=list,
|
||||
description="Non-blocking warnings copied from low-confidence wrapper hints.",
|
||||
)
|
||||
# Compatibility aliases for older imports. The JSON fields are now generic
|
||||
# workflow-surface guidance, not wrapper-only policy.
|
||||
WrapperDraftPatchExample = NextActionPatchExample
|
||||
WrapperDraftNextActions = NextActions
|
||||
|
||||
|
||||
class CreateDraftWorkspaceFromCapabilityResult(DraftWorkspaceResult):
|
||||
@@ -398,7 +369,7 @@ class CreateDraftWorkspaceFromCapabilityResult(DraftWorkspaceResult):
|
||||
"Use this to patch uncertain maps or schemas by revision."
|
||||
)
|
||||
)
|
||||
next_actions: WrapperDraftNextActions = Field(
|
||||
next_actions: NextActions = Field(
|
||||
description=(
|
||||
"Advisory next step guidance derived from wrapper_hints. "
|
||||
"The server does not enforce can_save_now."
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import StrEnum
|
||||
from typing import Any, Self
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from .wrapper_hints import WrapperAuthoringHints
|
||||
|
||||
|
||||
class NextActionTool(StrEnum):
|
||||
"""Stable MCP workflow tools that guidance may recommend."""
|
||||
|
||||
PATCH_DRAFT_WORKSPACE = "wf.workflow.patch_draft_workspace"
|
||||
VALIDATE_DRAFT_WORKSPACE = "wf.workflow.validate_draft_workspace"
|
||||
VALIDATE_DEPLOYMENT = "wf.workflow.validate_deployment"
|
||||
RUN_DEPLOYMENT = "wf.workflow.run_deployment"
|
||||
RESUME_RUN = "wf.workflow.resume_run"
|
||||
READ_RUN_TRACE = "wf.workflow.read_run_trace"
|
||||
|
||||
|
||||
class NextActionPatchExample(BaseModel):
|
||||
"""Concrete example request for a recommended MCP workflow tool."""
|
||||
|
||||
description: str = Field(description="Human-readable reason for this example.")
|
||||
tool: NextActionTool = Field(description="MCP workflow tool to call.")
|
||||
request: dict[str, Any] = Field(
|
||||
description="JSON request payload to pass to the tool."
|
||||
)
|
||||
|
||||
|
||||
class NextActions(BaseModel):
|
||||
"""Advisory continuation hints for MCP workflow clients.
|
||||
|
||||
This object is guidance, not authority. Validation diagnostics and runtime
|
||||
status remain the source of truth; clients should treat this as a compact
|
||||
answer to "what tool should I call next?"
|
||||
"""
|
||||
|
||||
can_continue: bool = Field(
|
||||
description=(
|
||||
"Whether there is an obvious next workflow-surface tool call. "
|
||||
"Advisory only."
|
||||
)
|
||||
)
|
||||
can_save_now: bool | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Advisory wrapper-authoring signal. False means review is "
|
||||
"recommended before saving; the server does not enforce this."
|
||||
),
|
||||
)
|
||||
recommended_next_tool: NextActionTool | None = Field(
|
||||
default=None,
|
||||
description="Suggested next MCP workflow tool, if one is obvious.",
|
||||
)
|
||||
reason: str = Field(description="Short explanation for the recommendation.")
|
||||
patch_examples: list[NextActionPatchExample] = Field(
|
||||
default_factory=list,
|
||||
description="Concrete JSON Patch examples for common missing decisions.",
|
||||
)
|
||||
warnings: list[str] = Field(
|
||||
default_factory=list,
|
||||
description="Non-blocking warnings copied from low-confidence hints.",
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_wrapper_hints(
|
||||
cls,
|
||||
*,
|
||||
workspace_id: str,
|
||||
revision: int,
|
||||
hints: WrapperAuthoringHints | dict[str, Any],
|
||||
) -> Self:
|
||||
"""Create guidance after bootstrapping a wrapper draft workspace."""
|
||||
payload = (
|
||||
hints.model_dump(mode="json")
|
||||
if isinstance(hints, WrapperAuthoringHints)
|
||||
else hints
|
||||
)
|
||||
confidence = str(payload.get("confidence", "low"))
|
||||
missing_decisions = payload.get("missing_decisions")
|
||||
notes = [
|
||||
str(note) for note in payload.get("notes", []) if isinstance(note, str)
|
||||
]
|
||||
has_missing = isinstance(missing_decisions, list) and len(missing_decisions) > 0
|
||||
can_save_now = confidence == "high" and not has_missing
|
||||
if can_save_now:
|
||||
return cls(
|
||||
can_continue=True,
|
||||
can_save_now=True,
|
||||
recommended_next_tool=NextActionTool.VALIDATE_DRAFT_WORKSPACE,
|
||||
reason="Wrapper hints are high confidence and have no missing decisions.",
|
||||
patch_examples=[],
|
||||
warnings=[],
|
||||
)
|
||||
|
||||
return cls(
|
||||
can_continue=True,
|
||||
can_save_now=False,
|
||||
recommended_next_tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
|
||||
reason="Review missing wrapper decisions before saving.",
|
||||
patch_examples=_wrapper_draft_patch_examples(
|
||||
workspace_id=workspace_id,
|
||||
revision=revision,
|
||||
hints=payload,
|
||||
),
|
||||
warnings=notes,
|
||||
)
|
||||
|
||||
|
||||
def _wrapper_draft_patch_examples(
|
||||
*,
|
||||
workspace_id: str,
|
||||
revision: int,
|
||||
hints: dict[str, Any],
|
||||
) -> list[NextActionPatchExample]:
|
||||
"""Return conservative JSON Patch examples without guessing semantics."""
|
||||
examples: list[NextActionPatchExample] = []
|
||||
missing_decisions = hints.get("missing_decisions")
|
||||
if not isinstance(missing_decisions, list):
|
||||
return examples
|
||||
decision_kinds = {
|
||||
str(decision.get("kind"))
|
||||
for decision in missing_decisions
|
||||
if isinstance(decision, dict)
|
||||
}
|
||||
if {"choose_output_fields", "review_nested_output"} & decision_kinds:
|
||||
examples.append(
|
||||
NextActionPatchExample(
|
||||
description=(
|
||||
"Replace output bindings after choosing which capability "
|
||||
"outputs should be written to workflow state."
|
||||
),
|
||||
tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
|
||||
request={
|
||||
"workspace_id": workspace_id,
|
||||
"revision": revision,
|
||||
"patch": [
|
||||
{
|
||||
"op": "replace",
|
||||
"path": "/draft/steps/call/output",
|
||||
"value": [],
|
||||
}
|
||||
],
|
||||
},
|
||||
)
|
||||
)
|
||||
if "confirm_boolean_outcomes" in decision_kinds:
|
||||
examples.append(
|
||||
NextActionPatchExample(
|
||||
description=(
|
||||
"Review boolean output candidates before adding routing; "
|
||||
"do not route on boolean fields automatically."
|
||||
),
|
||||
tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
|
||||
request={
|
||||
"workspace_id": workspace_id,
|
||||
"revision": revision,
|
||||
"patch": [],
|
||||
},
|
||||
)
|
||||
)
|
||||
return examples
|
||||
Reference in New Issue
Block a user