Re: [PR] [ISSUE #22]Feature Request: Upgrade to Serverless Workflow DSL 1.0.3 [eventmesh-workflow]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #23:
URL: https://github.com/apache/eventmesh-workflow/pull/23#discussion_r3506569270


##########
third_party/swf/swf.go:
##########
@@ -16,14 +16,416 @@
 package swf
 
 import (
-	"github.com/gogf/gf/util/gconv"
-	"github.com/serverlessworkflow/sdk-go/v2/model"
-	"github.com/serverlessworkflow/sdk-go/v2/parser"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"gopkg.in/yaml.v3"
 )
 
-func Parse(source string) (*model.Workflow, error) {
-	if len(source) == 0 {
+func Parse(source string) (*Workflow, error) {
+	if len(strings.TrimSpace(source)) == 0 {
+		return nil, nil
+	}
+	var raw map[string]interface{}
+	if err := yaml.Unmarshal([]byte(source), &raw); err != nil {
+		return nil, err
+	}
+	var wf *Workflow
+	var err error
+	if _, ok := raw["document"]; ok {
+		wf, err = parseV1Workflow(raw)
+	} else {
+		wf, err = parseLegacyWorkflow(raw)
+	}
+	if err != nil {
+		return nil, err
+	}
+	if wf == nil {
 		return nil, nil
 	}
-	return parser.FromYAMLSource(gconv.Bytes(source))
+	if verr := wf.Validate(); verr != nil {
+		return nil, verr
+	}
+	return wf, nil
+}
+
+func parseV1Workflow(raw map[string]interface{}) (*Workflow, error) {
+	doc := asMap(raw["document"])
+	name := asString(doc["name"])
+	if name == "" {
+		return nil, errors.New("workflow document.name is required")
+	}
+	version := asString(doc["version"])
+	if version == "" {
+		return nil, errors.New("workflow document.version is required")
+	}
+	dsl := asString(doc["dsl"])
+	if dsl == "" {
+		return nil, errors.New("workflow document.dsl is required")
+	}
+	tasks, err := parseV1TaskList(raw["do"])
+	if err != nil {
+		return nil, err
+	}
+	if len(tasks) == 0 {
+		return nil, errors.New("workflow do must contain at least one task")
+	}
+	return &Workflow{
+		ID:        name,
+		Name:      name,
+		Version:   version,
+		DSL:       dsl,
+		Namespace: asString(doc["namespace"]),
+		Start:     tasks[0].Name,
+		Tasks:     tasks,
+		Functions: parseReusableFunctions(raw),
+		Schedule:  parseSchedule(raw["schedule"]),
+		Input:     parseWorkflowInput(raw["input"]),
+		Output:    expressionString(asMap(raw["output"])["as"]),
+		Legacy:    false,
+	}, nil
+}
+
+func parseLegacyWorkflow(raw map[string]interface{}) (*Workflow, error) {
+	id := asString(raw["id"])
+	if id == "" {
+		return nil, errors.New("workflow id is required")
+	}
+	states := asSlice(raw["states"])
+	if len(states) == 0 {
+		return nil, errors.New("workflow states must contain at least one state")
+	}
+	functions := make(map[string]*Function)
+	for _, item := range asSlice(raw["functions"]) {
+		fnMap := asMap(item)
+		name := asString(fnMap["name"])
+		if name == "" {
+			continue
+		}
+		functions[name] = &Function{Name: name, Operation: asString(fnMap["operation"]), Type: asString(fnMap["type"])}
+	}
+	var tasks []*Task
+	for _, item := range states {
+		stateMap := asMap(item)
+		name := asString(stateMap["name"])
+		if name == "" {
+			continue
+		}
+		task := &Task{Name: name, Type: asString(stateMap["type"]), Then: legacyThen(stateMap), ExplicitThen: true}
+		if filter := asMap(stateMap["stateDataFilter"]); len(filter) > 0 {
+			task.InputFilter = asString(filter["input"])
+		}
+		task.Actions = parseLegacyActions(stateMap, functions)
+		if task.Type == TaskTypeSwitch {
+			task.Cases = parseLegacySwitchCases(stateMap)
+		}
+		tasks = append(tasks, task)
+	}
+	start := asString(raw["start"])
+	if start == "" && len(tasks) > 0 {
+		start = tasks[0].Name
+	}
+	return &Workflow{
+		ID:        id,
+		Name:      asString(raw["name"]),
+		Version:   asString(raw["version"]),
+		DSL:       asString(raw["specVersion"]),
+		Start:     start,
+		Tasks:     tasks,
+		Functions: functions,
+		Legacy:    true,
+	}, nil
+}
+
+func parseSchedule(value interface{}) *Schedule {
+	scheduleMap := asMap(value)
+	if len(scheduleMap) == 0 {
+		return nil
+	}
+	return &Schedule{
+		Start: asString(scheduleMap["start"]),
+		Cron:  asString(scheduleMap["cron"]),
+		After: asString(scheduleMap["after"]),
+	}
+}
+
+func parseWorkflowInput(value interface{}) string {
+	inputMap := asMap(value)
+	if len(inputMap) == 0 {
+		return ""
+	}
+	return expressionString(inputMap["from"])
+}
+
+func parseV1TaskList(value interface{}) ([]*Task, error) {
+	items := asSlice(value)
+	if len(items) == 0 {
+		return nil, nil
+	}
+	var tasks []*Task
+	for _, item := range items {
+		itemMap := asMap(item)
+		for taskName, taskDef := range itemMap {
+			task, err := parseV1Task(taskName, asMap(taskDef))
+			if err != nil {
+				return nil, err
+			}
+			tasks = append(tasks, task)
+		}
+	}
+	return tasks, nil
+}
+
+func parseV1Task(taskName string, def map[string]interface{}) (*Task, error) {
+	if taskName == "" {
+		return nil, errors.New("task name is required")
+	}
+	task := &Task{Name: taskName, Type: detectV1TaskType(def), Raw: mustJSON(def)}
+	if input := asMap(def["input"]); len(input) > 0 {
+		task.InputFilter = expressionString(input["from"])
+	}
+	if data := asString(def["data"]); data != "" {
+		task.InlineData = data
+	}
+	if output := asMap(def["output"]); len(output) > 0 {
+		task.OutputFilter = expressionString(output["as"])
+	}
+	if _, ok := def["then"]; ok {
+		task.Then = expressionString(def["then"])
+		task.ExplicitThen = true
+	}
+	switch task.Type {
+	case TaskTypeOperation:
+		task.Actions = []*Action{parseV1CallAction(def)}
+	case TaskTypeEvent:
+		task.Actions = []*Action{parseV1CallAction(def)}
+	case TaskTypeSet:
+		task.Actions = []*Action{{OperationName: taskName, OperationType: TaskTypeSet}}
+	case TaskTypeSwitch:
+		task.Cases = parseV1SwitchCases(def["switch"])
+	case TaskTypeDo:
+		children, err := parseV1TaskList(def["do"])
+		if err != nil {
+			return nil, err
+		}
+		task.Children = children
+	case TaskTypeFork:
+		task.Children = parseV1ForkBranches(def["fork"])
+	case TaskTypeFor:
+		children, err := parseV1TaskList(asMap(def["for"])["do"])
+		if err != nil {
+			return nil, err
+		}
+		task.Children = children
+	case TaskTypeTry:
+		children, err := parseV1TaskList(def["try"])
+		if err != nil {
+			return nil, err
+		}
+		task.Children = children
+		if catchMap := asMap(def["catch"]); len(catchMap) > 0 {
+			if when := catchMap["when"]; when != nil {
+				catchTasks, err := parseV1TaskList(when)
+				if err == nil {
+					task.Children = append(task.Children, catchTasks...)
+				}
+			}
+		}
+	case TaskTypeEmit, TaskTypeListen, TaskTypeWait, TaskTypeRaise, TaskTypeRun:
+		task.Actions = []*Action{{OperationName: taskName, OperationType: task.Type}}
+	}
+	return task, nil
+}
+
+func detectV1TaskType(def map[string]interface{}) string {
+	if _, ok := def["call"]; ok {
+		return TaskTypeOperation
+	}
+	for _, key := range []string{TaskTypeSwitch, TaskTypeSet, TaskTypeDo, TaskTypeFork, TaskTypeFor, TaskTypeTry, TaskTypeWait, TaskTypeRaise, TaskTypeRun, TaskTypeEmit, TaskTypeListen} {
+		if _, ok := def[key]; ok {
+			return key
+		}
+	}
+	return TaskTypeOperation
+}
+
+func parseV1CallAction(def map[string]interface{}) *Action {
+	operationType := expressionString(def["call"])
+	operationName := operationType
+	with := asMap(def["with"])
+	switch operationType {
+	case "http":
+		operationName = expressionString(with["endpoint"])
+	case "openapi":
+		operationName = firstNonEmpty(expressionString(with["operationId"]), expressionString(with["operation"]), expressionString(with["document"]))
+	case "asyncapi":
+		operationName = firstNonEmpty(expressionString(with["operation"]), expressionString(with["channel"]), expressionString(with["document"]))
+	case "grpc":
+		operationName = firstNonEmpty(expressionString(with["service"]), expressionString(with["method"]))
+	default:
+		operationName = firstNonEmpty(operationName, expressionString(with["operation"]), expressionString(with["endpoint"]))
+	}

Review Comment:
   `parseV1CallAction` falls into the `default` branch for `call: a2a`, and `firstNonEmpty(operationName, ..., endpoint)` will always keep `operationName` ("a2a") even when `with.endpoint` is provided. This makes the stored OperationName unusable as an agent URL (and prevents A2A execution from working).



##########
internal/task/switch_task.go:
##########
@@ -41,50 +42,70 @@ func NewSwitchTask(instance *model.WorkflowTaskInstance) Task {
 	if instance == nil || instance.Task == nil {
 		return nil
 	}
-	t.baseTask = baseTask{taskID: instance.TaskID, input: instance.Input,
-		workflowID: instance.WorkflowID, workflowInstanceID: instance.WorkflowInstanceID,
-		taskType: instance.Task.TaskType}
+	t.baseTask = newBaseTask(instance)
 	t.transitions = instance.Task.ChildTasks
-	t.baseTask.queue = queue.GetQueue(config.GlobalConfig().Flow.Queue.Store)
+	t.queue = queue.GetQueue(config.GlobalConfig().Flow.Queue.Store)
 	t.workflowDAL = dal.NewWorkflowDAL()
 	t.jq = jqer.NewJQ()
 	return &t
 }
 
 func (t *switchTask) Run() error {
-	metrics.Inc(constants.MetricsSwitchTask, constants.MetricsTotal)
+	_ = metrics.Inc(constants.MetricsSwitchTask, constants.MetricsTotal)
 	if len(t.transitions) == 0 {
 		return nil
 	}
+	var defaultTransition *model.WorkflowTaskRelation
 	for _, transition := range t.transitions {
-		if transition.ToTaskID == constants.TaskEndID {
+		if transition == nil {
 			continue
 		}
-		var jqData interface{}
-		err := json.Unmarshal([]byte(t.input), &jqData)
-		if err != nil {
-			return err
-		}
-		res, err := t.jq.One(jqData, transition.Condition)
-		if err != nil {
-			return err
+		if transition.Condition == "" {
+			defaultTransition = transition
+			continue
 		}
-		boolValue, err := strconv.ParseBool(gconv.String(res))
+		matched, err := t.matchCondition(transition.Condition)
 		if err != nil {
 			return err
 		}
-		if !boolValue {
-			metrics.Inc(constants.MetricsSwitchTask, constants.MetricsSwitchReject)
+		if !matched {
+			_ = metrics.Inc(constants.MetricsSwitchTask, constants.MetricsSwitchReject)
 			continue
 		}
+		_ = metrics.Inc(constants.MetricsSwitchTask, constants.MetricsSwitchPass)
+		return t.publishOrComplete(transition)
+	}
+	if defaultTransition != nil {
+		_ = metrics.Inc(constants.MetricsSwitchTask, constants.MetricsSwitchPass)
+		return t.publishOrComplete(defaultTransition)
+	}
+	return t.completeWorkflow()
+}
 
-		metrics.Inc(constants.MetricsSwitchTask, constants.MetricsSwitchPass)
-		var taskInstance = model.WorkflowTaskInstance{WorkflowInstanceID: t.workflowInstanceID,
-			WorkflowID: t.workflowID, TaskID: transition.ToTaskID, TaskInstanceID: uuid.New().String(),
-			Status: constants.TaskInstanceWaitStatus, Input: t.baseTask.input}
-		return t.baseTask.queue.Publish([]*model.WorkflowTaskInstance{&taskInstance})
+func (t *switchTask) matchCondition(condition string) (bool, error) {
+	var jqData interface{}
+	err := json.Unmarshal([]byte(t.input), &jqData)
+	if err != nil {
+		return false, err
 	}
-	// not match
+	res, err := t.jq.One(jqData, condition)
+	if err != nil {
+		return false, err
+	}
+	return strconv.ParseBool(gconv.String(res))

Review Comment:
   `matchCondition` passes bare expressions (e.g. `.status == "ok"`) directly to `jq.One`, but `jqer` only executes expressions wrapped as `${ ... }`. This will make v1 `when:` conditions fail (and `ParseBool` will error) unless the DSL author manually wraps every condition.



##########
third_party/swf/swf.go:
##########
@@ -16,14 +16,416 @@
 package swf
 
 import (
-	"github.com/gogf/gf/util/gconv"
-	"github.com/serverlessworkflow/sdk-go/v2/model"
-	"github.com/serverlessworkflow/sdk-go/v2/parser"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"gopkg.in/yaml.v3"
 )
 
-func Parse(source string) (*model.Workflow, error) {
-	if len(source) == 0 {
+func Parse(source string) (*Workflow, error) {
+	if len(strings.TrimSpace(source)) == 0 {
+		return nil, nil
+	}
+	var raw map[string]interface{}
+	if err := yaml.Unmarshal([]byte(source), &raw); err != nil {
+		return nil, err
+	}
+	var wf *Workflow
+	var err error
+	if _, ok := raw["document"]; ok {
+		wf, err = parseV1Workflow(raw)
+	} else {
+		wf, err = parseLegacyWorkflow(raw)
+	}
+	if err != nil {
+		return nil, err
+	}
+	if wf == nil {
 		return nil, nil
 	}
-	return parser.FromYAMLSource(gconv.Bytes(source))
+	if verr := wf.Validate(); verr != nil {
+		return nil, verr
+	}
+	return wf, nil
+}
+
+func parseV1Workflow(raw map[string]interface{}) (*Workflow, error) {
+	doc := asMap(raw["document"])
+	name := asString(doc["name"])
+	if name == "" {
+		return nil, errors.New("workflow document.name is required")
+	}
+	version := asString(doc["version"])
+	if version == "" {
+		return nil, errors.New("workflow document.version is required")
+	}
+	dsl := asString(doc["dsl"])
+	if dsl == "" {
+		return nil, errors.New("workflow document.dsl is required")
+	}
+	tasks, err := parseV1TaskList(raw["do"])
+	if err != nil {
+		return nil, err
+	}
+	if len(tasks) == 0 {
+		return nil, errors.New("workflow do must contain at least one task")
+	}
+	return &Workflow{
+		ID:        name,
+		Name:      name,
+		Version:   version,
+		DSL:       dsl,
+		Namespace: asString(doc["namespace"]),
+		Start:     tasks[0].Name,
+		Tasks:     tasks,
+		Functions: parseReusableFunctions(raw),
+		Schedule:  parseSchedule(raw["schedule"]),
+		Input:     parseWorkflowInput(raw["input"]),
+		Output:    expressionString(asMap(raw["output"])["as"]),
+		Legacy:    false,
+	}, nil
+}
+
+func parseLegacyWorkflow(raw map[string]interface{}) (*Workflow, error) {
+	id := asString(raw["id"])
+	if id == "" {
+		return nil, errors.New("workflow id is required")
+	}
+	states := asSlice(raw["states"])
+	if len(states) == 0 {
+		return nil, errors.New("workflow states must contain at least one state")
+	}
+	functions := make(map[string]*Function)
+	for _, item := range asSlice(raw["functions"]) {
+		fnMap := asMap(item)
+		name := asString(fnMap["name"])
+		if name == "" {
+			continue
+		}
+		functions[name] = &Function{Name: name, Operation: asString(fnMap["operation"]), Type: asString(fnMap["type"])}
+	}
+	var tasks []*Task
+	for _, item := range states {
+		stateMap := asMap(item)
+		name := asString(stateMap["name"])
+		if name == "" {
+			continue
+		}
+		task := &Task{Name: name, Type: asString(stateMap["type"]), Then: legacyThen(stateMap), ExplicitThen: true}
+		if filter := asMap(stateMap["stateDataFilter"]); len(filter) > 0 {
+			task.InputFilter = asString(filter["input"])
+		}
+		task.Actions = parseLegacyActions(stateMap, functions)
+		if task.Type == TaskTypeSwitch {
+			task.Cases = parseLegacySwitchCases(stateMap)
+		}
+		tasks = append(tasks, task)
+	}
+	start := asString(raw["start"])
+	if start == "" && len(tasks) > 0 {
+		start = tasks[0].Name
+	}
+	return &Workflow{
+		ID:        id,
+		Name:      asString(raw["name"]),
+		Version:   asString(raw["version"]),
+		DSL:       asString(raw["specVersion"]),
+		Start:     start,
+		Tasks:     tasks,
+		Functions: functions,
+		Legacy:    true,
+	}, nil
+}
+
+func parseSchedule(value interface{}) *Schedule {
+	scheduleMap := asMap(value)
+	if len(scheduleMap) == 0 {
+		return nil
+	}
+	return &Schedule{
+		Start: asString(scheduleMap["start"]),
+		Cron:  asString(scheduleMap["cron"]),
+		After: asString(scheduleMap["after"]),
+	}
+}
+
+func parseWorkflowInput(value interface{}) string {
+	inputMap := asMap(value)
+	if len(inputMap) == 0 {
+		return ""
+	}
+	return expressionString(inputMap["from"])
+}
+
+func parseV1TaskList(value interface{}) ([]*Task, error) {
+	items := asSlice(value)
+	if len(items) == 0 {
+		return nil, nil
+	}
+	var tasks []*Task
+	for _, item := range items {
+		itemMap := asMap(item)
+		for taskName, taskDef := range itemMap {
+			task, err := parseV1Task(taskName, asMap(taskDef))
+			if err != nil {
+				return nil, err
+			}
+			tasks = append(tasks, task)
+		}
+	}
+	return tasks, nil
+}
+
+func parseV1Task(taskName string, def map[string]interface{}) (*Task, error) {
+	if taskName == "" {
+		return nil, errors.New("task name is required")
+	}
+	task := &Task{Name: taskName, Type: detectV1TaskType(def), Raw: mustJSON(def)}
+	if input := asMap(def["input"]); len(input) > 0 {
+		task.InputFilter = expressionString(input["from"])
+	}
+	if data := asString(def["data"]); data != "" {
+		task.InlineData = data
+	}
+	if output := asMap(def["output"]); len(output) > 0 {
+		task.OutputFilter = expressionString(output["as"])
+	}
+	if _, ok := def["then"]; ok {
+		task.Then = expressionString(def["then"])
+		task.ExplicitThen = true
+	}
+	switch task.Type {
+	case TaskTypeOperation:
+		task.Actions = []*Action{parseV1CallAction(def)}
+	case TaskTypeEvent:
+		task.Actions = []*Action{parseV1CallAction(def)}
+	case TaskTypeSet:
+		task.Actions = []*Action{{OperationName: taskName, OperationType: TaskTypeSet}}
+	case TaskTypeSwitch:
+		task.Cases = parseV1SwitchCases(def["switch"])
+	case TaskTypeDo:
+		children, err := parseV1TaskList(def["do"])
+		if err != nil {
+			return nil, err
+		}
+		task.Children = children
+	case TaskTypeFork:
+		task.Children = parseV1ForkBranches(def["fork"])
+	case TaskTypeFor:
+		children, err := parseV1TaskList(asMap(def["for"])["do"])
+		if err != nil {
+			return nil, err
+		}
+		task.Children = children
+	case TaskTypeTry:
+		children, err := parseV1TaskList(def["try"])
+		if err != nil {
+			return nil, err
+		}
+		task.Children = children
+		if catchMap := asMap(def["catch"]); len(catchMap) > 0 {
+			if when := catchMap["when"]; when != nil {
+				catchTasks, err := parseV1TaskList(when)
+				if err == nil {
+					task.Children = append(task.Children, catchTasks...)
+				}
+			}
+		}
+	case TaskTypeEmit, TaskTypeListen, TaskTypeWait, TaskTypeRaise, TaskTypeRun:
+		task.Actions = []*Action{{OperationName: taskName, OperationType: task.Type}}
+	}
+	return task, nil
+}
+
+func detectV1TaskType(def map[string]interface{}) string {
+	if _, ok := def["call"]; ok {
+		return TaskTypeOperation
+	}
+	for _, key := range []string{TaskTypeSwitch, TaskTypeSet, TaskTypeDo, TaskTypeFork, TaskTypeFor, TaskTypeTry, TaskTypeWait, TaskTypeRaise, TaskTypeRun, TaskTypeEmit, TaskTypeListen} {
+		if _, ok := def[key]; ok {
+			return key
+		}
+	}
+	return TaskTypeOperation
+}
+
+func parseV1CallAction(def map[string]interface{}) *Action {
+	operationType := expressionString(def["call"])
+	operationName := operationType
+	with := asMap(def["with"])
+	switch operationType {
+	case "http":
+		operationName = expressionString(with["endpoint"])
+	case "openapi":
+		operationName = firstNonEmpty(expressionString(with["operationId"]), expressionString(with["operation"]), expressionString(with["document"]))
+	case "asyncapi":
+		operationName = firstNonEmpty(expressionString(with["operation"]), expressionString(with["channel"]), expressionString(with["document"]))
+	case "grpc":
+		operationName = firstNonEmpty(expressionString(with["service"]), expressionString(with["method"]))
+	default:
+		operationName = firstNonEmpty(operationName, expressionString(with["operation"]), expressionString(with["endpoint"]))
+	}
+	return &Action{OperationName: operationName, OperationType: operationType}
+}
+
+func parseReusableFunctions(raw map[string]interface{}) map[string]*Function {
+	functions := make(map[string]*Function)
+	use := asMap(raw["use"])
+	for name, item := range asMap(use["functions"]) {
+		def := asMap(item)
+		action := parseV1CallAction(def)
+		functions[name] = &Function{Name: name, Operation: action.OperationName, Type: action.OperationType}
+	}
+	return functions
+}
+
+func parseV1SwitchCases(value interface{}) []*SwitchCase {
+	var cases []*SwitchCase
+	for _, item := range asSlice(value) {
+		itemMap := asMap(item)
+		for caseName, caseDef := range itemMap {
+			def := asMap(caseDef)
+			cases = append(cases, &SwitchCase{
+				Name:      caseName,
+				Condition: expressionString(def["when"]),
+				Then:      expressionString(def["then"]),
+				IsDefault: caseName == "default" || def["when"] == nil,
+			})
+		}
+	}
+	return cases
+}
+
+func parseV1ForkBranches(value interface{}) []*Task {
+	forkDef := asMap(value)
+	branches := asSlice(forkDef["branches"])
+	var tasks []*Task
+	for _, branch := range branches {
+		branchTasks, err := parseV1TaskList([]interface{}{branch})
+		if err == nil {
+			tasks = append(tasks, branchTasks...)
+		}
+	}
+	return tasks
+}
+
+func parseLegacyActions(stateMap map[string]interface{}, functions map[string]*Function) []*Action {
+	var actionValues []interface{}
+	if stateMap["actions"] != nil {
+		actionValues = asSlice(stateMap["actions"])
+	}
+	for _, onEvent := range asSlice(stateMap["onEvents"]) {
+		onEventMap := asMap(onEvent)
+		actionValues = append(actionValues, asSlice(onEventMap["actions"])...)
+	}
+	var actions []*Action
+	for _, actionValue := range actionValues {
+		actionMap := asMap(actionValue)
+		functionRef := asMap(actionMap["functionRef"])
+		functionName := asString(functionRef["refName"])
+		fn := functions[functionName]
+		if fn == nil {
+			continue
+		}
+		actions = append(actions, &Action{OperationName: fn.Operation, OperationType: fn.Type})
+	}
+	return actions
+}
+
+func parseLegacySwitchCases(stateMap map[string]interface{}) []*SwitchCase {
+	var cases []*SwitchCase
+	for _, conditionValue := range asSlice(stateMap["dataConditions"]) {
+		condition := asMap(conditionValue)
+		then := asString(condition["transition"])
+		if then == "" && asBool(condition["end"]) {
+			then = "end"
+		}
+		cases = append(cases, &SwitchCase{Name: asString(condition["name"]), Condition: expressionString(condition["condition"]), Then: then})
+	}
+	defaultCondition := asMap(stateMap["defaultCondition"])
+	if len(defaultCondition) > 0 {
+		then := asString(defaultCondition["transition"])
+		if then == "" && asBool(defaultCondition["end"]) {
+			then = "end"
+		}
+		cases = append(cases, &SwitchCase{Name: "default", Then: then, IsDefault: true})
+	}
+	return cases
+}
+
+func legacyThen(stateMap map[string]interface{}) string {
+	if transition := asString(stateMap["transition"]); transition != "" {
+		return transition
+	}
+	if asBool(stateMap["end"]) {
+		return "end"
+	}
+	return ""
+}
+
+func asMap(value interface{}) map[string]interface{} {
+	if value == nil {
+		return nil
+	}
+	if m, ok := value.(map[string]interface{}); ok {
+		return m
+	}
+	if m, ok := value.(map[interface{}]interface{}); ok {
+		res := make(map[string]interface{}, len(m))
+		for key, item := range m {
+			res[fmt.Sprint(key)] = item
+		}
+		return res
+	}
+	return nil
+}
+
+func asSlice(value interface{}) []interface{} {
+	if value == nil {
+		return nil
+	}
+	if items, ok := value.([]interface{}); ok {
+		return items
+	}
+	return nil
+}
+
+func asString(value interface{}) string {
+	if value == nil {
+		return ""
+	}
+	return fmt.Sprint(value)
+}
+
+func asBool(value interface{}) bool {
+	if b, ok := value.(bool); ok {
+		return b
+	}
+	return strings.EqualFold(asString(value), "true")
+}
+
+func expressionString(value interface{}) string {
+	text := strings.TrimSpace(asString(value))
+	text = strings.TrimPrefix(text, "${")
+	text = strings.TrimSuffix(text, "}")
+	return strings.TrimSpace(text)
+}

Review Comment:
   `expressionString` strips only the `${` prefix and trailing `}`, which breaks legacy `${{ ... }}` expressions by leaving a leading `{` (and only removing one `}`), producing invalid JQ expressions (e.g. "{ .x }"), impacting legacy switch conditions / filters.



##########
internal/task/a2a_integration.go:
##########
@@ -0,0 +1,50 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package task
+
+import (
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/internal/bridge"
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/internal/dal/model"
+)
+
+// isA2ATask returns true if the task action targets an A2A agent.
+// An action is considered A2A if OperationType is "a2a" or "agent",
+// or if OperationName starts with "a2a:" or "agent:".
+func isA2ATask(action *model.WorkflowTaskAction) bool {
+	if action == nil {
+		return false
+	}
+	switch action.OperationType {
+	case "a2a", "agent":
+		return true
+	}
+	if len(action.OperationName) > 0 {
+		if action.OperationName[:3] == "a2a" {
+			return true
+		}
+	}
+	return false

Review Comment:
   `isA2ATask` can panic when `OperationName` is shorter than 3 characters because it slices `action.OperationName[:3]` after only checking `len(...) > 0`. This is a runtime crash risk for short operation names.



##########
internal/dal/workflow.go:
##########
@@ -310,189 +308,195 @@ func (w *workflowDALImpl) create(ctx context.Context, tx *gorm.DB, record *model
 	return util.GoAndWait(handlers...)
 }
 
-func (w *workflowDALImpl) buildTask(workflow *pmodel.Workflow) []*model.WorkflowTask {
-	if workflow == nil || len(workflow.States) == 0 {
+func (w *workflowDALImpl) buildTask(workflow *swf.Workflow) []*model.WorkflowTask {
+	if workflow == nil || len(workflow.Tasks) == 0 {
 		return nil
 	}
 	var tasks []*model.WorkflowTask
-
-	for _, state := range workflow.States {
+	for _, workflowTask := range workflow.FlattenTasks() {
 		var task = model.WorkflowTask{}
 		task.WorkflowID = workflow.ID
 		task.TaskID = uuid.New().String()
-		task.TaskName = state.GetName()
+		task.TaskName = workflowTask.Name
 		task.Status = constants.NormalStatus
-		task.TaskType = gconv.String(state.GetType())
+		task.TaskType = normalizeTaskType(workflowTask.Type)
+		task.TaskInputFilter = workflowTask.InputFilter
+		task.TaskOutputFilter = workflowTask.OutputFilter
+		task.TaskDefinition = workflowTask.Raw
 		task.CreateTime = time.Now()
 		task.UpdateTime = time.Now()
-		task.Actions = w.buildTaskAction(task.TaskID, workflow, state)
-		w.fillTaskFilterIfExist(state, &task)
+		task.Actions = w.buildTaskAction(task.TaskID, workflow.ID, workflowTask)
 		tasks = append(tasks, &task)
 	}
 	return tasks
 }
 
-func (w *workflowDALImpl) fillTaskFilterIfExist(workflowState pmodel.State, task *model.WorkflowTask) {
-	filter := workflowState.GetStateDataFilter()
-	if filter != nil {
-		task.TaskInputFilter = filter.Input
+func normalizeTaskType(taskType string) string {
+	switch taskType {
+	case swf.TaskTypeOperation:
+		return constants.TaskTypeOperation
+	case swf.TaskTypeEvent:
+		return constants.TaskTypeEvent
+	case swf.TaskTypeSwitch:
+		return constants.TaskTypeSwitch
+	case swf.TaskTypeSet:
+		return constants.TaskTypeSet
+	case swf.TaskTypeDo:
+		return constants.TaskTypeDo
+	case swf.TaskTypeFork:
+		return constants.TaskTypeFork
+	case swf.TaskTypeFor:
+		return constants.TaskTypeFor
+	case swf.TaskTypeTry:
+		return constants.TaskTypeTry
+	case swf.TaskTypeWait:
+		return constants.TaskTypeWait
+	case swf.TaskTypeRaise:
+		return constants.TaskTypeRaise
+	case swf.TaskTypeRun:
+		return constants.TaskTypeRun
+	case swf.TaskTypeEmit:
+		return constants.TaskTypeEmit
+	case swf.TaskTypeListen:
+		return constants.TaskTypeListen
+	default:
+		return constants.TaskTypeOperation
 	}
 }
 
-func (w *workflowDALImpl) buildTaskAction(taskID string, workflow *pmodel.Workflow,
-	state pmodel.State) []*model.WorkflowTaskAction {
-	var functions = make(map[string]*pmodel.Function)
-	for i, function := range workflow.Functions {
-		functions[function.Name] = &workflow.Functions[i]
+func (w *workflowDALImpl) buildTaskAction(taskID string, workflowID string,
+	workflowTask *swf.Task) []*model.WorkflowTaskAction {
+	if workflowTask == nil {
+		return nil
 	}
-	switch state.GetType() {
-	case pmodel.StateTypeOperation:
-		return w.doBuildOperationTaskAction(workflow.ID, taskID, functions, state)
-	case pmodel.StateTypeEvent:
-		return w.doBuildEventTaskAction(workflow.ID, taskID, functions, state)
+	var actions []*model.WorkflowTaskAction
+	for _, action := range workflowTask.Actions {
+		if action == nil {
+			continue
+		}
+		var taskAction model.WorkflowTaskAction
+		taskAction.WorkflowID = workflowID
+		taskAction.TaskID = taskID
+		taskAction.OperationName = action.OperationName
+		taskAction.OperationType = action.OperationType
+		taskAction.Status = constants.NormalStatus
+		taskAction.CreateTime = time.Now()
+		taskAction.UpdateTime = time.Now()
+		actions = append(actions, &taskAction)
 	}
-	return nil
+	if len(actions) == 0 {
+		var taskAction model.WorkflowTaskAction
+		taskAction.WorkflowID = workflowID
+		taskAction.TaskID = taskID
+		taskAction.OperationName = workflowTask.Name
+		taskAction.OperationType = normalizeTaskType(workflowTask.Type)
+		taskAction.Status = constants.NormalStatus
+		taskAction.CreateTime = time.Now()
+		taskAction.UpdateTime = time.Now()
+		actions = append(actions, &taskAction)
+	}
+	return actions
 }
 
-func (w *workflowDALImpl) buildTaskRelation(workflow *pmodel.Workflow,
+func (w *workflowDALImpl) buildTaskRelation(workflow *swf.Workflow,
 	tasks []*model.WorkflowTask) []*model.WorkflowTaskRelation {
-	if workflow == nil || len(workflow.States) == 0 {
+	if workflow == nil || len(workflow.Tasks) == 0 {
 		return nil
 	}
 	var taskIDs = make(map[string]string)
 	for _, task := range tasks {
 		taskIDs[task.TaskName] = task.TaskID
 	}
 	var taskRelations []*model.WorkflowTaskRelation
-	for _, state := range workflow.States {
-		if workflow.Start.StateName == state.GetName() {
-			taskRelations = append(taskRelations, w.doBuildStartTaskRelation(workflow, state, taskIDs))
+	if startTaskID := taskIDs[workflow.Start]; startTaskID != "" {
+		taskRelations = append(taskRelations, newTaskRelation(workflow.ID, constants.TaskStartID, startTaskID, ""))
+	}
+	for _, workflowTask := range workflow.FlattenTasks() {
+		fromTaskID := taskIDs[workflowTask.Name]
+		if fromTaskID == "" {
+			continue
 		}
-		switch state.GetType() {
-		case pmodel.StateTypeOperation:
-			fallthrough
-		case pmodel.StateTypeEvent:
-			taskRelations = append(taskRelations, w.doBuildTaskRelation(workflow, state, taskIDs))
-		case pmodel.StateTypeSwitch:
-			taskRelations = append(taskRelations, w.doBuildSwitchTaskRelation(workflow, state, taskIDs)...)
-		default:
-			log.Errorf("buildTaskRelation=not support type=%s", state.GetType())
+		if workflowTask.Type == swf.TaskTypeSwitch {
+			taskRelations = append(taskRelations, w.buildSwitchTaskRelation(workflow.ID, workflowTask, fromTaskID, taskIDs)...)
+			continue
 		}
+		if workflowTask.Type == swf.TaskTypeFork {
+			taskRelations = append(taskRelations, w.buildForkTaskRelation(workflow.ID, workflowTask, fromTaskID, taskIDs)...)
+			continue

Review Comment:
   In `buildTaskRelation`, fork tasks only get relations to each branch head (`buildForkTaskRelation`) and then `continue`, so a fork's explicit `then:` target (e.g. a join task) is never linked and becomes unreachable in the task graph.



##########
internal/task/local_runtime.go:
##########
@@ -0,0 +1,305 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package task
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/internal/constants"
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/internal/dal/model"
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/internal/filter"
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/third_party/jqer"
+)
+
+type localRuntimeTask struct {
+	baseTask
+	action      *model.WorkflowTaskAction
+	transitions []*model.WorkflowTaskRelation
+	definition  map[string]interface{}
+}
+
+func NewLocalRuntimeTask(instance *model.WorkflowTaskInstance) Task {
+	var t localRuntimeTask
+	if instance == nil || instance.Task == nil {
+		return nil
+	}
+	t.baseTask = newBaseTask(instance)
+	if len(instance.Task.Actions) > 0 {
+		t.action = instance.Task.Actions[0]
+	}
+	t.transitions = instance.Task.ChildTasks
+	if instance.Task.TaskDefinition != "" {
+		_ = json.Unmarshal([]byte(instance.Task.TaskDefinition), &t.definition)
+	}
+	return &t
+}
+
+func (t *localRuntimeTask) Run() error {
+	nextInput, err := t.execute()
+	if err != nil {
+		return err
+	}
+	nextInput = filter.FilterWorkflowTaskOutputData(nextInput, t.outputFilter)
+	if len(t.transitions) == 0 || t.transitions[0] == nil || t.transitions[0].ToTaskID == constants.TaskEndID {
+		return completeWorkflow(t.baseTask)
+	}
+	if len(t.transitions) == 1 {
+		return publishNextOrComplete(t.baseTask, t.transitions[0], nextInput)
+	}
+	return publishNextTasks(t.baseTask, t.transitions, nextInput)
+}
+
+func (t *localRuntimeTask) execute() (string, error) {
+	switch t.taskType {
+	case constants.TaskTypeSet:
+		return t.executeSet()
+	case constants.TaskTypeWait:
+		return t.executeWait()
+	case constants.TaskTypeRaise:
+		return "", t.executeRaise()
+	case constants.TaskTypeRun:
+		return t.input, t.executeRun()
+	case constants.TaskTypeFork:
+		return t.input, nil
+	case constants.TaskTypeTry:
+		return t.executeTry()
+	case constants.TaskTypeFor:
+		return t.executeFor()
+	case constants.TaskTypeDo:
+		return t.executeDo()
+	case constants.TaskTypeEmit:
+		return t.input, t.executeEmit()
+	default:
+		return t.input, nil
+	}
+}
+
+func (t *localRuntimeTask) executeSet() (string, error) {
+	setValue, ok := t.definition[constants.TaskTypeSet]
+	if !ok {
+		return t.input, nil
+	}
+	var input interface{}
+	if t.input != "" {
+		if err := json.Unmarshal([]byte(t.input), &input); err != nil {
+			return "", err
+		}
+	}
+	jq := jqer.NewJQ()
+	ret, err := jq.Object(input, setValue)
+	if err != nil {
+		return "", err
+	}
+	bytes, err := json.Marshal(ret)
+	if err != nil {
+		return "", err
+	}
+	return string(bytes), nil
+}
+
+func (t *localRuntimeTask) executeWait() (string, error) {
+	waitDef := asMap(t.definition[constants.TaskTypeWait])
+	durationText := firstNonEmpty(asString(waitDef["duration"]), asString(waitDef["seconds"]), asString(waitDef["milliseconds"]))
+	if durationText == "" {
+		return t.input, nil
+	}
+	duration, err := time.ParseDuration(durationText)
+	if err != nil {
+		return "", err
+	}
+	if duration > 0 {
+		time.Sleep(duration)
+	}
+	return t.input, nil
+}
+
+func (t *localRuntimeTask) executeRaise() error {
+	raiseDef := asMap(t.definition[constants.TaskTypeRaise])
+	if len(raiseDef) == 0 {
+		return errors.New("workflow raise task failed")
+	}
+	errorDef := asMap(raiseDef["error"])
+	if len(errorDef) == 0 {
+		return fmt.Errorf("workflow raise task failed: %v", raiseDef)
+	}
+	return fmt.Errorf("workflow raise task failed: type=%s title=%s status=%s detail=%s",
+		asString(errorDef["type"]), asString(errorDef["title"]), asString(errorDef["status"]), asString(errorDef["detail"]))
+}
+
+func (t *localRuntimeTask) executeRun() error {
+	if t.action == nil || t.action.OperationName == "" {
+		return nil
+	}
+	return publishEvent(t.workflowInstanceID, t.taskInstanceID, t.action.OperationName, t.input)
+}
+
+func (t *localRuntimeTask) executeEmit() error {
+	if t.action == nil || t.action.OperationName == "" {
+		return nil
+	}
+	return publishEvent(t.workflowInstanceID, t.taskInstanceID, t.action.OperationName, t.input)
+}
+
+func (t *localRuntimeTask) executeTry() (string, error) {
+	tryTasks := asSlice(t.definition["try"])
+	if len(tryTasks) == 0 {
+		return t.input, nil
+	}

Review Comment:
   `executeTry` does not implement catch semantics at all: it reads only `definition["try"]` and ignores `definition["catch"]`, so failures cannot trigger the configured `catch.when` tasks as described in the design/docs.



##########
internal/dal/model/workflow_task.go:
##########
@@ -24,6 +24,8 @@ type WorkflowTask struct {
 	TaskName           string                  `json:"task_name" gorm:"column:task_name;type:varchar;size:1024"`
 	TaskType           string                  `json:"task_type" gorm:"column:task_type;type:varchar;size:64"`
 	TaskInputFilter    string                  `json:"task_input_filter" gorm:"column:task_input_filter;type:varchar;size:1024"`
+	TaskOutputFilter   string                  `json:"task_output_filter" gorm:"column:task_output_filter;type:varchar;size:1024"`
+	TaskDefinition     string                  `json:"task_definition" gorm:"-"`
 	Status             int                     `json:"status" gorm:"column:status;type:int"`

Review Comment:
   `TaskDefinition` is marked `gorm:"-"` (not persisted), but local runtime execution depends on `instance.Task.TaskDefinition` to parse the task structure. This means structural tasks (set/do/for/try/...) will lose their definitions after being stored and reloaded from MySQL and will become no-ops.



##########
internal/task/operation_task.go:
##########
@@ -37,37 +36,54 @@ func NewOperationTask(instance *model.WorkflowTaskInstance) Task {
 	if instance == nil || instance.Task == nil {
 		return nil
 	}
-	t.baseTask = baseTask{taskID: instance.TaskID, taskInstanceID: instance.TaskInstanceID, input: instance.Input,
-		workflowID: instance.WorkflowID, workflowInstanceID: instance.WorkflowInstanceID, taskType: instance.Task.TaskType}
-	t.action = instance.Task.Actions[0]
-	t.transition = instance.Task.ChildTasks[0]
-	t.baseTask.queue = queue.GetQueue(config.GlobalConfig().Flow.Queue.Store)
+	t.baseTask = newBaseTask(instance)
+	if len(instance.Task.Actions) > 0 {
+		t.action = instance.Task.Actions[0]
+	}
+	if len(instance.Task.ChildTasks) > 0 {
+		t.transition = instance.Task.ChildTasks[0]
+	}
+	t.queue = queue.GetQueue(config.GlobalConfig().Flow.Queue.Store)
 	t.workflowDAL = dal.NewWorkflowDAL()
 	return &t
 }
 
 func (t *operationTask) Run() error {
-	metrics.Inc(constants.MetricsOperationTask, constants.MetricsTotal)
-	if t.action == nil {
-		return nil
-	}
-	// match end
-	if t.transition.ToTaskID == constants.TaskEndID {
-		if t.action != nil {
-			if err := publishEvent(t.workflowInstanceID, uuid.New().String(), t.action.OperationName, t.input); err != nil {
-				return err
-			}
+	_ = metrics.Inc(constants.MetricsOperationTask, constants.MetricsTotal)
+	if t.transition == nil || t.transition.ToTaskID == constants.TaskEndID {
+		if err := t.runAction(uuid.New().String()); err != nil {
+			return err
 		}
-		return t.workflowDAL.UpdateInstance(context.Background(),
-			&model.WorkflowInstance{WorkflowInstanceID: t.workflowInstanceID,
-				WorkflowStatus: constants.WorkflowInstanceSuccessStatus})
+		return publishNextOrComplete(t.baseTask, t.transition, t.input)
 	}
 	var taskInstanceID = uuid.New().String()
 	var taskInstance = model.WorkflowTaskInstance{WorkflowInstanceID: t.workflowInstanceID, WorkflowID: t.workflowID,
 		TaskID: t.transition.ToTaskID, TaskInstanceID: taskInstanceID, Status: constants.TaskInstanceSleepStatus,
-		Input: t.baseTask.input}
-	if err := t.baseTask.queue.Publish([]*model.WorkflowTaskInstance{&taskInstance}); err != nil {
+		Input: t.input}
+	if err := t.queue.Publish([]*model.WorkflowTaskInstance{&taskInstance}); err != nil {

Review Comment:
   For A2A actions, `Run()` always enqueues the next task instance in SLEEP status and then calls `runA2AAction()`, but A2A execution does not publish an event or call Transition, so the next task will never be awakened. A2A tasks should execute synchronously and publish the next task in WAIT status with the updated input.



##########
internal/bridge/workflow_agent.go:
##########
@@ -0,0 +1,191 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package bridge
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"sync"
+
+	"github.com/apache/incubator-eventmesh/eventmesh-server-go/log"
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/flow"
+	"github.com/google/uuid"
+)
+
+// WorkflowAgent exposes workflow execution capabilities through the A2A protocol.
+// A2A clients can submit tasks that trigger workflow executions and monitor their status.
+
+type WorkflowAgent struct {
+	card           A2AAgentCard
+	engine         *flow.Engine
+	mu             sync.RWMutex
+	taskToWorkflow map[string]string // A2A task ID → workflow instance ID
+}
+
+func NewWorkflowAgent(name, baseURL string) *WorkflowAgent {
+	return &WorkflowAgent{
+		card:           BuildAgentCard(name, name, baseURL),
+		engine:         flow.NewEngine(),
+		taskToWorkflow: make(map[string]string),
+	}
+}
+
+func (wa *WorkflowAgent) AgentCard() A2AAgentCard {
+	return wa.card
+}
+
+func (wa *WorkflowAgent) RegisterRoutes(mux *http.ServeMux) {
+	mux.HandleFunc("/.well-known/agent-card.json", wa.handleAgentCard)
+	mux.HandleFunc("/a2a/tasks", wa.handleSubmitTask)
+	mux.HandleFunc("/a2a/tasks/", wa.handleTaskStatus)
+	mux.HandleFunc("/a2a/health", wa.handleHealth)
+}
+
+// A2A HTTP Handlers
+
+func (wa *WorkflowAgent) handleAgentCard(w http.ResponseWriter, r *http.Request) {
+	body, _ := wa.card.ToJSON()
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("Access-Control-Allow-Origin", "*")
+	if _, err := w.Write(body); err != nil {
+		log.Errorf("fail to write agent card response: %v", err)
+	}
+}
+
+func (wa *WorkflowAgent) handleSubmitTask(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+		return
+	}
+	var req A2ATaskRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeJSON(w, http.StatusBadRequest, A2ATaskResponse{
+			Error: &A2AError{Message: "invalid request: " + err.Error()},
+		})
+		return
+	}
+	input := extractTaskInput(req)
+	if input == "" {
+		writeJSON(w, http.StatusBadRequest, A2ATaskResponse{
+			Error: &A2AError{Message: "task message must contain text"},
+		})
+		return
+	}
+
+	// Start workflow execution
+	instanceID := uuid.New().String()
+	flowParam := &flow.WorkflowParam{ID: wa.card.Name, Input: input}
+	wfInstanceID, err := wa.engine.Start(context.Background(), flowParam)
+	if err != nil {
+		log.Errorf("WorkflowAgent: start workflow failed: %v", err)
+		writeJSON(w, http.StatusInternalServerError, A2ATaskResponse{
+			Error: &A2AError{Message: "workflow start failed: " + err.Error()},
+		})
+		return
+	}
+
+	wa.mu.Lock()
+	wa.taskToWorkflow[instanceID] = wfInstanceID
+	wa.mu.Unlock()
+
+	a2aTaskID := instanceID
+	if req.ID != "" {
+		a2aTaskID = req.ID
+		wa.mu.Lock()
+		wa.taskToWorkflow[a2aTaskID] = wfInstanceID
+		wa.mu.Unlock()
+	}
+
+	resp := A2ATaskResponse{
+		ID:     a2aTaskID,
+		Status: A2ATaskStatusWorking,
+		Message: A2AMessage{
+			Role: "agent",
+			Parts: []A2ATextPart{
+				{Type: "text", Text: "Workflow execution started: " + wfInstanceID},
+			},
+		},
+	}
+	writeJSON(w, http.StatusOK, resp)
+}
+
+func (wa *WorkflowAgent) handleTaskStatus(w http.ResponseWriter, r *http.Request) {
+	taskID := extractTaskIDFromPath(r.URL.Path, "/a2a/tasks/")
+	if taskID == "" {
+		http.NotFound(w, r)
+		return
+	}
+
+	wa.mu.RLock()
+	wfInstanceID, ok := wa.taskToWorkflow[taskID]
+	wa.mu.RUnlock()
+
+	if !ok {
+		writeJSON(w, http.StatusNotFound, A2ATaskResponse{
+			Error: &A2AError{Message: "task not found"},
+		})
+		return
+	}
+
+	resp := A2ATaskResponse{
+		ID:     taskID,
+		Status: A2ATaskStatusWorking,
+		Message: A2AMessage{

Review Comment:
   `handleTaskStatus` always returns `Status: working` and never checks the actual workflow instance status, so A2A clients cannot observe completion/failure and will poll indefinitely.



##########
internal/task/local_runtime.go:
##########
@@ -0,0 +1,305 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package task
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/internal/constants"
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/internal/dal/model"
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/internal/filter"
+	"github.com/apache/incubator-eventmesh/eventmesh-workflow-go/third_party/jqer"
+)
+
+type localRuntimeTask struct {
+	baseTask
+	action      *model.WorkflowTaskAction
+	transitions []*model.WorkflowTaskRelation
+	definition  map[string]interface{}
+}
+
+func NewLocalRuntimeTask(instance *model.WorkflowTaskInstance) Task {
+	var t localRuntimeTask
+	if instance == nil || instance.Task == nil {
+		return nil
+	}
+	t.baseTask = newBaseTask(instance)
+	if len(instance.Task.Actions) > 0 {
+		t.action = instance.Task.Actions[0]
+	}
+	t.transitions = instance.Task.ChildTasks
+	if instance.Task.TaskDefinition != "" {
+		_ = json.Unmarshal([]byte(instance.Task.TaskDefinition), &t.definition)
+	}
+	return &t
+}
+
+func (t *localRuntimeTask) Run() error {
+	nextInput, err := t.execute()
+	if err != nil {
+		return err
+	}
+	nextInput = filter.FilterWorkflowTaskOutputData(nextInput, t.outputFilter)
+	if len(t.transitions) == 0 || t.transitions[0] == nil || t.transitions[0].ToTaskID == constants.TaskEndID {
+		return completeWorkflow(t.baseTask)
+	}
+	if len(t.transitions) == 1 {
+		return publishNextOrComplete(t.baseTask, t.transitions[0], nextInput)
+	}
+	return publishNextTasks(t.baseTask, t.transitions, nextInput)
+}
+
+func (t *localRuntimeTask) execute() (string, error) {
+	switch t.taskType {
+	case constants.TaskTypeSet:
+		return t.executeSet()
+	case constants.TaskTypeWait:
+		return t.executeWait()
+	case constants.TaskTypeRaise:
+		return "", t.executeRaise()
+	case constants.TaskTypeRun:
+		return t.input, t.executeRun()
+	case constants.TaskTypeFork:
+		return t.input, nil
+	case constants.TaskTypeTry:
+		return t.executeTry()
+	case constants.TaskTypeFor:
+		return t.executeFor()
+	case constants.TaskTypeDo:
+		return t.executeDo()
+	case constants.TaskTypeEmit:
+		return t.input, t.executeEmit()
+	default:
+		return t.input, nil
+	}
+}
+
+func (t *localRuntimeTask) executeSet() (string, error) {
+	setValue, ok := t.definition[constants.TaskTypeSet]
+	if !ok {
+		return t.input, nil
+	}
+	var input interface{}
+	if t.input != "" {
+		if err := json.Unmarshal([]byte(t.input), &input); err != nil {
+			return "", err
+		}
+	}
+	jq := jqer.NewJQ()
+	ret, err := jq.Object(input, setValue)
+	if err != nil {
+		return "", err
+	}
+	bytes, err := json.Marshal(ret)
+	if err != nil {
+		return "", err
+	}
+	return string(bytes), nil
+}
+
+func (t *localRuntimeTask) executeWait() (string, error) {
+	waitDef := asMap(t.definition[constants.TaskTypeWait])
+	durationText := firstNonEmpty(asString(waitDef["duration"]), asString(waitDef["seconds"]), asString(waitDef["milliseconds"]))
+	if durationText == "" {
+		return t.input, nil
+	}
+	duration, err := time.ParseDuration(durationText)
+	if err != nil {
+		return "", err
+	}
+	if duration > 0 {
+		time.Sleep(duration)
+	}
+	return t.input, nil
+}
+
+func (t *localRuntimeTask) executeRaise() error {
+	raiseDef := asMap(t.definition[constants.TaskTypeRaise])
+	if len(raiseDef) == 0 {
+		return errors.New("workflow raise task failed")
+	}
+	errorDef := asMap(raiseDef["error"])
+	if len(errorDef) == 0 {
+		return fmt.Errorf("workflow raise task failed: %v", raiseDef)
+	}
+	return fmt.Errorf("workflow raise task failed: type=%s title=%s status=%s detail=%s",
+		asString(errorDef["type"]), asString(errorDef["title"]), asString(errorDef["status"]), asString(errorDef["detail"]))
+}
+
+func (t *localRuntimeTask) executeRun() error {
+	if t.action == nil || t.action.OperationName == "" {
+		return nil
+	}
+	return publishEvent(t.workflowInstanceID, t.taskInstanceID, t.action.OperationName, t.input)
+}
+
+func (t *localRuntimeTask) executeEmit() error {
+	if t.action == nil || t.action.OperationName == "" {
+		return nil
+	}
+	return publishEvent(t.workflowInstanceID, t.taskInstanceID, t.action.OperationName, t.input)
+}
+
+func (t *localRuntimeTask) executeTry() (string, error) {
+	tryTasks := asSlice(t.definition["try"])
+	if len(tryTasks) == 0 {
+		return t.input, nil
+	}
+	for _, taskItem := range tryTasks {
+		taskMap := asMap(taskItem)
+		for taskName, taskDef := range taskMap {
+			def := asMap(taskDef)
+			taskType := detectTaskType(def)
+			switch taskType {
+			case constants.TaskTypeSet:
+				return t.executeSetFromDef(def)
+			case constants.TaskTypeRaise:
+				return "", fmt.Errorf("try task %s raised: %v", taskName, def)
+			case constants.TaskTypeCall:
+				return t.input, nil
+			}
+		}
+	}
+	return t.input, nil
+}
+
+func (t *localRuntimeTask) executeFor() (string, error) {
+	forDef := asMap(t.definition["for"])
+	forDo := asMap(forDef["do"])
+	if len(forDo) == 0 {
+		return t.input, nil
+	}
+	var items []interface{}
+	if t.input != "" {
+		var inputData interface{}
+		if err := json.Unmarshal([]byte(t.input), &inputData); err == nil {
+			if arr, ok := inputData.([]interface{}); ok {
+				items = arr
+			}
+		}
+	}
+	if len(items) == 0 {
+		items = append(items, t.input)
+	}
+	for _, item := range items {
+		itemInput, _ := json.Marshal(item)
+		for _, taskItem := range asSlice(forDo["do"]) {
+			taskMap := asMap(taskItem)

Review Comment:
   `executeFor` treats `forDef["do"]` as a map (`asMap`), but in the DSL it is a task list (slice). This makes `len(forDo)==0` and the loop never executes, so `for` currently becomes a no-op.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org