From 2cb338ae1278996f8e54ba75ab8e1c6638eadc4d Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 31 Aug 2026 11:21:53 -0400 Subject: [PATCH 1/8] fix(control-plane): bound golden-run name and tag writes into run metadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /api/ui/v2/workflow-runs/:run_id/golden wrote the caller-supplied name and tag list into workflow_runs.metadata with no bounds at all. The name was only TrimSpace'd, so a 1 MiB name persisted verbatim; sanitizeStringList trimmed and de-duped but capped neither the entry count nor the entry length, and it preallocated its output slice (and an unbounded de-dupe map) straight from the attacker-controlled input length. That row is re-read and re-serialised on every runs-list page that contains the run, so both are stored amplification vectors (#944). Cap tags at 20 entries of at most 64 runes each, and truncate the name at 200 runes. Over-long tags are dropped rather than truncated: byte-slicing can land mid-rune and json.Marshal silently rewrites the invalid UTF-8 to U+FFFD. Lengths are counted with utf8.RuneCountInString so a 64-rune CJK tag survives. The output slice and de-dupe map are now sized min(len(values), maxCount) and the loop stops once maxCount survivors are collected, so a multi-million-entry tag array cannot force a large allocation before the cap applies. This route is UI-private and its only caller sends one hard-coded tag, so oversized input is bounded silently rather than rejected — a 400 would break the existing "Save as golden run" button. The name fallback is unchanged: an empty name still falls back to run_id, and run_id itself is not truncated. Forward-only. Nothing re-validates on read, so rows that already hold oversized golden metadata keep reading back exactly as they do today. The caps are named constants so the follow-up run-metadata endpoint can reuse the same bounds and the same helper. Co-Authored-By: Claude Fable 5 --- .../internal/handlers/ui/workflow_runs.go | 66 +++++++++++++++++-- 1 file changed, 61 insertions(+), 5 deletions(-) diff --git a/control-plane/internal/handlers/ui/workflow_runs.go b/control-plane/internal/handlers/ui/workflow_runs.go index ad5e98342..8dea2e327 100644 --- a/control-plane/internal/handlers/ui/workflow_runs.go +++ b/control-plane/internal/handlers/ui/workflow_runs.go @@ -9,6 +9,7 @@ import ( "strconv" "strings" "time" + "unicode/utf8" "github.com/Agent-Field/agentfield/control-plane/internal/handlers" "github.com/Agent-Field/agentfield/control-plane/internal/logger" @@ -80,6 +81,19 @@ type GoldenRunMetadata struct { SavedAt string `json:"saved_at,omitempty"` } +// Bounds on the golden-run metadata a client may write into +// workflow_runs.metadata. This row is re-read and re-serialised on every +// runs-list page that contains the run, so an unbounded name or tag list is +// a stored amplification vector (issue #944). The UI's save-as-golden button +// sends one hard-coded tag, so these caps are unreachable in practice today. +// They are package-level constants rather than inline literals so the +// run-metadata endpoint can reuse the same bounds. +const ( + maxGoldenTags = 20 + maxGoldenTagRunes = 64 + maxGoldenNameRunes = 200 +) + type WorkflowRunListResponse struct { Runs []WorkflowRunSummary `json:"runs"` TotalCount int `json:"total_count"` @@ -261,8 +275,9 @@ func (h *WorkflowRunHandler) SaveGoldenRunHandler(c *gin.Context) { } now := time.Now().UTC() - name := strings.TrimSpace(req.Name) + name := truncateRunes(strings.TrimSpace(req.Name), maxGoldenNameRunes) if name == "" { + // Do not truncate the run ID fallback; only the caller-supplied name is bounded. name = runID } metadata := map[string]interface{}{} @@ -271,7 +286,7 @@ func (h *WorkflowRunHandler) SaveGoldenRunHandler(c *gin.Context) { } metadata["golden"] = GoldenRunMetadata{ Name: name, - Tags: sanitizeStringList(req.Tags), + Tags: sanitizeStringList(req.Tags, maxGoldenTags, maxGoldenTagRunes), SavedBy: "user", SavedAt: now.Format(time.RFC3339), } @@ -645,14 +660,37 @@ func decodeWorkflowRunMetadata(raw json.RawMessage) map[string]interface{} { return metadata } -func sanitizeStringList(values []string) []string { - out := make([]string, 0, len(values)) - seen := map[string]struct{}{} +// sanitizeStringList trims, de-duplicates and bounds a caller-supplied list of +// short strings, preserving input order. +// +// Entries longer than maxRunes runes are DROPPED rather than truncated: cutting +// a string at a byte offset can land mid-rune, and json.Marshal silently +// rewrites the resulting invalid UTF-8 to U+FFFD. Rune counting (not len) +// keeps a maxRunes-rune multi-byte tag legal. +// +// At most maxCount entries are returned. Both the output slice and the de-dupe +// map are sized from min(len(values), maxCount) and the loop stops once +// maxCount survivors are collected, so a huge attacker-supplied array cannot +// force a large allocation before the cap applies. maxRunes <= 0 disables the +// length cap; maxCount <= 0 yields an empty result. +func sanitizeStringList(values []string, maxCount, maxRunes int) []string { + if maxCount <= 0 { + return nil + } + size := min(len(values), maxCount) + out := make([]string, 0, size) + seen := make(map[string]struct{}, size) for _, value := range values { + if len(out) >= maxCount { + break + } trimmed := strings.TrimSpace(value) if trimmed == "" { continue } + if maxRunes > 0 && utf8.RuneCountInString(trimmed) > maxRunes { + continue + } if _, ok := seen[trimmed]; ok { continue } @@ -662,6 +700,24 @@ func sanitizeStringList(values []string) []string { return out } +// truncateRunes returns value limited to at most maxRunes runes, cutting on a +// rune boundary so the result is always valid UTF-8. It walks the string by +// rune start offsets instead of materialising a []rune, so a multi-megabyte +// input costs no extra allocation. +func truncateRunes(value string, maxRunes int) string { + if maxRunes <= 0 { + return "" + } + count := 0 + for i := range value { + if count == maxRunes { + return value[:i] + } + count++ + } + return value +} + func deriveOverallStatusForUI(executions []*types.Execution) string { counts := make(map[string]int) active := 0 From 54399838146e4f5154b4806521ef53fd446a5a76 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 31 Aug 2026 11:22:04 -0400 Subject: [PATCH 2/8] test(control-plane): cover golden-run metadata bounds and forward-only reads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the behaviour tests for the golden-run caps: - sanitizeStringList: 100 inputs cap to the first 20; a 65-rune ASCII entry is dropped while its 64-rune neighbour survives; a 64-rune CJK entry is kept byte-identical (proving it is not truncated into U+FFFD) while a 65-rune one is dropped; the legacy trim/de-dupe/order behaviour is pinned unchanged; the maxCount<=0 and maxRunes<=0 guards are exercised. - A 100k-entry input asserts cap(out) <= 20, which is the allocation contract — a length assertion alone would still pass with the old preallocation. - truncateRunes: cut on a rune boundary, result always valid UTF-8 with no replacement character, and the non-positive cap guard. - Handler round-trip: a POST with 50 tags plus one over-long tag stores exactly 20, a 1 MiB name stores 200 runes and keeps the whole metadata blob under 4 KiB, and a blank name still falls back to run_id. - Read-back: a pre-seeded row holding 50 tags and a 500-rune name still surfaces in full on both the runs list and the run detail, pinning that this change is forward-only and read paths do not re-validate. The two pre-existing golden-route tests are left untouched as the behaviour-unchanged regression guard. Co-Authored-By: Claude Fable 5 --- ...verage_execution_workflow_handlers_test.go | 194 ++++++++++++++++++ 1 file changed, 194 insertions(+) diff --git a/control-plane/internal/handlers/ui/coverage_execution_workflow_handlers_test.go b/control-plane/internal/handlers/ui/coverage_execution_workflow_handlers_test.go index d5945511d..187c4cadf 100644 --- a/control-plane/internal/handlers/ui/coverage_execution_workflow_handlers_test.go +++ b/control-plane/internal/handlers/ui/coverage_execution_workflow_handlers_test.go @@ -7,9 +7,11 @@ import ( "net/http" "net/http/httptest" "path/filepath" + "strconv" "strings" "testing" "time" + "unicode/utf8" "github.com/Agent-Field/agentfield/control-plane/internal/storage" "github.com/Agent-Field/agentfield/control-plane/pkg/types" @@ -504,6 +506,198 @@ func TestWorkflowRunHandlerSaveGoldenRunPreservesLineageMetadata(t *testing.T) { require.Contains(t, metadata, "golden") } +func TestSanitizeStringListBounds(t *testing.T) { + distinct := make([]string, 100) + for i := range distinct { + distinct[i] = "tag-" + strconv.Itoa(i) + } + longASCII := strings.Repeat("a", 65) + maxASCII := strings.Repeat("a", 64) + maxCJK := strings.Repeat("漢", 64) + longCJK := strings.Repeat("漢", 65) + veryLong := strings.Repeat("z", 1000) + + tests := []struct { + name string + values []string + maxCount int + maxRunes int + want []string + wantEmpty bool + }{ + {name: "count cap preserves first entries", values: distinct, maxCount: 20, maxRunes: 64, want: distinct[:20]}, + {name: "overlong ASCII dropped", values: []string{longASCII, maxASCII}, maxCount: 20, maxRunes: 64, want: []string{maxASCII}}, + {name: "multibyte rune cap", values: []string{maxCJK, longCJK}, maxCount: 20, maxRunes: 64, want: []string{maxCJK}}, + {name: "legacy sanitizing", values: []string{" smoke ", "smoke", "", " ", "restart"}, maxCount: 20, maxRunes: 64, want: []string{"smoke", "restart"}}, + {name: "nil input", values: nil, maxCount: 20, maxRunes: 64, wantEmpty: true}, + {name: "zero count", values: []string{"tag"}, maxCount: 0, maxRunes: 64, wantEmpty: true}, + {name: "zero runes disables length cap", values: []string{veryLong}, maxCount: 20, maxRunes: 0, want: []string{veryLong}}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := sanitizeStringList(tc.values, tc.maxCount, tc.maxRunes) + if tc.wantEmpty { + require.Empty(t, got) + return + } + require.Equal(t, tc.want, got) + }) + } + + require.Equal(t, 64, utf8.RuneCountInString(maxCJK)) + require.Equal(t, maxCJK, sanitizeStringList([]string{maxCJK}, 20, 64)[0]) +} + +func TestSanitizeStringListDoesNotPreallocateFromInputLength(t *testing.T) { + values := make([]string, 100000) + for i := range values { + values[i] = "tag-" + strconv.Itoa(i) + } + + out := sanitizeStringList(values, 20, 64) + require.Len(t, out, 20) + require.LessOrEqual(t, cap(out), 20) +} + +func TestTruncateRunes(t *testing.T) { + tests := []struct { + name string + value string + maxRunes int + want string + }{ + {name: "shorter than cap", value: "short", maxRunes: 10, want: "short"}, + {name: "exactly at cap", value: "exact", maxRunes: 5, want: "exact"}, + {name: "longer ASCII", value: "abcdef", maxRunes: 5, want: "abcde"}, + {name: "multibyte boundary", value: "漢字仮名", maxRunes: 3, want: "漢字仮"}, + {name: "zero cap", value: "value", maxRunes: 0, want: ""}, + {name: "negative cap", value: "value", maxRunes: -1, want: ""}, + {name: "empty input", value: "", maxRunes: 5, want: ""}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := truncateRunes(tc.value, tc.maxRunes) + require.Equal(t, tc.want, got) + require.True(t, utf8.ValidString(got)) + require.NotContains(t, got, "�") + if utf8.RuneCountInString(tc.value) > tc.maxRunes && tc.maxRunes > 0 { + require.Equal(t, tc.maxRunes, utf8.RuneCountInString(got)) + } + }) + } +} + +func TestWorkflowRunHandlerSaveGoldenRunBoundsNameAndTags(t *testing.T) { + gin.SetMode(gin.TestMode) + ls, ctx := setupUIHandlerStorage(t) + runID := "run-golden-bounds" + executionID := "exec-golden-bounds" + now := time.Date(2026, 4, 10, 12, 0, 0, 0, time.UTC) + completed := now.Add(time.Second) + require.NoError(t, ls.CreateExecutionRecord(ctx, &types.Execution{ + ExecutionID: executionID, RunID: runID, AgentNodeID: "agent-alpha", NodeID: "agent-alpha", + ReasonerID: "planner", Status: types.ExecutionStatusSucceeded, InputPayload: json.RawMessage(`{"input":{}}`), + StartedAt: now, CompletedAt: &completed, CreatedAt: now, UpdatedAt: completed, + })) + + handler := NewWorkflowRunHandler(ls) + router := gin.New() + router.POST("/api/ui/v1/workflow-runs/:run_id/golden", handler.SaveGoldenRunHandler) + tags := make([]string, 0, 51) + for i := 0; i < 50; i++ { + tags = append(tags, "tag-"+strconv.Itoa(i)) + } + overlongTag := strings.Repeat("漢", 65) + tags = append([]string{overlongTag}, tags...) + payload, err := json.Marshal(saveGoldenRunRequest{Name: strings.Repeat("a", 1<<20), Tags: tags}) + require.NoError(t, err) + + req := httptest.NewRequest(http.MethodPost, "/api/ui/v1/workflow-runs/"+runID+"/golden", strings.NewReader(string(payload))) + req.Header.Set("Content-Type", "application/json") + resp := httptest.NewRecorder() + router.ServeHTTP(resp, req) + require.Equal(t, http.StatusOK, resp.Code) + var summary WorkflowRunSummary + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &summary)) + require.NotNil(t, summary.Golden) + require.Len(t, summary.Golden.Tags, 20) + require.NotContains(t, summary.Golden.Tags, overlongTag) + + run, err := ls.GetWorkflowRun(ctx, runID) + require.NoError(t, err) + require.Less(t, len(run.Metadata), 4096) + metadata := decodeWorkflowRunMetadata(run.Metadata) + encodedGolden, err := json.Marshal(metadata["golden"]) + require.NoError(t, err) + var golden GoldenRunMetadata + require.NoError(t, json.Unmarshal(encodedGolden, &golden)) + require.Len(t, golden.Tags, 20) + require.NotEmpty(t, golden.Name) + require.Equal(t, maxGoldenNameRunes, utf8.RuneCountInString(golden.Name)) + + emptyNameReq := httptest.NewRequest(http.MethodPost, "/api/ui/v1/workflow-runs/"+runID+"/golden", strings.NewReader(`{"name":" "}`)) + emptyNameReq.Header.Set("Content-Type", "application/json") + emptyNameResp := httptest.NewRecorder() + router.ServeHTTP(emptyNameResp, emptyNameReq) + require.Equal(t, http.StatusOK, emptyNameResp.Code) + parsed := handler.loadRunMetadata(ctx, runID) + require.NotNil(t, parsed) + require.NotNil(t, parsed.Golden) + require.Equal(t, runID, parsed.Golden.Name) +} + +func TestWorkflowRunHandlerGoldenReadBackIsNotRevalidated(t *testing.T) { + gin.SetMode(gin.TestMode) + ls, ctx := setupUIHandlerStorage(t) + runID := "run-golden-existing-oversized" + executionID := "exec-golden-existing-oversized" + now := time.Date(2026, 4, 11, 12, 0, 0, 0, time.UTC) + completed := now.Add(time.Second) + tags := make([]string, 50) + for i := range tags { + tags[i] = "legacy-tag-" + strconv.Itoa(i) + } + name := strings.Repeat("漢", 500) + metadata, err := json.Marshal(map[string]interface{}{"golden": GoldenRunMetadata{Name: name, Tags: tags}}) + require.NoError(t, err) + require.NoError(t, ls.StoreWorkflowRun(ctx, &types.WorkflowRun{ + RunID: runID, RootWorkflowID: runID, RootExecutionID: &executionID, + Status: string(types.ExecutionStatusSucceeded), TotalSteps: 1, CompletedSteps: 1, + Metadata: metadata, CreatedAt: now, UpdatedAt: completed, + })) + require.NoError(t, ls.CreateExecutionRecord(ctx, &types.Execution{ + ExecutionID: executionID, RunID: runID, AgentNodeID: "agent-alpha", NodeID: "agent-alpha", + ReasonerID: "planner", Status: types.ExecutionStatusSucceeded, InputPayload: json.RawMessage(`{"input":{}}`), + StartedAt: now, CompletedAt: &completed, CreatedAt: now, UpdatedAt: completed, + })) + + handler := NewWorkflowRunHandler(ls) + router := gin.New() + router.GET("/api/ui/v2/workflow-runs", handler.ListWorkflowRunsHandler) + router.GET("/api/ui/v2/workflow-runs/:run_id", handler.GetWorkflowRunDetailHandler) + + listResp := httptest.NewRecorder() + router.ServeHTTP(listResp, httptest.NewRequest(http.MethodGet, "/api/ui/v2/workflow-runs?page_size=200", nil)) + require.Equal(t, http.StatusOK, listResp.Code) + var list WorkflowRunListResponse + require.NoError(t, json.Unmarshal(listResp.Body.Bytes(), &list)) + require.Len(t, list.Runs, 1) + require.NotNil(t, list.Runs[0].Golden) + require.Equal(t, name, list.Runs[0].Golden.Name) + require.Len(t, list.Runs[0].Golden.Tags, 50) + + detailResp := httptest.NewRecorder() + router.ServeHTTP(detailResp, httptest.NewRequest(http.MethodGet, "/api/ui/v2/workflow-runs/"+runID, nil)) + require.Equal(t, http.StatusOK, detailResp.Code) + var detail WorkflowRunDetailResponse + require.NoError(t, json.Unmarshal(detailResp.Body.Bytes(), &detail)) + require.NotNil(t, detail.Run.Golden) + require.Equal(t, name, detail.Run.Golden.Name) + require.Len(t, detail.Run.Golden.Tags, 50) +} + func TestWorkflowRunHandlerSaveGoldenRunErrors(t *testing.T) { gin.SetMode(gin.TestMode) From 43f8ae770110c5cb51f65e677114924c6e377e1a Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 31 Aug 2026 14:45:04 -0400 Subject: [PATCH 3/8] feat(control-plane): client-settable run display name, labels and links A 'run' namespace in workflow_runs.metadata, written only through a namespace-merging transactional primitive (never the full-row upsert, which clobbers golden/lineage and resets state columns). POST /api/v1/runs/:run_id/metadata read-merge-writes it with strict caps (display_name<=200, labels<=20x64, links<=10, url<=2048, http/https only, no embedded credentials) and creates the carrier row on first write; an optional run_metadata execute field seeds it at dispatch, excluded from the replay dedupe key; restart lineage now writes through the same primitive. The run list, run detail, DAG and agentic overview surface it. external_status is deliberately out of scope. Co-Authored-By: Claude Fable 5 --- .../internal/handlers/agentic/run_overview.go | 17 +- control-plane/internal/handlers/execute.go | 3 + .../internal/handlers/execute_prepare.go | 68 +++++++- .../internal/handlers/execute_restart.go | 48 +++--- control-plane/internal/handlers/mcp.go | 1 + .../internal/handlers/run_metadata.go | 162 ++++++++++++++++++ .../internal/handlers/ui/workflow_runs.go | 23 ++- .../internal/handlers/workflow_dag.go | 28 +-- .../server/apicatalog/catalog_entries.go | 6 + control-plane/internal/server/routes_core.go | 1 + control-plane/internal/storage/local.go | 91 ++++++++++ control-plane/pkg/types/run_metadata.go | 77 +++++++++ 12 files changed, 467 insertions(+), 58 deletions(-) create mode 100644 control-plane/internal/handlers/run_metadata.go create mode 100644 control-plane/pkg/types/run_metadata.go diff --git a/control-plane/internal/handlers/agentic/run_overview.go b/control-plane/internal/handlers/agentic/run_overview.go index af53418ea..78b4fdfcb 100644 --- a/control-plane/internal/handlers/agentic/run_overview.go +++ b/control-plane/internal/handlers/agentic/run_overview.go @@ -1,6 +1,7 @@ package agentic import ( + "context" "net/http" "github.com/Agent-Field/agentfield/control-plane/internal/storage" @@ -8,6 +9,10 @@ import ( "github.com/gin-gonic/gin" ) +type workflowRunGetter interface { + GetWorkflowRun(context.Context, string) (*types.WorkflowRun, error) +} + // RunOverviewHandler returns everything about a workflow run in one call. func RunOverviewHandler(store storage.StorageProvider) gin.HandlerFunc { return func(c *gin.Context) { @@ -55,7 +60,7 @@ func RunOverviewHandler(store storage.StorageProvider) gin.HandlerFunc { agents = append(agents, a) } - respondOK(c, gin.H{ + payload := gin.H{ "run_id": runID, "executions": executions, "agents": agents, @@ -65,6 +70,14 @@ func RunOverviewHandler(store storage.StorageProvider) gin.HandlerFunc { "unique_agents": len(agents), }, "notes": allNotes, - }) + } + if getter, ok := store.(workflowRunGetter); ok { + if run, err := getter.GetWorkflowRun(ctx, runID); err == nil && run != nil { + if metadata := types.ParseRunMetadata(run.Metadata); metadata != nil { + payload["run_metadata"] = metadata + } + } + } + respondOK(c, payload) } } diff --git a/control-plane/internal/handlers/execute.go b/control-plane/internal/handlers/execute.go index b04893572..119b74949 100644 --- a/control-plane/internal/handlers/execute.go +++ b/control-plane/internal/handlers/execute.go @@ -43,6 +43,9 @@ type ExecuteRequest struct { Input map[string]interface{} `json:"input"` Context map[string]interface{} `json:"context,omitempty"` Webhook *WebhookRequest `json:"webhook,omitempty"` + // RunMetadata names, labels or links the run started by this request. It is + // excluded from replay dedupe and ignored on child executions. + RunMetadata *RunMetadataInput `json:"run_metadata,omitempty"` } // WebhookRequest represents webhook registration parameters supplied by the client. diff --git a/control-plane/internal/handlers/execute_prepare.go b/control-plane/internal/handlers/execute_prepare.go index 8430891d2..4ca2796f6 100644 --- a/control-plane/internal/handlers/execute_prepare.go +++ b/control-plane/internal/handlers/execute_prepare.go @@ -55,6 +55,11 @@ func (c *executionController) prepareExecutionForTargetWithAdmission(ctx context if req.Input == nil { req.Input = map[string]interface{}{} } + if req.RunMetadata != nil && headers.parentExecutionID == nil { + if _, err := applyRunMetadataInput(types.RunMetadata{}, *req.RunMetadata); err != nil { + return nil, fmt.Errorf("invalid run_metadata: %w", err) + } + } var ( sanitizedWebhook *normalizedWebhookConfig @@ -167,14 +172,7 @@ func (c *executionController) prepareExecutionForTargetWithAdmission(ctx context executionID := utils.GenerateExecutionID() now := time.Now().UTC() - clientPayload := map[string]interface{}{ - "input": req.Input, - } - if len(req.Context) > 0 { - clientPayload["context"] = req.Context - } - - storedPayload, err := json.Marshal(clientPayload) + storedPayload, err := json.Marshal(buildClientPayload(req)) if err != nil { return nil, fmt.Errorf("encode execution payload: %w", err) } @@ -222,6 +220,9 @@ func (c *executionController) prepareExecutionForTargetWithAdmission(ctx context if err := c.store.CreateExecutionRecord(ctx, exec); err != nil { return nil, fmt.Errorf("create execution record: %w", err) } + if headers.parentExecutionID == nil && req.RunMetadata != nil { + c.persistExecuteRunMetadata(ctx, runID, *req.RunMetadata, headers.actorID) + } var webhookRegistered bool if sanitizedWebhook != nil && webhookError == nil { @@ -276,6 +277,57 @@ func (c *executionController) prepareExecutionForTargetWithAdmission(ctx context }, nil } +// buildClientPayload builds the blob persisted as executions.input_payload, +// which canonicalReplayPayload then hashes into the replay dedupe key. +// +// Only input and context belong in it. run_metadata is deliberately excluded: +// it names the run for humans and has no bearing on what the reasoner is asked +// to compute, so two executes that differ only in run_metadata must still +// replay-match each other. Adding a field here changes the dedupe key for every +// caller and silently turns existing replay hits into misses. +func buildClientPayload(req ExecuteRequest) map[string]interface{} { + payload := map[string]interface{}{ + "input": req.Input, + } + if len(req.Context) > 0 { + payload["context"] = req.Context + } + return payload +} + +// persistExecuteRunMetadata stores the run_metadata a root execute carried, by +// merging it into the run's "run" namespace. Best effort on purpose: an execute +// must not fail because a display name could not be recorded, so a failure is +// logged and swallowed — the same contract persistRestartRunMetadata uses for +// the lineage seed. Callers must have already checked that this is a root +// execute; only the run root establishes run identity. +func (c *executionController) persistExecuteRunMetadata(ctx context.Context, runID string, input RunMetadataInput, actorID *string) { + writer, ok := c.store.(workflowRunMetadataWriter) + if !ok { + return + } + actor := "api" + if actorID != nil && strings.TrimSpace(*actorID) != "" { + actor = strings.TrimSpace(*actorID) + } + if err := writer.UpdateWorkflowRunMetadata(ctx, runID, func(namespaces map[string]json.RawMessage) error { + current := types.RunMetadata{} + if raw := namespaces[types.RunMetadataNamespace]; raw != nil { + _ = json.Unmarshal(raw, ¤t) + } + merged, err := applyRunMetadataInput(current, input) + if err != nil { + return err + } + merged.SetBy = actor + merged.UpdatedAt = time.Now().UTC().Format(time.RFC3339) + namespaces[types.RunMetadataNamespace], err = json.Marshal(merged) + return err + }); err != nil { + logger.Logger.Warn().Err(err).Str("run_id", runID).Msg("failed to persist execute run metadata") + } +} + // findReplayHit returns a previously-succeeded child output to reuse for the // current app.call, or nil to run it normally. Only child executions (those with // a parent) are eligible — the restarted root always re-runs. diff --git a/control-plane/internal/handlers/execute_restart.go b/control-plane/internal/handlers/execute_restart.go index 82e7da070..eb1d67bee 100644 --- a/control-plane/internal/handlers/execute_restart.go +++ b/control-plane/internal/handlers/execute_restart.go @@ -50,7 +50,7 @@ type restartExecutionResponse struct { } type workflowRunMetadataStore interface { - StoreWorkflowRun(ctx context.Context, run *types.WorkflowRun) error + UpdateWorkflowRunMetadata(context.Context, string, func(map[string]json.RawMessage) error) error } // RestartExecutionHandler starts a new execution/run from an existing workflow @@ -151,6 +151,7 @@ func (c *executionController) handleRestart(ctx *gin.Context) { } target := fmt.Sprintf("%s.%s", restartExec.NodeID, restartExec.ReasonerID) + // Restarts mint a new run identity and deliberately leave RunMetadata nil. plan, err := c.prepareExecutionForTarget(reqCtx, target, ExecuteRequest{ Input: input, Context: contextPayload, @@ -247,40 +248,31 @@ func (c *executionController) persistRestartRunMetadata(ctx context.Context, pla if !ok { return } - now := time.Now().UTC() - metadata := map[string]interface{}{ - "lineage": map[string]interface{}{ - "kind": kind, - "source_run_id": sourceExec.RunID, - "source_execution_id": sourceExec.ExecutionID, - "restarted_execution_id": restartExec.ExecutionID, - "reuse": reuse, - "scope": scope, - }, + lineage := map[string]interface{}{ + "kind": kind, + "source_run_id": sourceExec.RunID, + "source_execution_id": sourceExec.ExecutionID, + "restarted_execution_id": restartExec.ExecutionID, + "reuse": reuse, + "scope": scope, } + var encodedReason json.RawMessage if trimmed := strings.TrimSpace(reason); trimmed != "" { - metadata["reason"] = trimmed + encodedReason, _ = json.Marshal(trimmed) } - encoded, err := json.Marshal(metadata) + encoded, err := json.Marshal(lineage) if err != nil { logger.Logger.Warn().Err(err).Str("run_id", plan.exec.RunID).Msg("failed to encode restart run metadata") return } - // This workflow_runs row exists only to carry lineage/golden metadata for the - // new run; it is the sole writer of this row for restart runs. Status and - // TotalSteps are seeded at enqueue time and are NOT kept current as the run - // progresses — every UI read path (run list, run detail, DAG) derives live - // status and step counts from execution aggregation and only reads the - // lineage/golden fields here. Do not treat these columns as authoritative. - if err := store.StoreWorkflowRun(ctx, &types.WorkflowRun{ - RunID: plan.exec.RunID, - RootWorkflowID: plan.exec.RunID, - RootExecutionID: &plan.exec.ExecutionID, - Status: string(types.ExecutionStatusQueued), - TotalSteps: 1, - Metadata: json.RawMessage(encoded), - CreatedAt: now, - UpdatedAt: now, + if err := store.UpdateWorkflowRunMetadata(ctx, plan.exec.RunID, func(namespaces map[string]json.RawMessage) error { + namespaces["lineage"] = encoded + if encodedReason != nil { + namespaces["reason"] = encodedReason + } else { + delete(namespaces, "reason") + } + return nil }); err != nil { logger.Logger.Warn().Err(err).Str("run_id", plan.exec.RunID).Msg("failed to persist restart run metadata") } diff --git a/control-plane/internal/handlers/mcp.go b/control-plane/internal/handlers/mcp.go index f03fbe286..b64ba1e97 100644 --- a/control-plane/internal/handlers/mcp.go +++ b/control-plane/internal/handlers/mcp.go @@ -512,6 +512,7 @@ func (s *mcpServer) toolWaitRun(c *gin.Context, rawArgs json.RawMessage) (map[st // the MCP tool. It returns as soon as the job is enqueued. func (s *mcpServer) startAsyncRun(ctx context.Context, target string, input map[string]interface{}, headers executionHeaders, callerDID, targetDID string) (runID, execID string, err error) { controller := newExecutionController(s.store, s.payloads, s.webhooks, s.timeout, s.internalToken) + // MCP creates an ordinary execution and deliberately leaves RunMetadata nil. plan, err := controller.prepareExecutionForTarget(ctx, target, ExecuteRequest{Input: input}, headers, callerDID, targetDID) if err != nil { return "", "", err diff --git a/control-plane/internal/handlers/run_metadata.go b/control-plane/internal/handlers/run_metadata.go new file mode 100644 index 000000000..d724cf48d --- /dev/null +++ b/control-plane/internal/handlers/run_metadata.go @@ -0,0 +1,162 @@ +package handlers + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" + "unicode/utf8" + + "github.com/Agent-Field/agentfield/control-plane/pkg/types" + "github.com/gin-gonic/gin" +) + +// RunMetadataInput preserves absent, null and value as distinct patch states. +type RunMetadataInput struct { + DisplayName json.RawMessage `json:"display_name,omitempty"` + Labels json.RawMessage `json:"labels,omitempty"` + Links json.RawMessage `json:"links,omitempty"` +} + +type workflowRunMetadataWriter interface { + UpdateWorkflowRunMetadata(context.Context, string, func(map[string]json.RawMessage) error) error +} + +// SetRunMetadataHandler handles POST /api/v1/runs/:run_id/metadata. +func SetRunMetadataHandler(store ExecutionStore) gin.HandlerFunc { + return func(c *gin.Context) { + runID := strings.TrimSpace(c.Param("run_id")) + if runID == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "run_id is required"}) + return + } + var input RunMetadataInput + if err := c.ShouldBindJSON(&input); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request body: " + err.Error()}) + return + } + if _, err := applyRunMetadataInput(types.RunMetadata{}, input); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + executions, err := store.QueryExecutionRecords(c.Request.Context(), types.ExecutionFilter{RunID: &runID, Limit: 1}) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + if len(executions) == 0 { + c.JSON(http.StatusNotFound, gin.H{"error": "workflow run not found"}) + return + } + writer, ok := store.(workflowRunMetadataWriter) + if !ok { + c.JSON(http.StatusNotImplemented, gin.H{"error": "run metadata storage is not supported"}) + return + } + actor := strings.TrimSpace(c.GetHeader("X-Actor-ID")) + if actor == "" { + actor = "api" + } + var merged types.RunMetadata + err = writer.UpdateWorkflowRunMetadata(c.Request.Context(), runID, func(namespaces map[string]json.RawMessage) error { + current := types.RunMetadata{} + if raw, exists := namespaces[types.RunMetadataNamespace]; exists { + _ = json.Unmarshal(raw, ¤t) + } + var err error + merged, err = applyRunMetadataInput(current, input) + if err != nil { + return err + } + merged.SetBy = actor + merged.UpdatedAt = time.Now().UTC().Format(time.RFC3339) + if merged.DisplayName == "" && len(merged.Labels) == 0 && len(merged.Links) == 0 { + delete(namespaces, types.RunMetadataNamespace) + return nil + } + namespaces[types.RunMetadataNamespace], err = json.Marshal(merged) + return err + }) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, merged) + } +} + +// applyRunMetadataInput merges a public API patch. Unlike the UI-private golden +// route, which truncates its fixed button input, this public endpoint rejects +// values outside the documented bounds. +func applyRunMetadataInput(current types.RunMetadata, input RunMetadataInput) (types.RunMetadata, error) { + if input.DisplayName != nil { + if string(input.DisplayName) == "null" { + current.DisplayName = "" + } else { + var value string + if err := json.Unmarshal(input.DisplayName, &value); err != nil { + return current, fmt.Errorf("display_name must be a string") + } + value = strings.TrimSpace(value) + if utf8.RuneCountInString(value) > types.MaxRunDisplayNameRunes { + return current, fmt.Errorf("display_name exceeds %d runes", types.MaxRunDisplayNameRunes) + } + current.DisplayName = value + } + } + if input.Labels != nil { + if string(input.Labels) == "null" { + current.Labels = nil + } else { + var values []string + if err := json.Unmarshal(input.Labels, &values); err != nil { + return current, fmt.Errorf("labels must be an array of strings") + } + if len(values) > types.MaxRunLabels { + return current, fmt.Errorf("labels exceeds %d items", types.MaxRunLabels) + } + seen := make(map[string]struct{}, len(values)) + current.Labels = nil + for _, value := range values { + value = strings.TrimSpace(value) + if utf8.RuneCountInString(value) > types.MaxRunLabelRunes { + return current, fmt.Errorf("label exceeds %d runes", types.MaxRunLabelRunes) + } + if value == "" { + continue + } + if _, exists := seen[value]; exists { + continue + } + seen[value] = struct{}{} + current.Labels = append(current.Labels, value) + } + } + } + if input.Links != nil { + if string(input.Links) == "null" { + current.Links = nil + } else { + var links []types.RunMetadataLink + if err := json.Unmarshal(input.Links, &links); err != nil { + return current, fmt.Errorf("links must be an array of links") + } + if len(links) > types.MaxRunLinks { + return current, fmt.Errorf("links exceeds %d items", types.MaxRunLinks) + } + for i := range links { + links[i].Label = strings.TrimSpace(links[i].Label) + if utf8.RuneCountInString(links[i].Label) > types.MaxRunLinkLabelRunes { + return current, fmt.Errorf("link label exceeds %d runes", types.MaxRunLinkLabelRunes) + } + if err := types.ValidateRunLinkURL(links[i].URL); err != nil { + return current, fmt.Errorf("invalid link url: %w", err) + } + } + current.Links = links + } + } + return current, nil +} diff --git a/control-plane/internal/handlers/ui/workflow_runs.go b/control-plane/internal/handlers/ui/workflow_runs.go index 8dea2e327..9d961ce2d 100644 --- a/control-plane/internal/handlers/ui/workflow_runs.go +++ b/control-plane/internal/handlers/ui/workflow_runs.go @@ -60,9 +60,10 @@ type WorkflowRunSummary struct { // this run, when one exists. Populated by walking the root execution's // VC chain back to the parent trigger_event VC. Nil for runs invoked // directly or by another reasoner. - Trigger *types.TriggerEventMetadata `json:"trigger,omitempty"` - Lineage *RunLineageMetadata `json:"lineage,omitempty"` - Golden *GoldenRunMetadata `json:"golden,omitempty"` + Trigger *types.TriggerEventMetadata `json:"trigger,omitempty"` + Lineage *RunLineageMetadata `json:"lineage,omitempty"` + Golden *GoldenRunMetadata `json:"golden,omitempty"` + RunMetadata *types.RunMetadata `json:"run_metadata,omitempty"` } type RunLineageMetadata struct { @@ -89,9 +90,9 @@ type GoldenRunMetadata struct { // They are package-level constants rather than inline literals so the // run-metadata endpoint can reuse the same bounds. const ( - maxGoldenTags = 20 - maxGoldenTagRunes = 64 - maxGoldenNameRunes = 200 + maxGoldenTags = types.MaxRunLabels + maxGoldenTagRunes = types.MaxRunLabelRunes + maxGoldenNameRunes = types.MaxRunDisplayNameRunes ) type WorkflowRunListResponse struct { @@ -118,6 +119,7 @@ type WorkflowRunDetailResponse struct { CompletedAt *string `json:"completed_at,omitempty"` Lineage *RunLineageMetadata `json:"lineage,omitempty"` Golden *GoldenRunMetadata `json:"golden,omitempty"` + RunMetadata *types.RunMetadata `json:"run_metadata,omitempty"` } `json:"run"` Executions []apiWorkflowExecution `json:"executions"` } @@ -475,6 +477,7 @@ func (h *WorkflowRunHandler) GetWorkflowRunDetailHandler(c *gin.Context) { if metadata := h.loadRunMetadata(ctx, runID); metadata != nil { detail.Run.Lineage = metadata.Lineage detail.Run.Golden = metadata.Golden + detail.Run.RunMetadata = metadata.Run } if agg := h.loadRunSummary(ctx, runID); agg != nil { @@ -598,6 +601,7 @@ func summarizeRun(runID string, executions []*types.Execution) WorkflowRunSummar type parsedRunMetadata struct { Lineage *RunLineageMetadata Golden *GoldenRunMetadata + Run *types.RunMetadata } func (h *WorkflowRunHandler) enrichRunMetadata(ctx context.Context, summary *WorkflowRunSummary) { @@ -610,6 +614,7 @@ func (h *WorkflowRunHandler) enrichRunMetadata(ctx context.Context, summary *Wor } summary.Lineage = metadata.Lineage summary.Golden = metadata.Golden + summary.RunMetadata = metadata.Run } func (h *WorkflowRunHandler) loadRunMetadata(ctx context.Context, runID string) *parsedRunMetadata { @@ -626,7 +631,7 @@ func (h *WorkflowRunHandler) loadRunMetadata(ctx context.Context, runID string) return nil } - parsed := &parsedRunMetadata{} + parsed := &parsedRunMetadata{Run: types.ParseRunMetadata(run.Metadata)} if raw, ok := metadata["lineage"]; ok { if encoded, err := json.Marshal(raw); err == nil { var lineage RunLineageMetadata @@ -643,7 +648,7 @@ func (h *WorkflowRunHandler) loadRunMetadata(ctx context.Context, runID string) } } } - if parsed.Lineage == nil && parsed.Golden == nil { + if parsed.Lineage == nil && parsed.Golden == nil && parsed.Run == nil { return nil } return parsed @@ -719,6 +724,8 @@ func truncateRunes(value string, maxRunes int) string { } func deriveOverallStatusForUI(executions []*types.Execution) string { + // Run display metadata never participates in derived execution status. The + // deferred external_status field must not feed this calculation either. counts := make(map[string]int) active := 0 for _, exec := range executions { diff --git a/control-plane/internal/handlers/workflow_dag.go b/control-plane/internal/handlers/workflow_dag.go index 703108d16..2f5b20107 100644 --- a/control-plane/internal/handlers/workflow_dag.go +++ b/control-plane/internal/handlers/workflow_dag.go @@ -62,9 +62,10 @@ type WorkflowDAGResponse struct { // Trigger describes the inbound webhook (or schedule) that originated // this run, when one exists. Populated by walking the root execution's // VC chain back to the parent trigger_event VC. - Trigger *types.TriggerEventMetadata `json:"trigger,omitempty"` - Lineage *RunLineageMetadata `json:"lineage,omitempty"` - Golden *GoldenRunMetadata `json:"golden,omitempty"` + Trigger *types.TriggerEventMetadata `json:"trigger,omitempty"` + Lineage *RunLineageMetadata `json:"lineage,omitempty"` + Golden *GoldenRunMetadata `json:"golden,omitempty"` + RunMetadata *types.RunMetadata `json:"run_metadata,omitempty"` } type RunLineageMetadata struct { @@ -160,9 +161,10 @@ type WorkflowDAGLightweightResponse struct { WebhookFailures []WebhookFailurePreview `json:"webhook_failures,omitempty"` // Trigger describes the inbound webhook (or schedule) that originated // this run, when one exists. - Trigger *types.TriggerEventMetadata `json:"trigger,omitempty"` - Lineage *RunLineageMetadata `json:"lineage,omitempty"` - Golden *GoldenRunMetadata `json:"golden,omitempty"` + Trigger *types.TriggerEventMetadata `json:"trigger,omitempty"` + Lineage *RunLineageMetadata `json:"lineage,omitempty"` + Golden *GoldenRunMetadata `json:"golden,omitempty"` + RunMetadata *types.RunMetadata `json:"run_metadata,omitempty"` } type workflowRunMetadataGetter interface { @@ -196,7 +198,7 @@ func (s *executionGraphService) handleGetWorkflowDAG(c *gin.Context) { } rootExecID := findRootExecutionID(executions) - lineage, golden := s.loadRunMetadata(ctx, runID) + lineage, golden, runMetadata := s.loadRunMetadata(ctx, runID) if isLightweightRequest(c) { timeline, workflowStatus, workflowName, sessionID, actorID, maxDepth := buildLightweightExecutionDAG(executions) @@ -224,6 +226,7 @@ func (s *executionGraphService) handleGetWorkflowDAG(c *gin.Context) { Trigger: TriggerForRun(ctx, s.store, runID, rootExecID), Lineage: lineage, Golden: golden, + RunMetadata: runMetadata, } c.JSON(http.StatusOK, response) @@ -251,23 +254,24 @@ func (s *executionGraphService) handleGetWorkflowDAG(c *gin.Context) { Trigger: TriggerForRun(ctx, s.store, runID, rootExecID), Lineage: lineage, Golden: golden, + RunMetadata: runMetadata, } c.JSON(http.StatusOK, response) } -func (s *executionGraphService) loadRunMetadata(ctx context.Context, runID string) (*RunLineageMetadata, *GoldenRunMetadata) { +func (s *executionGraphService) loadRunMetadata(ctx context.Context, runID string) (*RunLineageMetadata, *GoldenRunMetadata, *types.RunMetadata) { getter, ok := s.store.(workflowRunMetadataGetter) if !ok { - return nil, nil + return nil, nil, nil } run, err := getter.GetWorkflowRun(ctx, runID) if err != nil || run == nil || len(run.Metadata) == 0 { - return nil, nil + return nil, nil, nil } var raw map[string]interface{} if err := json.Unmarshal(run.Metadata, &raw); err != nil { - return nil, nil + return nil, nil, nil } var lineage *RunLineageMetadata if value, ok := raw["lineage"]; ok { @@ -287,7 +291,7 @@ func (s *executionGraphService) loadRunMetadata(ctx context.Context, runID strin } } } - return lineage, golden + return lineage, golden, types.ParseRunMetadata(run.Metadata) } // findRootExecutionID returns the execution_id of the root node — the diff --git a/control-plane/internal/server/apicatalog/catalog_entries.go b/control-plane/internal/server/apicatalog/catalog_entries.go index 63dd23492..8700ded9b 100644 --- a/control-plane/internal/server/apicatalog/catalog_entries.go +++ b/control-plane/internal/server/apicatalog/catalog_entries.go @@ -90,6 +90,12 @@ func DefaultEntries() []EndpointEntry { {Method: "POST", Path: "/api/v1/executions/:execution_id/resume", Group: "executions", Summary: "Resume a paused execution", AuthLevel: "api_key", Tags: []string{"executions", "resume"}}, {Method: "POST", Path: "/api/v1/workflows/:workflowId/cancel-tree", Group: "workflows", Summary: "Cancel every non-terminal execution in a run (bottom-up)", AuthLevel: "api_key", Tags: []string{"workflows", "executions", "cancel"}}, + // --- Runs --- + {Method: "POST", Path: "/api/v1/runs/:run_id/metadata", Group: "runs", Summary: "Set run display name, labels and links", AuthLevel: "api_key", Tags: []string{"runs", "metadata", "labels"}, + Parameters: []ParamEntry{{Name: "run_id", In: "path", Required: true, Type: "string", Desc: "Run ID"}}, + RequestBody: &BodyEntry{ContentType: "application/json", Fields: map[string]string{"display_name": "string|null", "labels": "string[]|null", "links": "array|null"}}, + }, + // --- Approval --- {Method: "POST", Path: "/api/v1/executions/:execution_id/request-approval", Group: "approval", Summary: "Request approval for an execution", AuthLevel: "api_key", Tags: []string{"approval", "request"}}, {Method: "GET", Path: "/api/v1/executions/:execution_id/approval-status", Group: "approval", Summary: "Get approval status", AuthLevel: "api_key", Tags: []string{"approval", "status"}}, diff --git a/control-plane/internal/server/routes_core.go b/control-plane/internal/server/routes_core.go index 05ce43bb0..70256caf4 100644 --- a/control-plane/internal/server/routes_core.go +++ b/control-plane/internal/server/routes_core.go @@ -142,6 +142,7 @@ func (s *AgentFieldServer) registerCoreRoutes(agentAPI *gin.RouterGroup) { agentAPI.POST("/executions/:execution_id/pause", handlers.PauseExecutionHandler(s.storage)) agentAPI.POST("/executions/:execution_id/resume", handlers.ResumeExecutionHandler(s.storage)) agentAPI.POST("/executions/:execution_id/restart", handlers.RestartExecutionHandler(s.storage, s.payloadStore, s.webhookDispatcher, s.config.AgentField.ExecutionQueue.AgentCallTimeout, s.config.Features.DID.Authorization.InternalToken)) + agentAPI.POST("/runs/:run_id/metadata", handlers.SetRunMetadataHandler(s.storage)) agentAPI.POST("/workflows/:workflowId/cancel-tree", handlers.CancelWorkflowTreeHandler(s.storage)) // Approval workflow endpoints — CP manages execution state only; diff --git a/control-plane/internal/storage/local.go b/control-plane/internal/storage/local.go index 789e78c6a..9d01bb94a 100644 --- a/control-plane/internal/storage/local.go +++ b/control-plane/internal/storage/local.go @@ -9,6 +9,7 @@ import ( "fmt" "reflect" "regexp" + "sort" "strings" "sync" "time" @@ -322,6 +323,96 @@ func (ls *LocalStorage) GetWorkflowRun(ctx context.Context, runID string) (*type return &run, nil } +// UpdateWorkflowRunMetadata applies mutate to the decoded metadata object in one transaction. +// Untouched namespaces are retained as json.RawMessage values. This deliberately is not +// StoreWorkflowRun: that full-row upsert resets status, counts and event-version columns. +func (ls *LocalStorage) UpdateWorkflowRunMetadata(ctx context.Context, runID string, mutate func(map[string]json.RawMessage) error) error { + runID = strings.TrimSpace(runID) + if runID == "" { + return fmt.Errorf("run_id cannot be empty") + } + if mutate == nil { + return fmt.Errorf("metadata mutator cannot be nil") + } + + operationID := "UpdateWorkflowRunMetadata:" + runID + // SQLite connections use _txlock=immediate, so BeginTx acquires the write + // reservation before reading. PostgreSQL first seeds the row conflict-safely, + // then locks and re-reads it. Thus every mutator starts from the preceding + // writer's committed namespaces on both backends. + return ls.retryDatabaseOperation(ctx, operationID, func() error { + db := ls.requireSQLDB() + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer rollbackTx(tx, operationID) + + now := time.Now().UTC() + // This row exists only to carry metadata. Status and step counts are seeded, + // are not kept current, and no read path treats them as authoritative. + _, err = tx.ExecContext(ctx, `INSERT INTO workflow_runs ( + run_id, root_workflow_id, root_execution_id, status, total_steps, + completed_steps, failed_steps, state_version, last_event_sequence, + metadata, created_at, updated_at + ) VALUES (?, ?, NULL, 'pending', 0, 0, 0, 0, 0, '{}', ?, ?) + ON CONFLICT(run_id) DO NOTHING`, runID, runID, now, now) + if err != nil { + return err + } + + var raw sql.NullString + err = tx.QueryRowContext(ctx, `SELECT metadata FROM workflow_runs WHERE run_id = ?`+tx.forUpdate(), runID).Scan(&raw) + if err != nil { + return err + } + + metadata := make(map[string]json.RawMessage) + if raw.Valid && strings.TrimSpace(raw.String) != "" { + if err := json.Unmarshal([]byte(raw.String), &metadata); err != nil { + metadata = make(map[string]json.RawMessage) + } + } + if err := mutate(metadata); err != nil { + return err + } + encoded, err := marshalMetadataNamespaces(metadata) + if err != nil { + return err + } + _, err = tx.ExecContext(ctx, `UPDATE workflow_runs SET metadata = ?, updated_at = ? WHERE run_id = ?`, string(encoded), time.Now().UTC(), runID) + if err != nil { + return err + } + return tx.Commit() + }) +} + +func marshalMetadataNamespaces(metadata map[string]json.RawMessage) ([]byte, error) { + keys := make([]string, 0, len(metadata)) + for key := range metadata { + keys = append(keys, key) + } + sort.Strings(keys) + var encoded bytes.Buffer + encoded.WriteByte('{') + for i, key := range keys { + raw := metadata[key] + if !json.Valid(raw) { + return nil, fmt.Errorf("metadata namespace %q is invalid JSON", key) + } + if i > 0 { + encoded.WriteByte(',') + } + encodedKey, _ := json.Marshal(key) + encoded.Write(encodedKey) + encoded.WriteByte(':') + encoded.Write(raw) + } + encoded.WriteByte('}') + return encoded.Bytes(), nil +} + func (ls *LocalStorage) StoreWorkflowRunEvent(ctx context.Context, event *types.WorkflowRunEvent) error { if event == nil { return fmt.Errorf("workflow run event cannot be nil") diff --git a/control-plane/pkg/types/run_metadata.go b/control-plane/pkg/types/run_metadata.go new file mode 100644 index 000000000..eb24de106 --- /dev/null +++ b/control-plane/pkg/types/run_metadata.go @@ -0,0 +1,77 @@ +package types + +import ( + "bytes" + "encoding/json" + "fmt" + "net/url" +) + +// RunMetadataNamespace is the namespace key inside workflow_runs.metadata. +const RunMetadataNamespace = "run" + +const ( + MaxRunDisplayNameRunes = 200 + MaxRunLabels = 20 + MaxRunLabelRunes = 64 + MaxRunLinks = 10 + MaxRunLinkLabelRunes = 64 + MaxRunLinkURLBytes = 2048 +) + +type RunMetadataLink struct { + Label string `json:"label,omitempty"` + URL string `json:"url"` +} + +type RunMetadata struct { + DisplayName string `json:"display_name,omitempty"` + Labels []string `json:"labels,omitempty"` + Links []RunMetadataLink `json:"links,omitempty"` + SetBy string `json:"set_by,omitempty"` + UpdatedAt string `json:"updated_at,omitempty"` +} + +// ParseRunMetadata decodes workflow_runs.metadata and returns its run namespace. +func ParseRunMetadata(raw json.RawMessage) *RunMetadata { + var namespaces map[string]json.RawMessage + if len(raw) == 0 || json.Unmarshal(raw, &namespaces) != nil { + return nil + } + value, ok := namespaces[RunMetadataNamespace] + if !ok { + return nil + } + value = bytes.TrimSpace(value) + if len(value) == 0 || value[0] != '{' { + return nil + } + var metadata RunMetadata + if json.Unmarshal(value, &metadata) != nil { + return nil + } + return &metadata +} + +// ValidateRunLinkURL permits ordinary HTTP(S) links with a host and no credentials. +// services.ValidateWebhookURL is deliberately not used: its private-address blocking +// would reject legitimate internal PR and ticket links. +func ValidateRunLinkURL(raw string) error { + if len(raw) > MaxRunLinkURLBytes { + return fmt.Errorf("url exceeds %d bytes", MaxRunLinkURLBytes) + } + parsed, err := url.Parse(raw) + if err != nil { + return fmt.Errorf("invalid url: %w", err) + } + if parsed.Scheme != "http" && parsed.Scheme != "https" { + return fmt.Errorf("url scheme must be http or https") + } + if parsed.Host == "" { + return fmt.Errorf("url must include a host") + } + if parsed.User != nil { + return fmt.Errorf("url must not include credentials") + } + return nil +} From f47eef6ff89fe78faa44312a78bc1a6ba01d2ec4 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 31 Aug 2026 14:45:04 -0400 Subject: [PATCH 4/8] fix(storage): open SQLite with _txlock=immediate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Write transactions now take the write reservation at BEGIN instead of on first write, so the read-merge-write metadata primitive (and every other BeginTx writer) cannot hit the read->write upgrade deadlock; WAL and the 60s busy timeout were already in place. Global, deliberate change — the full suite gates it. Co-Authored-By: Claude Fable 5 --- control-plane/internal/storage/local_init.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/control-plane/internal/storage/local_init.go b/control-plane/internal/storage/local_init.go index f8da77291..6563d1880 100644 --- a/control-plane/internal/storage/local_init.go +++ b/control-plane/internal/storage/local_init.go @@ -106,7 +106,7 @@ func (ls *LocalStorage) initializeSQLite(ctx context.Context) error { busyTimeout = 60000 } - dsn := fmt.Sprintf("%s?_journal_mode=WAL&_synchronous=NORMAL&_cache_size=10000&_foreign_keys=ON&_busy_timeout=%d&_wal_autocheckpoint=1000&_temp_store=MEMORY&_mmap_size=268435456", + dsn := fmt.Sprintf("%s?_journal_mode=WAL&_synchronous=NORMAL&_cache_size=10000&_foreign_keys=ON&_busy_timeout=%d&_txlock=immediate&_wal_autocheckpoint=1000&_temp_store=MEMORY&_mmap_size=268435456", dbPath, busyTimeout) db, err := sql.Open("sqlite3", dsn) From 0e79f0e686bd592ac6288aa56eaa4ac44b0d4404 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 31 Aug 2026 14:45:04 -0400 Subject: [PATCH 5/8] feat(web-ui): render run display names, label chips and safe external links Display name takes precedence in the run list row, labels render as chips, links render only after scheme re-validation (http/https, host required) with rel="noopener noreferrer"; nothing is treated as trusted HTML. Co-Authored-By: Claude Fable 5 --- .../web/client/src/pages/RunDetailPage.tsx | 18 +++++++- .../web/client/src/pages/RunsPage.tsx | 19 +++++++- .../web/client/src/services/workflowsApi.ts | 4 ++ .../src/test/pages/RunDetailPage.test.tsx | 35 +++++++++++++++ .../client/src/test/pages/RunsPage.test.tsx | 43 +++++++++++++++++++ .../web/client/src/types/workflows.ts | 16 +++++++ .../client/src/utils/safeExternalUrl.test.ts | 20 +++++++++ .../web/client/src/utils/safeExternalUrl.ts | 19 ++++++++ 8 files changed, 172 insertions(+), 2 deletions(-) create mode 100644 control-plane/web/client/src/utils/safeExternalUrl.test.ts create mode 100644 control-plane/web/client/src/utils/safeExternalUrl.ts diff --git a/control-plane/web/client/src/pages/RunDetailPage.tsx b/control-plane/web/client/src/pages/RunDetailPage.tsx index ef6931572..9d0bcc478 100644 --- a/control-plane/web/client/src/pages/RunDetailPage.tsx +++ b/control-plane/web/client/src/pages/RunDetailPage.tsx @@ -87,6 +87,7 @@ import { WorkflowDAGViewer } from "@/components/WorkflowDAG"; import { ErrorBoundary } from "@/components/ErrorBoundary"; import { ExecutionObservabilityPanel } from "@/components/execution"; import { normalizeExecutionStatus, isTerminalStatus } from "@/utils/status"; +import { safeExternalUrl } from "@/utils/safeExternalUrl"; import { StatusPill } from "@/components/ui/status-pill"; import type { TriggerInfo, @@ -595,6 +596,7 @@ export function RunDetailPage() { dag?.timeline.find((n) => n.workflow_depth === 0) ?? dag?.timeline[0]; const restartNodeForActions = pickRestartNode(dag?.timeline); const actionRunLabel = + dag?.run_metadata?.display_name?.trim() || dag?.workflow_name?.trim() || (rootNodeForActions?.agent_node_id && rootNodeForActions?.reasoner_id ? `${rootNodeForActions.agent_node_id}.${rootNodeForActions.reasoner_id}` @@ -906,7 +908,7 @@ export function RunDetailPage() { vcChain?.workflow_vc?.issuer_did?.trim() || ""; - const runTitle = dag.workflow_name?.trim() || rootNode?.reasoner_id || "Run"; + const runTitle = dag.run_metadata?.display_name?.trim() || dag.workflow_name?.trim() || rootNode?.reasoner_id || "Run"; const runTitleDisplay = truncateEnd(runTitle, RUN_DETAIL_TITLE_MAX_CHARS); const metaParts: string[] = []; @@ -1005,6 +1007,20 @@ export function RunDetailPage() { {lineage.kind === "fork" ? "Forked" : "Restarted"} ) : null} + {dag.run_metadata?.labels?.slice(0, 3).map((label) => ( + {label} + ))} + {(dag.run_metadata?.labels?.length ?? 0) > 3 ? ( + +{(dag.run_metadata?.labels?.length ?? 0) - 3} + ) : null} + {dag.run_metadata?.links?.map((link) => { + const href = safeExternalUrl(link.url); + return href ? ( + + {link.label || link.url} + + ) : null; + })} {sessionTrim ? ( diff --git a/control-plane/web/client/src/pages/RunsPage.tsx b/control-plane/web/client/src/pages/RunsPage.tsx index 21c178175..99f27fadf 100644 --- a/control-plane/web/client/src/pages/RunsPage.tsx +++ b/control-plane/web/client/src/pages/RunsPage.tsx @@ -98,6 +98,7 @@ import { import { useSidebar } from "@/components/ui/sidebar"; import { SortableHeaderCell } from "@/components/ui/CompactTable"; import { SourceIcon } from "@/components/triggers/SourceIcon"; +import { safeExternalUrl } from "@/utils/safeExternalUrl"; import { getExecutionDetails } from "@/services/executionsApi"; import { getWorkflowDAGLightweight } from "@/services/workflowsApi"; import { JsonHighlightedPre } from "@/components/ui/json-syntax-highlight"; @@ -612,6 +613,8 @@ export function RunsPage() { /** Human-readable label for a run — e.g. `demo-runs.slow_task`. */ const runDisplayLabel = useCallback((run: WorkflowSummary) => { + const customName = run.run_metadata?.display_name?.trim(); + if (customName) return customName; const reasoner = run.root_reasoner || run.display_name || "run"; return run.agent_id ? `${run.agent_id}.${reasoner}` : reasoner; }, []); @@ -1742,7 +1745,7 @@ function RunRow({ onRestartRun, }: RunRowProps) { const agentLabel = run.agent_id || run.agent_name || ""; - const reasonerLabel = run.root_reasoner || run.display_name || "—"; + const reasonerLabel = run.run_metadata?.display_name?.trim() || run.root_reasoner || run.display_name || "—"; const [copied, setCopied] = useState(false); const errorMeta = (run.root_execution_status ?? run.status) === "failed" @@ -1881,6 +1884,20 @@ function RunRow({ {run.lineage.kind === "fork" ? "Forked" : "Restarted"} ) : null} + {run.run_metadata?.labels?.slice(0, 3).map((label) => ( + {label} + ))} + {(run.run_metadata?.labels?.length ?? 0) > 3 ? ( + +{(run.run_metadata?.labels?.length ?? 0) - 3} + ) : null} + {run.run_metadata?.links?.map((link) => { + const href = safeExternalUrl(link.url); + return href ? ( + e.stopPropagation()}> + {link.label || link.url} + + ) : null; + })}