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

fourth slice: run lifecycle moves

Browse Source
This commit is contained in:
lda
2026-06-02 01:30:01 +07:00 Verified
parent c46c636694
commit 5f8c4224f0
7 changed files with 1132 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_mcp/broker/service/workflow_operation_context.py
+35 -4
View File
@@ -70,11 +70,42 @@ class WfMcpWorkflowRuntimeRunner(WorkflowRuntimeRunner):
service: WfMcpService
async def run_workflow_from_plan(self, plan, **kwargs):
return await self.service.run_workflow_from_plan(plan, **kwargs)
async def run_workflow_from_plan(
self,
plan,
workflow_input,
deployment=None,
artifact=None,
saved_subgraph_tree=None,
):
return await self.service.run_workflow_from_plan(
plan,
workflow_input,
deployment=deployment,
artifact=artifact,
saved_subgraph_tree=saved_subgraph_tree,
)
async def resume_workflow_from_plan(self, plan, **kwargs):
return await self.service.resume_workflow_from_plan(plan, **kwargs)
async def resume_workflow_from_plan(
self,
plan,
run,
*,
resume_payload,
resume_outcome,
deployment=None,
artifact=None,
saved_subgraph_tree=None,
):
return await self.service.resume_workflow_from_plan(
plan,
run,
resume_payload=resume_payload,
resume_outcome=resume_outcome,
deployment=deployment,
artifact=artifact,
saved_subgraph_tree=saved_subgraph_tree,
)
@dataclass(frozen=True, slots=True)
src/wf_mcp/workflow_surface/handlers.py
+13 -279
View File
@@ -1,25 +1,19 @@
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import asdict
from typing import TYPE_CHECKING, Any
from wf_artifacts import (
ArtifactKind,
AvailableCapability,
AvailableSource,
DependencyDiagnostic,
DiagnosticSeverity,
DraftWorkspaceStore,
RequiredCapability,
RunStore,
WorkflowArtifact,
WorkflowCapabilityRef,
WorkflowDeployment,
)
from wf_platform import (
CapabilitySource,
hash_json_schema,
)
from wf_authoring import build_async_registry
from wf_core import RuntimeContext
@@ -35,9 +29,9 @@ from wf_api.drafts import WorkflowDraftApi
from wf_api.models import RawWorkflowPlan
from wf_api.next_actions import NextActions
from wf_api.refs import parse_workflow_surface_capability_id
from wf_api.runs import WorkflowRunApi
from wf_api.saved_subgraphs import (
direct_wrapper_interrupt_diagnostic,
saved_subgraph_tree_from_snapshots,
)
from wf_api.wrapper_hints import (
workflow_output_schema_for_authoring,
@@ -47,19 +41,8 @@ from wf_api.wrapper_hints import (
from ..broker.service.workflow_operation_context import context_from_service
from ..shared import matches_query, paged_list_payload
from .models import TraceRange
from wf_api.run_lifecycle import (
create_pinned_environment,
has_blocking_diagnostics,
load_stored_run,
mark_resume_blocked,
persist_stopped_run,
restore_interrupted_run,
validate_pinned_resume_environment,
)
if TYPE_CHECKING:
from wf_core import RunState
from ..broker.service import WfMcpService
@@ -72,6 +55,7 @@ class WorkflowSurfaceHandlers:
self._drafts = WorkflowDraftApi(context)
self._artifacts = WorkflowArtifactApi(context)
self._deployments = WorkflowDeploymentApi(context)
self._runs = WorkflowRunApi(context)
async def list_artifacts(
self,
@@ -743,61 +727,10 @@ class WorkflowSurfaceHandlers:
workflow_input: dict[str, Any],
trace_range: TraceRange | None = None,
) -> dict[str, Any]:
deployment, artifact, diagnostics, tree = self._deployments.deployment_validation(
deployment_id
)
if diagnostics:
return _run_payload(
deployment=deployment,
artifact=artifact,
status="unrunnable",
diagnostics=diagnostics,
)
plan = _raw_plan_from_artifact(artifact)
run = await self.service.run_workflow_from_plan(
plan,
workflow_input,
deployment=deployment,
artifact=artifact,
saved_subgraph_tree=tree,
)
record = persist_stopped_run(
store=self._run_store(),
environment=create_pinned_environment(
deployment=deployment,
artifact=artifact,
tree=tree,
),
run=run,
)
return _run_payload(
deployment=deployment,
artifact=artifact,
status=run.status.value,
run_id=record.id,
resume_readiness=record.resume_readiness.value,
interrupt=_interrupt_payload(run),
outcome=run.outcome,
error=run.error,
output=run.output,
trace_count=len(run.trace),
trace=(
[
asdict(entry)
for entry in run.trace[
trace_range.start : trace_range.start + trace_range.limit
]
]
if trace_range is not None
else None
),
trace_start=trace_range.start if trace_range is not None else None,
trace_limit=trace_range.limit if trace_range is not None else None,
trace_truncated=(
trace_range is not None
and len(run.trace) > trace_range.start + trace_range.limit
),
return await self._runs.run_deployment(
deployment_id=deployment_id,
workflow_input=workflow_input,
trace_range=trace_range,
)
async def resume_run(
@@ -809,94 +742,16 @@ class WorkflowSurfaceHandlers:
trace_range: TraceRange | None = None,
) -> dict[str, Any]:
"""Resume one durable interrupted deployment run."""
record, stopped_run = restore_interrupted_run(self._run_store(), run_id)
environment = record.environment
diagnostics = validate_pinned_resume_environment(
record=record,
sources=_available_sources(self.service),
)
if has_blocking_diagnostics(diagnostics):
blocked = mark_resume_blocked(
store=self._run_store(),
record=record,
diagnostics=diagnostics,
)
return _run_payload(
deployment=environment.deployment,
artifact=environment.root_artifact,
status=stopped_run.status.value,
run_id=blocked.id,
resume_readiness=blocked.resume_readiness.value,
interrupt=_interrupt_payload(stopped_run),
outcome=stopped_run.outcome,
error=stopped_run.error,
output=stopped_run.output,
diagnostics=diagnostics,
trace_count=len(stopped_run.trace),
)
plan = _raw_plan_from_artifact(environment.root_artifact)
tree = saved_subgraph_tree_from_snapshots(environment.child_artifacts)
run = await self.service.resume_workflow_from_plan(
plan,
stopped_run,
return await self._runs.resume_run(
run_id=run_id,
resume_payload=resume_payload,
resume_outcome=resume_outcome,
deployment=environment.deployment,
artifact=environment.root_artifact,
saved_subgraph_tree=tree,
)
next_record = persist_stopped_run(
store=self._run_store(),
environment=environment,
run=run,
run_id=run_id,
)
return _run_payload(
deployment=environment.deployment,
artifact=environment.root_artifact,
status=run.status.value,
run_id=next_record.id,
resume_readiness=next_record.resume_readiness.value,
interrupt=_interrupt_payload(run),
outcome=run.outcome,
error=run.error,
output=run.output,
trace_count=len(run.trace),
trace=(
[
asdict(entry)
for entry in run.trace[
trace_range.start : trace_range.start + trace_range.limit
]
]
if trace_range is not None
else None
),
trace_start=trace_range.start if trace_range is not None else None,
trace_limit=trace_range.limit if trace_range is not None else None,
trace_truncated=(
trace_range is not None
and len(run.trace) > trace_range.start + trace_range.limit
),
trace_range=trace_range,
)
async def inspect_run(self, *, run_id: str) -> dict[str, Any]:
"""Return one durable stopped-run summary without debug trace entries."""
record, run = load_stored_run(self._run_store(), run_id)
environment = record.environment
return _run_payload(
deployment=environment.deployment,
artifact=environment.root_artifact,
status=record.status.value,
run_id=record.id,
resume_readiness=record.resume_readiness.value,
interrupt=_interrupt_payload(run),
outcome=run.outcome,
error=run.error,
output=run.output,
diagnostics=record.diagnostics,
trace_count=len(run.trace),
)
return await self._runs.inspect_run(run_id=run_id)
async def read_run_trace(
self,
@@ -905,69 +760,11 @@ class WorkflowSurfaceHandlers:
trace_range: TraceRange,
) -> dict[str, Any]:
"""Return only a caller-bounded debug trace slice from a stopped run."""
record, run = load_stored_run(self._run_store(), run_id)
environment = record.environment
end = trace_range.start + trace_range.limit
return _run_payload(
deployment=environment.deployment,
artifact=environment.root_artifact,
status=record.status.value,
run_id=record.id,
resume_readiness=record.resume_readiness.value,
diagnostics=record.diagnostics,
trace_count=len(run.trace),
trace=[asdict(entry) for entry in run.trace[trace_range.start : end]],
trace_start=trace_range.start,
trace_limit=trace_range.limit,
trace_truncated=len(run.trace) > end,
return await self._runs.read_run_trace(
run_id=run_id,
trace_range=trace_range,
)
def _run_store(self) -> RunStore:
"""Return the configured durable run store required by workflow runs."""
if self.service.run_store is None:
raise KeyError("workflow run store is not configured")
return self.service.run_store
def _available_sources(service: WfMcpService) -> list[AvailableSource]:
"""Convert broker capability sources into artifact validation snapshots."""
sources: list[AvailableSource] = []
for source in service.capability_sources.values():
node_spec_details = {
detail.name: detail
for detail in source.as_inventory().capabilities.node_spec_details
}
capabilities = {
capability_name: AvailableCapability(
name=capability_name,
kind="node_spec",
input_schema_hash=hash_json_schema(detail.input_schema),
output_schema_hash=hash_json_schema(detail.output_schema),
)
for spec in source.capabilities.node_specs.values()
if (capability_name := _capability_name(spec.name)) is not None
if (detail := node_spec_details.get(spec.name)) is not None
}
capabilities.update(
{
capability_name: AvailableCapability(
name=capability_name,
kind="reducer",
)
for reducer in source.capabilities.reducers.values()
if (capability_name := _capability_name(reducer.name)) is not None
}
)
sources.append(
AvailableSource(
id=source.id,
enabled=source.enabled,
capabilities=capabilities,
)
)
return sources
def _required_capability_payloads(
requirements: dict[str, RequiredCapability],
) -> dict[str, dict[str, Any]]:
@@ -1047,66 +844,3 @@ def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
f"workflow artifact {artifact.id}@{artifact.version} "
f"is missing plan field {field_name!r}"
) from exc
def _run_payload(
*,
deployment: WorkflowDeployment,
artifact: WorkflowArtifact,
status: str,
run_id: str | None = None,
resume_readiness: str | None = None,
interrupt: dict[str, Any] | None = None,
outcome: str | None = None,
error: str | None = None,
diagnostics: list[DependencyDiagnostic] | None = None,
output: dict[str, Any] | None = None,
trace_count: int = 0,
trace: list[dict[str, Any]] | None = None,
trace_start: int | None = None,
trace_limit: int | None = None,
trace_truncated: bool = False,
) -> dict[str, Any]:
payload = {
"deployment_id": deployment.id,
"artifact_id": artifact.id,
"artifact_version": artifact.version,
"status": status,
"run_id": run_id,
"resume_readiness": resume_readiness,
"interrupt": interrupt,
"outcome": outcome,
"error": error,
"output": output,
"diagnostics": [
diagnostic.model_dump(mode="json") for diagnostic in diagnostics or []
],
"trace_count": trace_count,
"next_actions": NextActions.from_run_result(
run_id=run_id,
status=status,
trace_count=trace_count,
diagnostics=diagnostics or [],
).model_dump(mode="json"),
}
if trace is not None:
# Trace entries can grow quickly, so the public run tool only includes
# a bounded debug slice when the caller explicitly asks for a range.
payload["trace_start"] = trace_start
payload["trace_limit"] = trace_limit
payload["trace"] = trace
payload["trace_truncated"] = trace_truncated
return payload
def _interrupt_payload(run: RunState) -> dict[str, Any] | None:
"""Return a JSON-safe interrupt payload for the current run, if paused."""
if run.interrupt is None:
return None
payload = asdict(run.interrupt)
route = payload.get("route")
if isinstance(route, dict) and "workflow_ref" in route:
workflow_ref = route["workflow_ref"]
if hasattr(workflow_ref, "model_dump"):
route["workflow_ref"] = workflow_ref.model_dump(mode="json")
return payload