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

feat: add draft validation repair hints

Browse Source
This commit is contained in:
lda
2026-06-26 19:07:29 +07:00 Verified
parent 8359933852
commit be3e21d9e8
11 changed files with 1025 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_artifacts/draft_workspaces/api.py
+11 -1
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import time
from collections.abc import Callable, Sequence
from typing import Any
from wf_artifacts.drafts import (
@@ -8,12 +9,14 @@ from wf_artifacts.drafts import (
patch_workflow_draft,
validate_workflow_draft,
)
from wf_core.models.schemas import NodeDef
from .models import WorkflowDraftWorkspace, summarize_draft_workspace
from .store import DraftWorkspaceConflictError, DraftWorkspaceStore
JsonObject = dict[str, Any]
JsonPatch = list[dict[str, Any]]
NodeDefsForDraft = Callable[[JsonObject], Sequence[NodeDef]]
def create_draft_workspace(
@@ -56,12 +59,19 @@ def patch_draft_workspace(
workspace_id: str,
revision: int,
patch: JsonPatch,
node_defs: Sequence[NodeDef] | None = None,
node_defs_for_draft: NodeDefsForDraft | None = None,
) -> JsonObject:
"""Apply JSON Patch to a stored workspace when the revision matches."""
workspace = store.get_workspace(workspace_id)
if workspace.revision != revision:
return _revision_conflict_payload(workspace, revision)
patched = patch_workflow_draft(workspace.draft, patch)
patched = patch_workflow_draft(
workspace.draft,
patch,
node_defs=node_defs,
node_defs_for_draft=node_defs_for_draft,
)
if "draft" not in patched:
# A malformed JSON Patch is not a draft revision. Return diagnostics
# without mutating the stored workspace or burning a revision number.
src/wf_artifacts/drafts/api.py
+119 -8
View File
@@ -1,11 +1,17 @@
from __future__ import annotations
from collections.abc import Callable
import re
from collections.abc import Callable, Sequence
from copy import deepcopy
from typing import Any
import jsonpatch
from pydantic import BaseModel, ValidationError
from pydantic import BaseModel, Field, ValidationError
from wf_core.models.schemas import NodeDef
from wf_core.models.steps import OutputBinding
from wf_core.models.workflow import Workflow
from wf_core.validation.issues import ValidationIssue, ValidationIssueCode
from .adapter import build_workflow_from_draft
from .models import WorkflowDraft
@@ -13,6 +19,7 @@ from .models import WorkflowDraft
JsonObject = dict[str, Any]
JsonPatch = list[dict[str, Any]]
OutcomeLookup = Callable[[str], tuple[str, ...] | None]
NodeDefsForDraft = Callable[[JsonObject], Sequence[NodeDef]]
class DraftDiagnostic(BaseModel):
@@ -22,6 +29,8 @@ class DraftDiagnostic(BaseModel):
path: str
step_id: str | None = None
message: str
repair_hint: str | None = None
details: dict[str, Any] = Field(default_factory=dict)
def compile_workflow_draft(draft: JsonObject) -> JsonObject:
@@ -35,16 +44,31 @@ def validate_workflow_draft(
draft: JsonObject,
*,
outcome_lookup: OutcomeLookup | None = None,
node_defs: Sequence[NodeDef] | None = None,
) -> JsonObject:
"""Return structured diagnostics instead of raising on a bad keyed draft."""
"""Return structured diagnostics instead of raising on a bad keyed draft.
When *node_defs* is supplied the structural validator can check
input/output bindings against declared node schemas. Without them
only draft-level parse errors are reported.
"""
try:
compiled_plan = compile_workflow_draft(draft)
parsed = WorkflowDraft.model_validate(draft)
workflow = build_workflow_from_draft(parsed)
except (ValidationError, KeyError, ValueError) as exc:
return _invalid_result(_diagnostic_from_exception(exc))
if node_defs is not None:
workflow = workflow.model_copy(update={"node_defs": list(node_defs)})
diagnostics = _diagnostics_from_workflow_issues(workflow)
if diagnostics:
return _invalid_result(*diagnostics)
if outcome_lookup is not None:
diagnostic = _validate_known_outcomes(draft, outcome_lookup)
if diagnostic is not None:
return _invalid_result(diagnostic)
compiled_plan = workflow.model_dump(
mode="json", by_alias=True, exclude={"node_defs"}
)
return {
"status": "valid",
"diagnostics": [],
@@ -52,7 +76,13 @@ def validate_workflow_draft(
}
def patch_workflow_draft(draft: JsonObject, patch: JsonPatch) -> JsonObject:
def patch_workflow_draft(
draft: JsonObject,
patch: JsonPatch,
*,
node_defs: Sequence[NodeDef] | None = None,
node_defs_for_draft: NodeDefsForDraft | None = None,
) -> JsonObject:
"""Patch the draft source document, then validate the patched result."""
try:
patched = jsonpatch.JsonPatch(patch).apply(deepcopy(draft), in_place=False)
@@ -72,16 +102,21 @@ def patch_workflow_draft(draft: JsonObject, patch: JsonPatch) -> JsonObject:
message="patched draft must be a JSON object",
)
)
result = validate_workflow_draft(patched)
effective_node_defs = (
node_defs_for_draft(patched) if node_defs_for_draft is not None else node_defs
)
result = validate_workflow_draft(patched, node_defs=effective_node_defs)
if result["status"] == "valid":
patched = WorkflowDraft.model_validate(patched).model_dump(mode="json")
return {"draft": patched, **result}
def _invalid_result(diagnostic: DraftDiagnostic) -> JsonObject:
def _invalid_result(*diagnostics: DraftDiagnostic) -> JsonObject:
return {
"status": "invalid",
"diagnostics": [diagnostic.model_dump(mode="json")],
"diagnostics": [
item.model_dump(mode="json", exclude_none=True) for item in diagnostics
],
}
@@ -104,6 +139,82 @@ def _format_location(location: tuple[object, ...]) -> str:
return ".".join(str(part) for part in location)
_NODE_OUTPUT_TARGET_RE = re.compile(
r"^nodes\[(?P<node_index>\d+)\]\.output\[(?P<output_index>\d+)\]\.target$"
)
def _diagnostics_from_workflow_issues(workflow: Workflow) -> list[DraftDiagnostic]:
report = workflow.validate_structure()
return [_diagnostic_from_issue(workflow, issue) for issue in report.errors]
def _diagnostic_from_issue(
workflow: Workflow,
issue: ValidationIssue,
) -> DraftDiagnostic:
step_id = _step_id_for_issue_path(workflow, issue.path)
return DraftDiagnostic(
code=str(issue.code),
path=issue.path,
step_id=step_id,
message=issue.message,
details=_details_for_issue(workflow, issue),
)
def _step_id_for_issue_path(workflow: Workflow, path: str) -> str | None:
match = re.match(r"^nodes\[(?P<node_index>\d+)\]", path)
if match is None:
return None
node_index = int(match.group("node_index"))
if node_index >= len(workflow.nodes):
return None
node_id = getattr(workflow.nodes[node_index], "id", None)
return node_id if isinstance(node_id, str) else None
def _details_for_issue(
workflow: Workflow,
issue: ValidationIssue,
) -> dict[str, Any]:
if issue.code is not ValidationIssueCode.INVALID_DESTINATION_PATH:
return {}
match = _NODE_OUTPUT_TARGET_RE.match(issue.path)
if match is None:
return {}
node_index = int(match.group("node_index"))
output_index = int(match.group("output_index"))
if node_index >= len(workflow.nodes):
return {}
outputs = getattr(workflow.nodes[node_index], "output", None)
if not isinstance(outputs, list) or output_index >= len(outputs):
return {}
binding = outputs[output_index]
if not isinstance(binding, OutputBinding):
return {}
output_field = _single_local_field(binding)
if output_field is None:
return {}
return {
"output_field": output_field,
"state_path": str(binding.target),
}
def _single_local_field(binding: OutputBinding) -> str | None:
from wf_core.local_paths import LocalPathError, split_local_path
try:
parts = split_local_path(binding.source)
except LocalPathError:
return None
if len(parts) != 1:
return None
field = parts[0]
return field if isinstance(field, str) else None
def _validate_known_outcomes(
draft: JsonObject,
outcome_lookup: OutcomeLookup,