From a6700843747df1e6e4c73543191fc64cfd3ea1b9 Mon Sep 17 00:00:00 2001 From: mlogclub Date: Tue, 23 Jun 2026 22:57:18 +0800 Subject: [PATCH] Enhance workflow execution with detailed run logging and error handling, including workflow run status tracking and node tracing --- internal/ai/application/runtime/service.go | 145 +++++++++++++++--- .../application/runtime/tool_catalog_test.go | 13 ++ internal/ai/application/runtime/types.go | 5 + .../runtime/workflow_summary_test.go | 68 +++++++- internal/ai/runtime/reply_helpers_test.go | 20 +++ internal/ai/runtime/reply_runlog_service.go | 36 +++++ internal/ai/runtime/runtime_reply_executor.go | 1 + internal/ai/runtime/workflow/executor.go | 81 +++++++++- 8 files changed, 335 insertions(+), 34 deletions(-) diff --git a/internal/ai/application/runtime/service.go b/internal/ai/application/runtime/service.go index b68a406..f56db47 100644 --- a/internal/ai/application/runtime/service.go +++ b/internal/ai/application/runtime/service.go @@ -21,6 +21,12 @@ type Service struct { prepare *prepareService } +const ( + workflowRunStatusCompleted = 1 + workflowRunStatusInterrupted = 2 + workflowRunStatusFailed = 3 +) + func NewService() *Service { catalog := newToolCatalog() return &Service{ @@ -45,12 +51,16 @@ func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) { AIConfig: req.AIConfig, }) if err != nil { + if workflowResult != nil { + _, _ = writeWorkflowRun(req, workflow, workflowResult, err.Error()) + } return nil, err } - if err := writeWorkflowRun(req, workflow, workflowResult, ""); err != nil { + workflowRunID, err := writeWorkflowRun(req, workflow, workflowResult, "") + if err != nil { return nil, err } - return toWorkflowSummary(workflowResult, req.AIConfig.ModelName), nil + return toWorkflowSummary(workflowResult, req.AIConfig.ModelName, workflow, workflowRunID), nil } func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, error) { @@ -67,9 +77,26 @@ func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, erro AIConfig: req.AIConfig, }, interrupt.RequestData, firstWorkflowResumeText(req.ResumeData)) if err != nil { + if workflowResult != nil { + _, _ = writeWorkflowRun(Request{ + Conversation: req.Conversation, + UserMessage: req.UserMessage, + AIAgent: req.AIAgent, + AIConfig: req.AIConfig, + }, workflow, workflowResult, err.Error()) + } return nil, err } - return toWorkflowSummary(workflowResult, req.AIConfig.ModelName), nil + workflowRunID, err := writeWorkflowRun(Request{ + Conversation: req.Conversation, + UserMessage: req.UserMessage, + AIAgent: req.AIAgent, + AIConfig: req.AIConfig, + }, workflow, workflowResult, "") + if err != nil { + return nil, err + } + return toWorkflowSummary(workflowResult, req.AIConfig.ModelName, workflow, workflowRunID), nil } toolSet, err := s.prepare.prepareToolsForResume(req) if err != nil { @@ -99,27 +126,34 @@ func firstWorkflowResumeText(data map[string]string) string { return "" } -func toWorkflowSummary(result *workflowexecutor.Result, modelName string) *Summary { +func toWorkflowSummary(result *workflowexecutor.Result, modelName string, workflow resolvedWorkflow, workflowRunID int64) *Summary { if result == nil { return nil } trace := map[string]any{ - "status": result.Status, - "nodePath": result.NodePath, + "status": result.Status, + "workflowId": workflow.WorkflowID, + "workflowVersionId": workflow.VersionID, + "workflowRunId": workflowRunID, + "nodePath": result.NodePath, } traceData, _ := json.Marshal(trace) return &Summary{ - Status: result.Status, - ReplyText: result.ReplyText, - ModelName: modelName, - PromptTokens: result.PromptTokens, - CompletionTokens: result.CompletionTokens, - RetrieverCount: result.RetrieverCount, - TraceData: string(traceData), - CheckPointID: result.CheckPointID, - CheckPointData: result.CheckPointData, - Interrupted: result.Interrupted, - Interrupts: toWorkflowInterruptSummaries(result.Interrupts), + Status: result.Status, + ReplyText: result.ReplyText, + ModelName: modelName, + PromptTokens: result.PromptTokens, + CompletionTokens: result.CompletionTokens, + RetrieverCount: result.RetrieverCount, + WorkflowID: workflow.WorkflowID, + WorkflowVersionID: workflow.VersionID, + WorkflowRunID: workflowRunID, + WorkflowNodePath: append([]string(nil), result.NodePath...), + TraceData: string(traceData), + CheckPointID: result.CheckPointID, + CheckPointData: result.CheckPointData, + Interrupted: result.Interrupted, + Interrupts: toWorkflowInterruptSummaries(result.Interrupts), } } @@ -138,9 +172,9 @@ func toWorkflowInterruptSummaries(items []workflowexecutor.InterruptSummary) []I return ret } -func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowexecutor.Result, errorMessage string) error { +func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowexecutor.Result, errorMessage string) (int64, error) { if result == nil { - return nil + return 0, nil } now := time.Now() endedAt := now @@ -148,29 +182,42 @@ func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowex for _, node := range workflow.Definition.Nodes { nodeTypes[node.ID] = node.Type } - return sqls.WithTransaction(func(ctx *sqls.TxContext) error { + runStatus := workflowRunStatus(result.Status, errorMessage) + var runID int64 + err := sqls.WithTransaction(func(ctx *sqls.TxContext) error { run := &models.AIWorkflowRun{ WorkflowID: workflow.WorkflowID, WorkflowVersionID: workflow.VersionID, ConversationID: req.Conversation.ID, AIAgentID: req.AIAgent.ID, MessageID: req.UserMessage.ID, - Status: 1, + Status: runStatus, StartedAt: now, EndedAt: &endedAt, + InterruptType: firstWorkflowInterruptType(result), + InterruptNodeID: firstWorkflowInterruptNodeID(result), ErrorMessage: errorMessage, } if err := repositories.AIWorkflowRunRepository.Create(ctx.Tx, run); err != nil { return err } - for _, nodeID := range result.NodePath { + runID = run.ID + nodeTraces := result.NodeTraces + if len(nodeTraces) == 0 { + nodeTraces = fallbackWorkflowNodeTraces(result.NodePath, nodeTypes, result.Status) + } + for _, nodeTrace := range nodeTraces { nodeRun := &models.AIWorkflowNodeRun{ WorkflowRunID: run.ID, - NodeID: nodeID, - NodeType: nodeTypes[nodeID], - Status: 1, + NodeID: nodeTrace.NodeID, + NodeType: firstNonEmpty(nodeTrace.NodeType, nodeTypes[nodeTrace.NodeID]), + Status: workflowRunStatus(nodeTrace.Status, nodeTrace.ErrorMessage), + InputPreview: nodeTrace.InputPreview, + OutputPreview: nodeTrace.OutputPreview, + ErrorMessage: nodeTrace.ErrorMessage, StartedAt: now, EndedAt: &endedAt, + DurationMS: nodeTrace.DurationMS, } if err := repositories.AIWorkflowNodeRunRepository.Create(ctx.Tx, nodeRun); err != nil { return err @@ -178,4 +225,52 @@ func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowex } return nil }) + return runID, err +} + +func workflowRunStatus(status string, errorMessage string) int { + if strings.TrimSpace(errorMessage) != "" || strings.TrimSpace(status) == "error" { + return workflowRunStatusFailed + } + switch strings.TrimSpace(status) { + case "interrupted": + return workflowRunStatusInterrupted + default: + return workflowRunStatusCompleted + } +} + +func fallbackWorkflowNodeTraces(nodePath []string, nodeTypes map[string]string, status string) []workflowexecutor.NodeTrace { + ret := make([]workflowexecutor.NodeTrace, 0, len(nodePath)) + for _, nodeID := range nodePath { + ret = append(ret, workflowexecutor.NodeTrace{ + NodeID: nodeID, + NodeType: nodeTypes[nodeID], + Status: status, + }) + } + return ret +} + +func firstWorkflowInterruptType(result *workflowexecutor.Result) string { + if result == nil || len(result.Interrupts) == 0 { + return "" + } + return strings.TrimSpace(result.Interrupts[0].Type) +} + +func firstWorkflowInterruptNodeID(result *workflowexecutor.Result) string { + if result == nil || len(result.Interrupts) == 0 { + return "" + } + return strings.TrimSpace(result.Interrupts[0].ID) +} + +func firstNonEmpty(items ...string) string { + for _, item := range items { + if strings.TrimSpace(item) != "" { + return strings.TrimSpace(item) + } + } + return "" } diff --git a/internal/ai/application/runtime/tool_catalog_test.go b/internal/ai/application/runtime/tool_catalog_test.go index 88aabe2..70bab59 100644 --- a/internal/ai/application/runtime/tool_catalog_test.go +++ b/internal/ai/application/runtime/tool_catalog_test.go @@ -199,6 +199,19 @@ func TestServiceRunExecutesPublishedWorkflow(t *testing.T) { if nodeRunCount != 4 { t.Fatalf("expected four workflow node runs, got %d", nodeRunCount) } + var replyNodeRun models.AIWorkflowNodeRun + if err := sqls.DB().First(&replyNodeRun, "node_id = ?", "reply_1").Error; err != nil { + t.Fatalf("find reply node run: %v", err) + } + if replyNodeRun.InputPreview == "" || replyNodeRun.OutputPreview == "" { + t.Fatalf("expected node input/output previews, got input=%q output=%q", replyNodeRun.InputPreview, replyNodeRun.OutputPreview) + } + if !strings.Contains(replyNodeRun.OutputPreview, "workflow reply") { + t.Fatalf("expected reply output preview, got %q", replyNodeRun.OutputPreview) + } + if replyNodeRun.DurationMS < 0 { + t.Fatalf("unexpected negative duration: %d", replyNodeRun.DurationMS) + } } func setupWorkflowRuntimeTestDB(t *testing.T) { diff --git a/internal/ai/application/runtime/types.go b/internal/ai/application/runtime/types.go index 0ec1832..27c9d7a 100644 --- a/internal/ai/application/runtime/types.go +++ b/internal/ai/application/runtime/types.go @@ -16,6 +16,7 @@ type Request struct { type ResumeRequest struct { Conversation models.Conversation + UserMessage models.Message AIAgent models.AIAgent AIConfig models.AIConfig CheckPointID string @@ -46,6 +47,10 @@ type Summary struct { ToolCallCount int ToolCodes []string InvokedToolCodes []string + WorkflowID int64 + WorkflowVersionID int64 + WorkflowRunID int64 + WorkflowNodePath []string CheckPointID string CheckPointData string Interrupted bool diff --git a/internal/ai/application/runtime/workflow_summary_test.go b/internal/ai/application/runtime/workflow_summary_test.go index f83cac9..166b943 100644 --- a/internal/ai/application/runtime/workflow_summary_test.go +++ b/internal/ai/application/runtime/workflow_summary_test.go @@ -27,7 +27,7 @@ func TestToWorkflowSummaryPreservesInterruptCheckpoint(t *testing.T) { Interrupts: []workflowexecutor.InterruptSummary{ {Type: "human_confirm", ID: "confirm_1", InfoPreview: `{"message":"请确认"}`}, }, - }, "test-model") + }, "test-model", resolvedWorkflow{WorkflowID: 11, VersionID: 22}, 33) if summary == nil || !summary.Interrupted { t.Fatalf("expected interrupted summary, got %#v", summary) @@ -38,6 +38,9 @@ func TestToWorkflowSummaryPreservesInterruptCheckpoint(t *testing.T) { if summary.CheckPointData == "" { t.Fatalf("expected checkpoint data") } + if summary.WorkflowID != 11 || summary.WorkflowVersionID != 22 || summary.WorkflowRunID != 33 { + t.Fatalf("unexpected workflow identity: workflow=%d version=%d run=%d", summary.WorkflowID, summary.WorkflowVersionID, summary.WorkflowRunID) + } if len(summary.Interrupts) != 1 || summary.Interrupts[0].ID != "confirm_1" { t.Fatalf("unexpected interrupts: %#v", summary.Interrupts) } @@ -69,6 +72,7 @@ func TestServiceResumeUsesWorkflowCheckpointData(t *testing.T) { summary, err := NewService().Resume(context.Background(), ResumeRequest{ Conversation: models.Conversation{ID: 1}, + UserMessage: models.Message{ID: 2, Content: "确认"}, AIAgent: models.AIAgent{ ID: 1, WorkflowVersionID: version.ID, @@ -85,6 +89,66 @@ func TestServiceResumeUsesWorkflowCheckpointData(t *testing.T) { if summary == nil || summary.Status != "completed" || summary.Interrupted { t.Fatalf("unexpected summary: %#v", summary) } + if summary.WorkflowRunID <= 0 { + t.Fatalf("expected workflow run id in resume summary") + } + var run models.AIWorkflowRun + if err := db.First(&run, summary.WorkflowRunID).Error; err != nil { + t.Fatalf("find resume workflow run: %v", err) + } + if run.MessageID != 2 || run.Status != workflowRunStatusCompleted { + t.Fatalf("unexpected resume workflow run: %#v", run) + } +} + +func TestServiceRunWritesFailedWorkflowRun(t *testing.T) { + db := setupWorkflowResumeTestDB(t) + def := dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, + {ID: "bad_1", Type: "unsupported_node", Name: "Bad"}, + }, + Edges: []dsl.Edge{ + {ID: "edge_start_bad", Source: "start_1", Target: "bad_1"}, + }, + } + version := models.AIWorkflowVersion{ + WorkflowID: 9, + Version: 1, + Status: enums.StatusOk, + Definition: mustMarshalDefinition(t, def), + } + if err := db.Create(&version).Error; err != nil { + t.Fatalf("create workflow version: %v", err) + } + + _, err := NewService().Run(context.Background(), Request{ + Conversation: models.Conversation{ID: 10}, + UserMessage: models.Message{ID: 20, Content: "hello"}, + AIAgent: models.AIAgent{ + ID: 30, + WorkflowVersionID: version.ID, + }, + }) + if err == nil { + t.Fatalf("expected workflow run error") + } + var run models.AIWorkflowRun + if err := db.First(&run, "workflow_version_id = ?", version.ID).Error; err != nil { + t.Fatalf("find failed workflow run: %v", err) + } + if run.Status != workflowRunStatusFailed || !strings.Contains(run.ErrorMessage, "unsupported workflow node type") { + t.Fatalf("unexpected failed workflow run: %#v", run) + } + var badNodeRun models.AIWorkflowNodeRun + if err := db.First(&badNodeRun, "workflow_run_id = ? AND node_id = ?", run.ID, "bad_1").Error; err != nil { + t.Fatalf("find failed node run: %v", err) + } + if badNodeRun.Status != workflowRunStatusFailed || badNodeRun.ErrorMessage == "" { + t.Fatalf("unexpected failed node run: %#v", badNodeRun) + } } func setupWorkflowResumeTestDB(t *testing.T) *gorm.DB { @@ -105,7 +169,7 @@ func setupWorkflowResumeTestDB(t *testing.T) *gorm.DB { _ = sqlDB.Close() } }) - if err := db.AutoMigrate(&models.AIWorkflowVersion{}, &models.ConversationInterrupt{}); err != nil { + if err := db.AutoMigrate(&models.AIWorkflowVersion{}, &models.AIWorkflowRun{}, &models.AIWorkflowNodeRun{}, &models.ConversationInterrupt{}); err != nil { t.Fatalf("auto migrate: %v", err) } sqls.SetDB(db) diff --git a/internal/ai/runtime/reply_helpers_test.go b/internal/ai/runtime/reply_helpers_test.go index e398f31..a22786e 100644 --- a/internal/ai/runtime/reply_helpers_test.go +++ b/internal/ai/runtime/reply_helpers_test.go @@ -27,6 +27,10 @@ func TestSummaryPrimaryToolCodePrefersToolSearchTarget(t *testing.T) { } func TestToRunLogFinalAction(t *testing.T) { + if got := toRunLogFinalAction(&applicationruntime.Summary{WorkflowVersionID: 66, ReplyText: "ok"}); got != "workflow_reply" { + t.Fatalf("expected workflow_reply final action, got %q", got) + } + if got := toRunLogFinalAction(&applicationruntime.Summary{PlannedSkillID: 44, ReplyText: "ok"}); got != "skill" { t.Fatalf("expected skill final action, got %q", got) } @@ -50,6 +54,22 @@ func TestToRunLogFinalAction(t *testing.T) { } } +func TestBuildRunLogPlanUsesWorkflowSummary(t *testing.T) { + plannedAction, plannedToolCode, planReason := buildRunLogPlan(&applicationruntime.Summary{ + WorkflowVersionID: 66, + Status: "completed", + }) + if plannedAction != "workflow" { + t.Fatalf("expected workflow planned action, got %q", plannedAction) + } + if plannedToolCode != "workflow/66" { + t.Fatalf("expected workflow planned tool code, got %q", plannedToolCode) + } + if planReason == "" { + t.Fatalf("expected plan reason") + } +} + func TestExtractInterruptMessageAndCheckpointError(t *testing.T) { if got := extractInterruptMessage(`{"message":"请补充订单号"}`); got != "请补充订单号" { t.Fatalf("unexpected interrupt message: %q", got) diff --git a/internal/ai/runtime/reply_runlog_service.go b/internal/ai/runtime/reply_runlog_service.go index 50a448f..a088ca9 100644 --- a/internal/ai/runtime/reply_runlog_service.go +++ b/internal/ai/runtime/reply_runlog_service.go @@ -3,6 +3,7 @@ package runtime import ( "encoding/json" "log/slog" + "strconv" "strings" "time" @@ -90,6 +91,9 @@ func buildRunLogPlan(summary *applicationruntime.Summary) (plannedAction, planne if summary == nil { return "", "", "" } + if isWorkflowSummary(summary) { + return "workflow", workflowPlannedToolCode(summary), "workflow executed" + } if summaryPlannedSkillID(summary) > 0 { reason := strings.TrimSpace(summary.PlanReason) if reason == "" { @@ -138,6 +142,9 @@ func toRunLogFinalAction(summary *applicationruntime.Summary) string { if summary == nil { return "" } + if isWorkflowSummary(summary) { + return workflowFinalAction(summary) + } if summaryPlannedSkillID(summary) > 0 && strings.TrimSpace(summary.ReplyText) != "" { return "skill" } @@ -160,6 +167,35 @@ func toRunLogFinalAction(summary *applicationruntime.Summary) string { } } +func isWorkflowSummary(summary *applicationruntime.Summary) bool { + return summary != nil && summary.WorkflowVersionID > 0 +} + +func workflowPlannedToolCode(summary *applicationruntime.Summary) string { + if summary == nil || summary.WorkflowVersionID <= 0 { + return "" + } + return "workflow/" + strconv.FormatInt(summary.WorkflowVersionID, 10) +} + +func workflowFinalAction(summary *applicationruntime.Summary) string { + if summary == nil { + return "" + } + switch strings.TrimSpace(summary.Status) { + case "interrupted": + return "workflow_interrupted" + case "error": + return "workflow_error" + case "expired": + return "workflow_expired" + } + if strings.TrimSpace(summary.ReplyText) != "" { + return "workflow_reply" + } + return "workflow_completed" +} + func buildRunLogReplyText(summary *applicationruntime.Summary) string { if summary == nil { return "" diff --git a/internal/ai/runtime/runtime_reply_executor.go b/internal/ai/runtime/runtime_reply_executor.go index 4bd26c6..67b6bd3 100644 --- a/internal/ai/runtime/runtime_reply_executor.go +++ b/internal/ai/runtime/runtime_reply_executor.go @@ -67,6 +67,7 @@ func (e *runtimeReplyExecutor) ResumePendingInterrupt(ctx context.Context, input } summary, err := Service.Resume(ctx, applicationruntime.ResumeRequest{ Conversation: input.Conversation, + UserMessage: input.Message, AIAgent: input.AIAgent, AIConfig: *aiConfig, CheckPointID: strings.TrimSpace(input.PendingInterrupt.CheckPointID), diff --git a/internal/ai/runtime/workflow/executor.go b/internal/ai/runtime/workflow/executor.go index 1ea50e9..bf07177 100644 --- a/internal/ai/runtime/workflow/executor.go +++ b/internal/ai/runtime/workflow/executor.go @@ -7,6 +7,7 @@ import ( "reflect" "strconv" "strings" + "time" "agent-desk/internal/ai" "agent-desk/internal/ai/runtime/graphs" @@ -34,6 +35,7 @@ type Result struct { Status string ReplyText string NodePath []string + NodeTraces []NodeTrace PromptTokens int CompletionTokens int RetrieverCount int @@ -44,6 +46,16 @@ type Result struct { Interrupts []InterruptSummary } +type NodeTrace struct { + NodeID string + NodeType string + Status string + InputPreview string + OutputPreview string + ErrorMessage string + DurationMS int +} + type InterruptSummary struct { Type string ID string @@ -127,23 +139,44 @@ func (e *Executor) executeFrom(ctx context.Context, state *runState, currentID s for step := 0; step < maxWorkflowSteps; step++ { node, ok := state.nodesByID[currentID] if !ok { - return nil, fmt.Errorf("workflow node does not exist: %s", currentID) + err := fmt.Errorf("workflow node does not exist: %s", currentID) + state.result.Status = "error" + return &state.result, err } state.result.NodePath = append(state.result.NodePath, node.ID) - if err := e.executeNode(ctx, state, node); err != nil { - return nil, err + trace := NodeTrace{ + NodeID: node.ID, + NodeType: node.Type, + Status: "running", + InputPreview: workflowPreviewJSON(state.nodeInputPreview(node)), } + startedAt := time.Now() + if err := e.executeNode(ctx, state, node); err != nil { + trace.Status = "failed" + trace.ErrorMessage = err.Error() + trace.DurationMS = int(time.Since(startedAt).Milliseconds()) + state.result.NodeTraces = append(state.result.NodeTraces, trace) + state.result.Status = "error" + return &state.result, err + } + trace.OutputPreview = workflowPreviewJSON(state.vars[node.ID]) + trace.DurationMS = int(time.Since(startedAt).Milliseconds()) if state.result.Interrupted { + trace.Status = "interrupted" + state.result.NodeTraces = append(state.result.NodeTraces, trace) state.result.Status = "interrupted" return &state.result, nil } + trace.Status = "completed" + state.result.NodeTraces = append(state.result.NodeTraces, trace) if node.Type == workflowregistry.NodeTypeEnd { state.result.Status = "completed" return &state.result, nil } nextID, ok, err := state.nextNodeID(node.ID) if err != nil { - return nil, err + state.result.Status = "error" + return &state.result, err } if !ok { state.result.Status = "completed" @@ -151,7 +184,9 @@ func (e *Executor) executeFrom(ctx context.Context, state *runState, currentID s } currentID = nextID } - return nil, fmt.Errorf("workflow exceeded max steps") + err := fmt.Errorf("workflow exceeded max steps") + state.result.Status = "error" + return &state.result, err } func newRunState(input Input) *runState { @@ -161,8 +196,9 @@ func newRunState(input Input) *runState { outgoing: make(map[string][]dsl.Edge), vars: make(map[string]map[string]any), result: Result{ - Status: "started", - NodePath: make([]string, 0), + Status: "started", + NodePath: make([]string, 0), + NodeTraces: make([]NodeTrace, 0), }, } for _, node := range input.Definition.Nodes { @@ -553,6 +589,37 @@ func (s *runState) resolveInput(node dsl.Node, inputName string) any { return s.resolveSelector(&selector) } +func (s *runState) nodeInputPreview(node dsl.Node) map[string]any { + inputs := make(map[string]any, len(node.Inputs)) + for name, selector := range node.Inputs { + inputs[name] = s.resolveSelector(&selector) + } + ret := map[string]any{ + "inputs": inputs, + } + if len(node.Config) > 0 { + var cfg any + if err := json.Unmarshal(node.Config, &cfg); err == nil { + ret["config"] = cfg + } else { + ret["config"] = string(node.Config) + } + } + return ret +} + +func workflowPreviewJSON(value any) string { + raw, err := json.Marshal(value) + if err != nil { + return "" + } + const maxPreviewBytes = 2000 + if len(raw) <= maxPreviewBytes { + return string(raw) + } + return string(raw[:maxPreviewBytes]) +} + func (s *runState) resolveSelector(selector *dsl.VariableSelector) any { if selector == nil { return nil