From 3187d8555e52d3664ecd8fa34dcc4fa7279ba4c1 Mon Sep 17 00:00:00 2001 From: Zpc001 <1903085231@qq.com> Date: Mon, 19 Oct 2026 22:27:20 +0800 Subject: [PATCH 1/2] feat(core): persist the run kickoff in the Start variable pool The Start node's `input` root now reads the kickoff text the run stored in its own input object, falling back to the authored Start instruction in the frozen snapshot. Runs created before the kickoff was stored have no such key, so they keep evaluating with exactly the text their snapshot froze; no migration or backfill is needed because `workflow_runs.input` already carries the object. A run's `input` key is the reserved kickoff selector, not a workflow variable: it is accepted only as a string, as desktop's `kickoff_input` is, so `[start, input]` and `{{#start.input#}}` answer with the same type on both sides. When neither the run nor the document supplies a value, the root is left out of the pool, matching desktop's unset `{start}.input`; an explicit empty kickoff remains present and empty. The simulator shares the resolver so its trace matches what the scheduler commits. Refs ora-space/circus#289 Co-Authored-By: Claude Code --- api/openapi.json | 2 +- frontend/src/api/generated.schemas.ts | 4 +- integration/workflow_admission_test.go | 2 +- integration/workflow_run_test.go | 27 ++++- internal/collab/workflow-run-simulator.go | 9 +- internal/contract/openapi.go | 2 +- internal/core/workflow_admission.go | 10 ++ internal/core/workflow_admission_test.go | 18 +++ internal/core/workflow_scheduler.go | 47 +++++-- internal/core/workflow_scheduler_db_test.go | 128 +++++++++++++++++--- internal/core/workflow_scheduler_test.go | 73 +++++++++-- 11 files changed, 274 insertions(+), 48 deletions(-) diff --git a/api/openapi.json b/api/openapi.json index 5db86698b..18a42d06d 100644 --- a/api/openapi.json +++ b/api/openapi.json @@ -5301,7 +5301,7 @@ }, "input": { "additionalProperties": true, - "description": "The kickoff input passed to the run's start node.", + "description": "The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff — including every run created before the key was stored — evaluates against the Start node's authored instruction instead.", "type": "object" }, "modelBindings": { diff --git a/frontend/src/api/generated.schemas.ts b/frontend/src/api/generated.schemas.ts index 665afb1eb..8642cfcb8 100644 --- a/frontend/src/api/generated.schemas.ts +++ b/frontend/src/api/generated.schemas.ts @@ -2373,7 +2373,7 @@ export interface WorkflowAssetImportList { export type WorkflowRunDefinitionSnapshot = { [key: string]: unknown } | null; /** - * The kickoff input passed to the run's start node. + * The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff — including every run created before the key was stored — evaluates against the Start node's authored instruction instead. */ export type WorkflowRunInput = { [key: string]: unknown }; @@ -2415,7 +2415,7 @@ export interface WorkflowRun { /** @nullable */ finishedAt: string | null; id: string; - /** The kickoff input passed to the run's start node. */ + /** The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff — including every run created before the key was stored — evaluates against the Start node's authored instruction instead. */ input: WorkflowRunInput; /** * The frozen per-slot model bindings this run executes against, ordered by slot name. Present on a detail read. Each binding is frozen at run creation and used for the run's whole lifetime; only `revoked` can turn true (the owning connection's grant was revoked). diff --git a/integration/workflow_admission_test.go b/integration/workflow_admission_test.go index b2a24db7c..bc2c86d6e 100644 --- a/integration/workflow_admission_test.go +++ b/integration/workflow_admission_test.go @@ -31,7 +31,7 @@ func agentRunGraph(slot string, legacy bool, slots []any) core.Object { core.Object{ "id": "start", "type": "workflow", "deletable": false, "position": core.Object{"x": 0, "y": 0}, - "data": core.Object{"kind": "start", "title": "开始", "description": ""}, + "data": core.Object{"kind": "start", "title": "开始", "description": "", "input": ""}, }, core.Object{ "id": "agent-1", "type": "workflow", diff --git a/integration/workflow_run_test.go b/integration/workflow_run_test.go index c8528f430..de9a3e2fc 100644 --- a/integration/workflow_run_test.go +++ b/integration/workflow_run_test.go @@ -98,7 +98,8 @@ func TestWorkflowRunLifecycle(t *testing.T) { t.Fatalf("run node_states missing nodes: %v", states) } // The start node reports the run form the way desktop's variable pool does: each run-form variable - // is its own root, and `input` is the authored kickoff prompt (unset in this fixture). + // is its own root, and `input` is the kickoff prompt. This run submitted none, so it falls back to + // the authored (empty) instruction in the frozen document. output := states.O("start").O("output") if output.S("branch") != "main" { t.Fatalf("start output did not expose the run input variable: %v", output) @@ -106,6 +107,26 @@ func TestWorkflowRunLifecycle(t *testing.T) { if prompt, ok := output["input"].(string); !ok || prompt != "" { t.Fatalf("start output input = %v; want the authored kickoff prompt as a string", output["input"]) } + + // The run's own kickoff is persisted on the row and is what its Start output reports, even when + // the authored instruction differs. The rest of the run form keeps its roots either way. + override := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{ + "name": "Override run", "input": core.Object{"branch": "dev", "input": "review the diff"}, "projectId": pid, + }, "run-override", 200).O("resource") + if override.O("input").S("input") != "review the diff" { + t.Fatalf("run row lost the kickoff: %v", override.O("input")) + } + overrideOutput := override.O("nodeStates").O("start").O("output") + if overrideOutput.S("input") != "review the diff" || overrideOutput.S("branch") != "dev" { + t.Fatalf("start output did not read the run's kickoff: %v", overrideOutput) + } + // The reserved selector accepts a string, the type desktop's kickoff_input carries, and nothing + // else: another JSON type would change what [start, input] means across the two sides. + if refusal := f.call("POST", f.path("/workflows/"+wfid+"/runs"), core.Object{ + "input": core.Object{"input": 7}, "projectId": pid, + }, "run-bad-kickoff", 400); refusal.S("code") != "invalid_field_value" { + t.Fatalf("non-string kickoff refusal = %v; want invalid_field_value", refusal) + } // definitionSnapshot is the frozen graph, not the live document: the run carries its own copy. ds := run.O("definitionSnapshot") if ds.O("viewport").N("zoom") != 1 { @@ -130,8 +151,8 @@ func TestWorkflowRunLifecycle(t *testing.T) { t.Fatalf("run read-back changed the trace: %v", one) } listed := issueItems(f.call("GET", f.path("/workflows/"+wfid+"/runs"), nil, "", 200)) - if len(listed) != 2 { - t.Fatalf("run list has %d rows, want 2", len(listed)) + if len(listed) != 3 { + t.Fatalf("run list has %d rows, want 3", len(listed)) } // The generic list returns full rows; page it with limit to prove the envelope carries nextCursor. page := f.call("GET", f.path("/workflows/"+wfid+"/runs")+"?limit=1", nil, "", 200) diff --git a/internal/collab/workflow-run-simulator.go b/internal/collab/workflow-run-simulator.go index 363921fad..f445e20a9 100644 --- a/internal/collab/workflow-run-simulator.go +++ b/internal/collab/workflow-run-simulator.go @@ -47,15 +47,18 @@ func (MockWorkflowRunSimulator) SimulateWorkflowRun(graph, input core.Object) (n } if isStart { // The one thing the sim can truthfully report: what it was kicked off with. Match the - // real Start output shape — `input` is the authored prompt and each run-form variable is - // a root — so a UI negative case reads the same pool the scheduler writes. + // real Start output shape — `input` is the run's kickoff (the submitted value, else the + // authored prompt) and each run-form variable is a root — so a UI negative case reads the + // same pool the scheduler writes. output := core.Object{} for key, value := range input { if key != "input" { output[key] = value } } - output["input"] = data.S("input") + if kickoff, ok := core.WorkflowStartKickoff(data, input); ok { + output["input"] = kickoff + } state["output"] = output } else { state["output"] = core.Object{"message": "Simulated " + title + " result (simulated; no real work was performed)."} diff --git a/internal/contract/openapi.go b/internal/contract/openapi.go index 633d53afb..f1021c45e 100644 --- a/internal/contract/openapi.go +++ b/internal/contract/openapi.go @@ -131,7 +131,7 @@ func Document() map[string]any { s["WorkflowRun"] = resource("id tenantId workflowId snapshotId name status input nodeStates rounds error startedAt finishedAt createdAt updatedAt workflowName", "startedAt finishedAt") workflowRunProps := properties(s, "WorkflowRun") workflowRunProps["status"] = enumeration("pending", "running", "awaiting_input", "succeeded", "failed", "cancelled") - workflowRunProps["input"] = obj{"type": "object", "additionalProperties": true, "description": "The kickoff input passed to the run's start node."} + workflowRunProps["input"] = obj{"type": "object", "additionalProperties": true, "description": "The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff — including every run created before the key was stored — evaluates against the Start node's authored instruction instead."} workflowRunProps["nodeStates"] = obj{"type": "object", "additionalProperties": true, "description": "Per-node execution state keyed by node id: status, displayName, startedAt, finishedAt, output, and error — the reason, present only when the node failed."} workflowRunProps["rounds"] = array(obj{"type": "object", "additionalProperties": true}) // definitionSnapshot lives only on a detail read (the history list trips without it); diff --git a/internal/core/workflow_admission.go b/internal/core/workflow_admission.go index e9b03de2d..2b9f3bf60 100644 --- a/internal/core/workflow_admission.go +++ b/internal/core/workflow_admission.go @@ -59,6 +59,16 @@ func workflowRunSnapshot(t *transaction, tid, wid, sid string) Object { // The check reads the snapshot's graph, never the live document, so a version published months ago // still accepts exactly what it accepted then. func requireWorkflowRunInput(graph, input Object) { + // `input` is the reserved kickoff selector, not a workflow variable. Desktop carries it on the + // creation request as a string (kickoff_input), so accepting another JSON type here would make + // `[start, input]` an object on Cloud and a string on desktop. A null is treated as absent, the + // same way the declared-variable check below treats it, and stays accepted for a run that named + // no override. + if v, ok := input["input"]; ok && v != nil { + if _, ok := v.(string); !ok { + rejectWith(400, "invalid_field_value", Object{"field": "input"}) + } + } fields := workflowInputFields(graph) for i := range fields { field := &fields[i] diff --git a/internal/core/workflow_admission_test.go b/internal/core/workflow_admission_test.go index ab1b50e9f..7909302f4 100644 --- a/internal/core/workflow_admission_test.go +++ b/internal/core/workflow_admission_test.go @@ -68,6 +68,24 @@ func TestRequireWorkflowRunInput(t *testing.T) { input: Object{"topic": "auth", "scope": "everything"}, want: &Fault{Code: "invalid_field_value", Status: 400, Params: Object{"field": "scope"}}, }, + { + // `input` is the reserved kickoff selector rather than a declared variable. Desktop + // carries it as a string, and accepting another type here would make [start, input] an + // object on Cloud and a string on desktop. + name: "the reserved kickoff selector as a string", + input: Object{"topic": "auth", "input": "review the diff"}, + }, + { + // A null override is a run that named none: it is treated as absent and falls back to + // the authored Start instruction, exactly like a declared variable left null. + name: "the reserved kickoff selector explicitly null", + input: Object{"topic": "auth", "input": nil}, + }, + { + name: "the reserved kickoff selector as an object", + input: Object{"topic": "auth", "input": Object{"text": "review the diff"}}, + want: &Fault{Code: "invalid_field_value", Status: 400, Params: Object{"field": "input"}}, + }, } for _, tc := range cases { diff --git a/internal/core/workflow_scheduler.go b/internal/core/workflow_scheduler.go index b41a035f8..2a3fd14cc 100644 --- a/internal/core/workflow_scheduler.go +++ b/internal/core/workflow_scheduler.go @@ -669,9 +669,10 @@ func executeWorkflowNode(node workflowNodeView, input Object, states map[string] // `[start, name]` resolved to nothing and `[start, input]` answered for the entire form — the same // document meant different things on the two sides. // -// The run form's values are keyed by variable name, so flattening them is exact. A key literally -// named `input` is dropped: desktop reserves that root, and letting a user variable shadow the -// kickoff selector would make the same selector mean a run value on Cloud and a prompt on desktop. +// The run form's values are keyed by variable name, so flattening them is exact. The `input` key is +// not flattened with the rest: it is the reserved kickoff selector, and letting a user variable +// shadow it would make the same selector mean a run value on Cloud and a prompt on desktop. Its value +// comes from WorkflowStartKickoff instead. func workflowStartOutput(node workflowNodeView, input Object) Object { output := make(Object, len(input)+1) for key, value := range input { @@ -680,15 +681,43 @@ func workflowStartOutput(node workflowNodeView, input Object) Object { } output[key] = value } - // Cloud does not persist the per-run prompt override, so the authored prompt is the only kickoff - // text available to the variable pool. The launch field still shows that same default. The text is - // used verbatim, whitespace included, because desktop's start.input is the kickoff prompt as the - // author wrote it; trimming here would make the same selector compare differently on the two - // sides. - output["input"] = node.Data.S("input") + if kickoff, ok := WorkflowStartKickoff(node.Data, input); ok { + output["input"] = kickoff + } return output } +// WorkflowStartKickoff resolves the Start node's kickoff text and reports whether it exists. The +// run's own submitted `input` wins; the authored Start instruction is the fallback. +// +// The two sources exist because runs outlive the code that created them. A run created before Cloud +// persisted the kickoff carries only the authored prompt in its input object, and the frozen snapshot +// is what it would have executed then, so falling back to that node's `data.input` keeps historical +// runs evaluating exactly as they did (circus#287). The fallback reads the snapshot's node, never the +// live workflow document. +// +// The text is used verbatim, whitespace included: desktop's start.input is the kickoff prompt as the +// author or the run submitted it, and trimming here would make the same selector compare differently +// on the two sides. When neither source has a value the root is left out of the pool rather than set +// to "": desktop leaves {start}.input unset when the run supplied no kickoff and the graph has no +// instruction, and exists/not_exists must answer the same way on both sides. +// +// It is exported for the development simulator, which writes the Start output outside this package; +// sharing the resolver keeps the simulated trace from drifting from what the scheduler commits. +func WorkflowStartKickoff(data, input Object) (string, bool) { + if v, ok := input["input"]; ok && v != nil { + if text, ok := v.(string); ok { + return text, true + } + } + if v, ok := data["input"]; ok && v != nil { + if text, ok := v.(string); ok { + return text, true + } + } + return "", false +} + // workflowNodeDisplayName is the label the run viewer shows: the author's title, or the node kind // when the title is empty — the same projection the simulator writes. func workflowNodeDisplayName(node workflowNodeView) string { diff --git a/internal/core/workflow_scheduler_db_test.go b/internal/core/workflow_scheduler_db_test.go index 8f1c85e12..8545e7f42 100644 --- a/internal/core/workflow_scheduler_db_test.go +++ b/internal/core/workflow_scheduler_db_test.go @@ -19,6 +19,13 @@ import ( // graph is the given document, returning the run id the assertions read back. The scene is minimal // on purpose: the advance pass reads workflow_runs joined to workflow_snapshots and nothing else. func seedWorkflowRunScene(t *testing.T, store *Store, graph Object) string { + t.Helper() + return seedWorkflowRunSceneWithInput(t, store, graph, Object{}) +} + +// seedWorkflowRunSceneWithInput is seedWorkflowRunScene with a run input, written as the jsonb the +// creation transaction stores so the pass reads exactly the document a real run carries. +func seedWorkflowRunSceneWithInput(t *testing.T, store *Store, graph, input Object) string { t.Helper() wid, sid, runID := uuid.NewString(), uuid.NewString(), uuid.NewString() tid, uid := uuid.NewString(), uuid.NewString() @@ -26,6 +33,10 @@ func seedWorkflowRunScene(t *testing.T, store *Store, graph Object) string { if err != nil { t.Fatalf("marshal graph: %v", err) } + in, err := json.Marshal(input) + if err != nil { + t.Fatalf("marshal run input: %v", err) + } tx, err := store.Pool.Begin() if err != nil { t.Fatalf("seed begin: %v", err) @@ -40,7 +51,7 @@ func seedWorkflowRunScene(t *testing.T, store *Store, graph Object) string { exec("membership", `INSERT INTO tenant_memberships(tenant_id,user_id,role,status) VALUES($1,$2,'admin','active')`, tid, uid) exec("workflow", `INSERT INTO workflows(id,tenant_id,name,graph) VALUES($1,$2,'wf',$3::jsonb)`, wid, tid, g) exec("snapshot", `INSERT INTO workflow_snapshots(id,tenant_id,workflow_id,version,name,graph) VALUES($1,$2,$3,1,'v1',$4::jsonb)`, sid, tid, wid, g) - exec("run", `INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status,input) VALUES($1,$2,$3,$4,'disp','pending','{}'::jsonb)`, runID, tid, wid, sid) + exec("run", `INSERT INTO workflow_runs(id,tenant_id,workflow_id,snapshot_id,name,status,input) VALUES($1,$2,$3,$4,'disp','pending',$5::jsonb)`, runID, tid, wid, sid, in) if err := tx.Commit(); err != nil { t.Fatalf("seed commit: %v", err) } @@ -66,8 +77,10 @@ func readWorkflowRun(t *testing.T, store *Store, runID string) (status, runErr s // run succeeded with every node's committed state on the row. func TestWorkflowAdvanceSucceedsLinearRun(t *testing.T) { store := dispatcherDB(t) + start := wfNode("s", "start", "") + start["data"].(Object)["input"] = "review the diff" runID := seedWorkflowRunScene(t, store, wfGraph( - []Object{wfNode("s", "start", ""), wfNode("o", "output", "done")}, + []Object{start, wfNode("o", "output", "done")}, [][3]string{{"s", "o", ""}}, )) @@ -90,8 +103,86 @@ func TestWorkflowAdvanceSucceedsLinearRun(t *testing.T) { t.Fatalf("node %s status = %q; want %q", id, state.S("status"), want) } } - if start := states["s"].(map[string]any)["output"].(map[string]any); start["input"] == nil { - t.Fatal("start node output does not carry the kickoff prompt root") + // No kickoff was submitted, so the authored Start instruction is the run's kickoff — the same + // fallback every row created before the kickoff was persisted depends on. + if got := states["s"].(map[string]any)["output"].(map[string]any)["input"]; got != "review the diff" { + t.Fatalf("start node output input = %v; want the authored prompt", got) + } +} + +// The kickoff stored on the run row is what [start, input] resolves, both as a selector and through +// a {{#start.input#}} expected-value template. The authored instruction is present too and +// different, so a pass that still read the document would select the other branches. +func TestWorkflowAdvanceStartInputUsesTheStoredKickoff(t *testing.T) { + store := dispatcherDB(t) + start := wfNode("s", "start", "") + start["data"].(Object)["input"] = "authored prompt" + rule := func(id string, selector []any, value any) Object { + return Object{"id": id, "logic": "and", "conditions": []any{ + Object{"variableSelector": selector, "operator": "equals", "value": value}, + }} + } + graph := wfGraph( + []Object{ + start, + wfConditionNode("c", []any{rule("literal", []any{"s", "input"}, "run override")}), + wfConditionNode("t", []any{rule("templated", []any{"s", "marker"}, "{{#s.input#}}")}), + wfNode("e", "output", ""), + wfNode("f", "output", ""), + wfNode("g", "output", ""), + wfNode("h", "output", ""), + }, + [][3]string{ + {"s", "c", ""}, + {"c", "e", "literal"}, + {"c", "f", workflowBranchElse}, + {"s", "t", ""}, + {"t", "g", "templated"}, + {"t", "h", workflowBranchElse}, + }, + ) + runID := seedWorkflowRunSceneWithInput(t, store, graph, + Object{"input": "run override", "marker": "run override"}) + + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance: %v", err) + } + _, _, states, _ := readWorkflowRun(t, store, runID) + branchOf := func(id string) string { + state, ok := asWorkflowObject(states[id]) + if !ok { + t.Fatalf("condition %s has no committed state", id) + } + return state.O("output").S("selectedBranch") + } + if got := branchOf("c"); got != "literal" { + t.Fatalf("condition selected %q; want literal — [start, input] did not read the stored kickoff", got) + } + if got := branchOf("t"); got != "templated" { + t.Fatalf("condition selected %q; want templated — {{#start.input#}} did not read the stored kickoff", got) + } + if got := states["s"].(map[string]any)["output"].(map[string]any)["input"]; got != "run override" { + t.Fatalf("start node output input = %v; want the stored kickoff", got) + } +} + +// With neither a stored kickoff nor an authored instruction {start}.input is unset, as it is in +// desktop's pool: `not_exists` matches and `exists` would not. It is not a present empty string, +// which would answer both existence operators the other way. +func TestWorkflowAdvanceConditionOnUnsetKickoffReadsAbsent(t *testing.T) { + store := dispatcherDB(t) + runID := seedWorkflowRunScene(t, store, wfGraph( + []Object{ + wfNode("s", "start", ""), + wfOneRuleNode("c", "absent", []any{"s", "input"}, "not_exists"), + wfNode("e", "output", ""), + wfNode("f", "output", ""), + }, + [][3]string{{"s", "c", ""}, {"c", "e", "absent"}, {"c", "f", workflowBranchElse}}, + )) + + if branch := wfSelectedBranch(t, store, runID); branch != "absent" { + t.Fatalf("condition selected %q; want absent — an unset kickoff read as present", branch) } } @@ -341,12 +432,15 @@ func wfOneRuleNode(id, caseID string, selector []any, operator string) Object { // wfConditionScene seeds a start -> condition -> {taken, else} graph whose single case is the given // operator over the given selector, and returns the run id. The run's input is {} — see -// seedWorkflowRunScene — which is the case the empty-input tests below turn on. +// seedWorkflowRunScene — and the Start node's authored prompt is empty, so {start}.input is present +// and empty: the case the empty-value tests below turn on. func wfConditionScene(t *testing.T, store *Store, caseID string, selector []any, operator string) string { t.Helper() + start := wfNode("s", "start", "") + start["data"].(Object)["input"] = "" return seedWorkflowRunScene(t, store, wfGraph( []Object{ - wfNode("s", "start", ""), + start, wfOneRuleNode("c", caseID, selector, operator), wfNode("e", "output", ""), wfNode("f", "output", ""), @@ -379,18 +473,16 @@ func wfNodeStatus(t *testing.T, states Object, nodeID string) string { return state.S("status") } -// A condition reading the run's own input sees an empty input as empty. This is the end-to-end shape -// of the named-type trap: the start node nests the run input as the package's Object, so the value a -// selector walks out is an Object too, and a type switch that only names map[string]any falls -// through it. Before asWorkflowObject was used there, `empty` answered false on `{}` and the run took -// the branch desktop's is_empty would not have taken — a legal start -> condition graph, decided -// wrong, with nothing in the run to say so. -func TestWorkflowAdvanceConditionOnEmptyRunInput(t *testing.T) { +// An authored Start instruction that is empty is present and empty, not unset: [start, input] is +// the empty string, so `empty` matches. The distinction matters because an unset root is a +// different state — `not_exists` matches it and `exists` does not — which the unset-kickoff test +// below pins. +func TestWorkflowAdvanceConditionOnEmptyAuthoredPrompt(t *testing.T) { store := dispatcherDB(t) runID := wfConditionScene(t, store, "was_empty", []any{"s", "input"}, "empty") if branch := wfSelectedBranch(t, store, runID); branch != "was_empty" { - t.Fatalf("condition selected %q; want was_empty — an empty run input read as not-empty", branch) + t.Fatalf("condition selected %q; want was_empty — an empty authored prompt read as not-empty", branch) } _, _, states, _ := readWorkflowRun(t, store, runID) if got := wfNodeStatus(t, states, "e"); got != "succeeded" { @@ -399,18 +491,18 @@ func TestWorkflowAdvanceConditionOnEmptyRunInput(t *testing.T) { // The untaken branch is committed as inactive rather than left absent: the run says which branch // it did not take, which is what makes a wrong decision visible after the fact. if got := wfNodeStatus(t, states, "f"); got != "inactive" { - t.Fatalf("else branch status = %q; want inactive — the else branch ran on an empty input", got) + t.Fatalf("else branch status = %q; want inactive — the else branch ran on an empty prompt", got) } } // not_empty is empty's negation on the same document: the same start -> condition graph, the same -// empty input, and the case must not match. -func TestWorkflowAdvanceConditionNotEmptyOnEmptyRunInput(t *testing.T) { +// empty authored prompt, and the case must not match. +func TestWorkflowAdvanceConditionNotEmptyOnEmptyAuthoredPrompt(t *testing.T) { store := dispatcherDB(t) runID := wfConditionScene(t, store, "was_full", []any{"s", "input"}, "not_empty") if branch := wfSelectedBranch(t, store, runID); branch != workflowBranchElse { - t.Fatalf("condition selected %q; want else — an empty run input read as not-empty", branch) + t.Fatalf("condition selected %q; want else — an empty authored prompt read as not-empty", branch) } } diff --git a/internal/core/workflow_scheduler_test.go b/internal/core/workflow_scheduler_test.go index 6a8cb713a..ff38c9397 100644 --- a/internal/core/workflow_scheduler_test.go +++ b/internal/core/workflow_scheduler_test.go @@ -426,16 +426,69 @@ func TestWorkflowConditionCaseLogic(t *testing.T) { // The start node's output is the variable pool desktop declares: `input` is the kickoff prompt and // each run-form variable is its own root, so selector [start, name] resolves the value the author -// selected in the editor instead of walking the whole form under [start, input, name]. The prompt is -// kept verbatim, whitespace included: desktop's start.input is the authored text, and trimming it -// here would let the same selector compare differently on the two sides. +// selected in the editor instead of walking the whole form under [start, input, name]. The run's +// submitted kickoff wins over the authored instruction; a run that named none falls back to the +// authored text (which is also what every run created before the kickoff was persisted has), and a +// graph with neither leaves the root unset. Every value is kept verbatim, whitespace included: +// desktop's start.input is the text as written, and trimming it here would let the same selector +// compare differently on the two sides. func TestWorkflowStartOutputMatchesDesktopVariableRoots(t *testing.T) { - node := wfNode("s", "start", "") - node["data"].(Object)["input"] = " review the diff " - output := workflowStartOutput(workflowNodeView{ID: "s", Kind: "start", Data: node["data"].(Object)}, - Object{"flag": true, "count": 2.0, "input": "shadowed"}) - want := Object{"flag": true, "count": 2.0, "input": " review the diff "} - if canonicalJSON(output) != canonicalJSON(want) { - t.Fatalf("start output = %s; want %s", canonicalJSON(output), canonicalJSON(want)) + startData := func(authored any) Object { + data := Object{"kind": "start"} + if authored != nil { + data["input"] = authored + } + return data + } + cases := []struct { + name string + data Object + input Object + want Object + }{ + { + name: "the run's submitted kickoff wins", + data: startData(" review the diff "), + input: Object{"flag": true, "count": 2.0, "input": "run override"}, + want: Object{"flag": true, "count": 2.0, "input": "run override"}, + }, + { + // Historical rows carry no kickoff key; the authored instruction in the frozen snapshot + // is what they executed with, so that is what their pool must keep resolving. + name: "a run with no submitted kickoff falls back to the authored prompt", + data: startData(" review the diff "), + input: Object{"flag": true, "count": 2.0}, + want: Object{"flag": true, "count": 2.0, "input": " review the diff "}, + }, + { + name: "an explicit null override falls back to the authored prompt", + data: startData("review the diff"), + input: Object{"input": nil}, + want: Object{"input": "review the diff"}, + }, + { + // An empty submitted kickoff is a value, not an absence: it must not silently become + // the authored prompt the user cleared. + name: "an empty submitted kickoff is kept", + data: startData("review the diff"), + input: Object{"input": ""}, + want: Object{"input": ""}, + }, + { + // Desktop leaves {start}.input unset when the run supplied no kickoff and the graph has + // no instruction, so exists/not_exists must see the same absence here. + name: "neither source leaves the root unset", + data: startData(nil), + input: Object{"flag": true}, + want: Object{"flag": true}, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + output := workflowStartOutput(workflowNodeView{ID: "s", Kind: "start", Data: tc.data}, tc.input) + if canonicalJSON(output) != canonicalJSON(tc.want) { + t.Fatalf("start output = %s; want %s", canonicalJSON(output), canonicalJSON(tc.want)) + } + }) } } From f9731b1069d32e83b2064b7d427fc165472f245c Mon Sep 17 00:00:00 2001 From: Zpc001 <1903085231@qq.com> Date: Tue, 20 Oct 2026 09:47:17 +0800 Subject: [PATCH 2/2] fix(core): shadow a Start variable named `input` under the reserved kickoff A graph declaring `input` as a non-string Start variable was unstartable: the reserved kickoff check demanded a string while the declared-field check demanded the declared type, so neither value passed. Skip the declared-field check for `input`, matching workflowStartOutput and desktop, where the kickoff overwrites a declared variable of the same name. Pin it with a test. Also correct the admission and scheduler doc comments, and soften the OpenAPI and PR wording that overclaimed historical rows lack the key. Co-Authored-By: Claude Code --- api/openapi.json | 2 +- frontend/src/api/generated.schemas.ts | 4 ++-- frontend/src/features/workflows/api.ts | 2 +- internal/contract/openapi.go | 2 +- internal/core/workflow_admission.go | 27 ++++++++++++++++++------ internal/core/workflow_admission_test.go | 24 +++++++++++++++++++++ internal/core/workflow_scheduler.go | 10 +++++---- 7 files changed, 56 insertions(+), 15 deletions(-) diff --git a/api/openapi.json b/api/openapi.json index 18a42d06d..47a136027 100644 --- a/api/openapi.json +++ b/api/openapi.json @@ -5301,7 +5301,7 @@ }, "input": { "additionalProperties": true, - "description": "The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff — including every run created before the key was stored — evaluates against the Start node's authored instruction instead.", + "description": "The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff evaluates against the Start node's authored instruction instead.", "type": "object" }, "modelBindings": { diff --git a/frontend/src/api/generated.schemas.ts b/frontend/src/api/generated.schemas.ts index 8642cfcb8..b942d10bb 100644 --- a/frontend/src/api/generated.schemas.ts +++ b/frontend/src/api/generated.schemas.ts @@ -2373,7 +2373,7 @@ export interface WorkflowAssetImportList { export type WorkflowRunDefinitionSnapshot = { [key: string]: unknown } | null; /** - * The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff — including every run created before the key was stored — evaluates against the Start node's authored instruction instead. + * The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff evaluates against the Start node's authored instruction instead. */ export type WorkflowRunInput = { [key: string]: unknown }; @@ -2415,7 +2415,7 @@ export interface WorkflowRun { /** @nullable */ finishedAt: string | null; id: string; - /** The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff — including every run created before the key was stored — evaluates against the Start node's authored instruction instead. */ + /** The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff evaluates against the Start node's authored instruction instead. */ input: WorkflowRunInput; /** * The frozen per-slot model bindings this run executes against, ordered by slot name. Present on a detail read. Each binding is frozen at run creation and used for the run's whole lifetime; only `revoked` can turn true (the owning connection's grant was revoked). diff --git a/frontend/src/features/workflows/api.ts b/frontend/src/features/workflows/api.ts index 485d114c7..29e04d35e 100644 --- a/frontend/src/features/workflows/api.ts +++ b/frontend/src/features/workflows/api.ts @@ -375,7 +375,7 @@ export interface CreateWorkflowRunInput { name?: string /** The exact snapshot to execute, or omit to pin the latest published one. */ snapshotId?: string - /** Kickoff input delivered to the start node. */ + /** The run's input object: form values keyed by variable name, plus the reserved `input` kickoff text. */ input?: Record } diff --git a/internal/contract/openapi.go b/internal/contract/openapi.go index f1021c45e..b1e2f181f 100644 --- a/internal/contract/openapi.go +++ b/internal/contract/openapi.go @@ -131,7 +131,7 @@ func Document() map[string]any { s["WorkflowRun"] = resource("id tenantId workflowId snapshotId name status input nodeStates rounds error startedAt finishedAt createdAt updatedAt workflowName", "startedAt finishedAt") workflowRunProps := properties(s, "WorkflowRun") workflowRunProps["status"] = enumeration("pending", "running", "awaiting_input", "succeeded", "failed", "cancelled") - workflowRunProps["input"] = obj{"type": "object", "additionalProperties": true, "description": "The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff — including every run created before the key was stored — evaluates against the Start node's authored instruction instead."} + workflowRunProps["input"] = obj{"type": "object", "additionalProperties": true, "description": "The run's input object: each form value keyed by variable name, plus the reserved `input` key carrying the kickoff text the run submitted. A run that submitted no kickoff evaluates against the Start node's authored instruction instead."} workflowRunProps["nodeStates"] = obj{"type": "object", "additionalProperties": true, "description": "Per-node execution state keyed by node id: status, displayName, startedAt, finishedAt, output, and error — the reason, present only when the node failed."} workflowRunProps["rounds"] = array(obj{"type": "object", "additionalProperties": true}) // definitionSnapshot lives only on a detail read (the history list trips without it); diff --git a/internal/core/workflow_admission.go b/internal/core/workflow_admission.go index 2b9f3bf60..918011c28 100644 --- a/internal/core/workflow_admission.go +++ b/internal/core/workflow_admission.go @@ -49,12 +49,19 @@ func workflowRunSnapshot(t *transaction, tid, wid, sid string) Object { // requireWorkflowRunInput refuses a run whose input does not answer the variables the frozen graph // declares: a required Start variable with no value, or a value of the wrong type for its control. // -// Only the declared variables are checked, and keys the graph does not declare are left alone. A run's -// input is an execution input object, not a form submission: it carries the platform's own material -// (context references, the issue context, a frozen executor reference) alongside the workflow's -// parameters, so treating an extra key as an error would refuse input the platform itself writes. The -// @ form is the one place where the set of keys *is* the contract, and it rejects unknown keys on its -// own — see validateFormValues. +// The reserved `input` kickoff selector is checked before the declared variables: when present it must +// be a string, and a null is treated as absent, the way desktop's kickoff_input is. Then only the +// declared variables are checked, and keys the graph does not declare are left alone. A run's input is +// an execution input object, not a form submission: it carries the platform's own material (context +// references, the issue context, a frozen executor reference) alongside the workflow's parameters, so +// treating an extra key as an error would refuse input the platform itself writes. The @ form is the +// one place where the set of keys *is* the contract, and it rejects unknown keys on its own — see +// validateFormValues. +// +// A variable that declares the name `input` is shadowed by the reserved selector: the kickoff carries +// the value `[start, input]` resolves to (see workflowStartOutput), so the declared variable's type is +// never judged here — desktop likewise lets the kickoff overwrite a declared variable of the same name +// rather than refusing it. // // The check reads the snapshot's graph, never the live document, so a version published months ago // still accepts exactly what it accepted then. @@ -72,6 +79,14 @@ func requireWorkflowRunInput(graph, input Object) { fields := workflowInputFields(graph) for i := range fields { field := &fields[i] + // A declared variable named `input` is shadowed by the reserved selector, whose value the Start + // output resolves from WorkflowStartKickoff rather than from this field. Desktop lets the kickoff + // overwrite a declared variable of the same name, so judging the declared type here would make a + // legal graph unstartable: a string kickoff fails a non-string declaration, and a value of the + // declared type fails the reserved-string check above. + if field.Key == "input" { + continue + } v, present := input[field.Key] if !present || v == nil { if field.Required { diff --git a/internal/core/workflow_admission_test.go b/internal/core/workflow_admission_test.go index 7909302f4..84080d323 100644 --- a/internal/core/workflow_admission_test.go +++ b/internal/core/workflow_admission_test.go @@ -102,6 +102,30 @@ func TestRequireWorkflowRunInput(t *testing.T) { } } +// TestRequireWorkflowRunInputKickoffShadowsInputVariable pins the desktop shadow semantics for a Start +// variable literally named `input`: the reserved kickoff selector overwrites the declared variable, so +// the declared type is never judged while the reserved key stays string-only. Without the shadow a +// graph declaring `input` as a number would reject every run — a string kickoff fails the declared +// type, and a number fails the reserved-string check — a Cloud-only dead end desktop runs fine. +func TestRequireWorkflowRunInputKickoffShadowsInputVariable(t *testing.T) { + graph := inputGraph(t, []any{ + Object{"name": "topic", "valueType": "string", "required": true}, + Object{"name": "input", "valueType": "number"}, + }) + + // A string `input` is the kickoff and is admitted even though the declared `input` variable is a + // number: the reserved selector shadows it, so the number's type is not judged. + noFault(t, func() { requireWorkflowRunInput(graph, Object{"topic": "auth", "input": "review the diff"}) }) + + // The reserved key is still string-only: a number fails the reserved check even though a declared + // `input` variable of that type would have been legal on its own. + fault := faultOf(t, func() { requireWorkflowRunInput(graph, Object{"topic": "auth", "input": 7}) }) + want := &Fault{Code: "invalid_field_value", Status: 400, Params: Object{"field": "input"}} + if !reflect.DeepEqual(fault, want) { + t.Fatalf("fault = %+v, want %+v", fault, want) + } +} + // TestRequireWorkflowRunInputReadsTheGraphItIsGiven covers the reason the check takes a graph rather // than reading the live document: the schema that governs a run is the one the version it executes // was published with. A variable the live document added after that publish must not be demanded of diff --git a/internal/core/workflow_scheduler.go b/internal/core/workflow_scheduler.go index 2a3fd14cc..c3eefbe8a 100644 --- a/internal/core/workflow_scheduler.go +++ b/internal/core/workflow_scheduler.go @@ -691,10 +691,12 @@ func workflowStartOutput(node workflowNodeView, input Object) Object { // run's own submitted `input` wins; the authored Start instruction is the fallback. // // The two sources exist because runs outlive the code that created them. A run created before Cloud -// persisted the kickoff carries only the authored prompt in its input object, and the frozen snapshot -// is what it would have executed then, so falling back to that node's `data.input` keeps historical -// runs evaluating exactly as they did (circus#287). The fallback reads the snapshot's node, never the -// live workflow document. +// persisted the per-run kickoff has no reserved `input` key in its own input object — the prompt it +// executed then lived in the frozen snapshot's Start node, not the run row — so falling back to that +// node's `data.input` keeps historical runs evaluating exactly as they did (circus#287). The fallback +// reads the snapshot's node, never the live workflow document. Cloud's canonical document stores the +// prompt under `data.input` only; the legacy `data.instruction` desktop also reads is migrated to +// `input` by the frontend when a document is read, so the server never has to fall back to it. // // The text is used verbatim, whitespace included: desktop's start.input is the kickoff prompt as the // author or the run submitted it, and trimming here would make the same selector compare differently