From 12a3fe4513f68949f1154507e63c47e2c5d65bfc Mon Sep 17 00:00:00 2001 From: Kyber5323 Date: Fri, 16 Oct 2026 09:33:53 +0800 Subject: [PATCH 1/6] feat: Cloud-side workflow scheduler advance pass Connect WorkflowRun to the first real execution step on the Cloud side (specs decisions/cloud/workflow 0-cloud-side-scheduler-owns-workflow-execution, landing order step 3, circus#70; run issue circus#141). A bounded retry-loop pass (limit 100, ordered created_at,id) claims pending workflow_runs and advances running ones, each in its own short transaction. The ready set is recomputed from committed state only: the frozen snapshot graph plus the node_states column, resolved to fixed point exactly like desktop's BranchProjection (unlabeled edge = else, condition edges matched by the committed selectedBranch, rootless non-start nodes inactive, pending upstream leaves downstream unresolved). Node results commit behind the run's version CAS, so a late or duplicate pass is a no-op and a replica that crashes mid-pass resumes from committed state. start/condition/output execute inline: start reports the run input, condition evaluates its cases against prior node outputs (desktop condition.rs operator set, first matching case wins, else fallback) and commits the selected branch, output marks the terminal node. An agent node is the phase-1 boundary: until the execution wiring lands it settles the run failed with reason agent_execution_not_wired - the honest terminal state, never a hang in running and never a success for work that did not happen. The loop is owned by cmd/server's process lifecycle like the other ten-second recovery loops. Tests cover the projection and condition semantics as pure core, plus DB tests on isolated schemas for claim+settle, the agent boundary and idempotence. Co-Authored-By: Claude Code --- cmd/server/main.go | 14 + internal/core/workflow_condition.go | 248 ++++++++++ internal/core/workflow_scheduler.go | 523 ++++++++++++++++++++ internal/core/workflow_scheduler_db_test.go | 156 ++++++ internal/core/workflow_scheduler_test.go | 279 +++++++++++ 5 files changed, 1220 insertions(+) create mode 100644 internal/core/workflow_condition.go create mode 100644 internal/core/workflow_scheduler.go create mode 100644 internal/core/workflow_scheduler_db_test.go create mode 100644 internal/core/workflow_scheduler_test.go diff --git a/cmd/server/main.go b/cmd/server/main.go index 0ff1a450f..d8210dbec 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -210,6 +210,20 @@ func run() (runErr error) { pluginmarket.RunSyncLoop(ctx, store.GiveUpStaleWorkspaceReleasesOnce, agentDeliveryInterval, pluginmarket.ContextSleep, log) }() } + // Workflow advance loop: the Cloud-side scheduler (specs#99 D1 条件②/④, landing order step 3). + // A bounded rescan of pending and running workflow runs, each advanced in its own short + // transaction; the ready set is recomputed from committed state only, so the pass is idempotent + // and a replica that crashes mid-tick resumes exactly where the committed state says. In its + // first step it executes start/condition/output inline and settles runs honestly; agent nodes + // fail the run fast with an explicit reason until the execution wiring lands in the next step. + { + const workflowAdvanceInterval = 10 * time.Second + syncGroup.Add(1) + go func() { + defer syncGroup.Done() + pluginmarket.RunSyncLoop(ctx, store.AdvanceWorkflowRunsOnce, workflowAdvanceInterval, pluginmarket.ContextSleep, log) + }() + } gin.SetMode(cfg.Server.Mode) server := &http.Server{Addr: fmt.Sprintf(":%d", cfg.Server.Port), Handler: router.New(store, auth, log, directory), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: cfg.Server.ReadTimeout, WriteTimeout: cfg.Server.WriteTimeout, IdleTimeout: 60 * time.Second} // The control listener is bound before serving so a taken port fails startup, not a Controller. diff --git a/internal/core/workflow_condition.go b/internal/core/workflow_condition.go new file mode 100644 index 000000000..ef0d0c0b0 --- /dev/null +++ b/internal/core/workflow_condition.go @@ -0,0 +1,248 @@ +package core + +import "strings" + +// Workflow Condition node semantics, ported from desktop's engine/condition.rs so a branch decision +// means the same thing on both sides (specs#99 D1 条件④: the scheduler boundary mirrors desktop's +// ports, and desktop is the reference implementation). +// +// The variable pool is the run's committed node outputs: a rule's variableSelector [nodeId, ...path] +// resolves by reading prior node states — selector[0] names the node, selector[1:] walks into its +// output object. A variable a prior node has not produced (unknown node, absent path, or a node that +// never ran) is absent, which the operators handle deliberately: exists / not_exists / empty / +// not_empty classify absence, and every other operator compares nothing — a comparator on an absent +// variable does not match. None of this is a run failure: an unevaluable condition has an else +// branch to fall back to, and a failed run should name a real execution failure, not a document the +// author must fix anyway. + +// workflowConditionOperator evaluates one comparison rule. absent reports whether the selector +// resolved to nothing; resolved is the variable when it is present. +type workflowConditionOperator = func(resolved any, absent bool, expected any) bool + +// workflowConditionOperators is desktop's full wire operator set (condition.rs +// ComparisonOperator::parse). `is`/`is_not` are the same comparisons as `equals`/`not_equals`. +var workflowConditionOperators = map[string]workflowConditionOperator{ + "equals": workflowCompareEquals, + "is": workflowCompareEquals, + "not_equals": workflowCompareNotEquals, + "is_not": workflowCompareNotEquals, + "contains": workflowCompareContains, + "not_contains": workflowCompareNegate(workflowCompareContains), + "starts_with": workflowCompareTextOrder("starts"), + "ends_with": workflowCompareTextOrder("ends"), + "empty": workflowCompareEmpty, + "not_empty": workflowCompareNegate(workflowCompareEmpty), + "exists": workflowCompareExists, + "not_exists": workflowCompareNegate(workflowCompareExists), + "greater_than": workflowCompareNumeric("greater"), + "greater_than_or_equal": workflowCompareNumeric("greater_eq"), + "less_than": workflowCompareNumeric("less"), + "less_than_or_equal": workflowCompareNumeric("less_eq"), +} + +func workflowCompareEquals(resolved any, absent bool, expected any) bool { + if absent { + return false + } + return workflowJSONEqual(resolved, expected) +} + +func workflowCompareNotEquals(resolved any, absent bool, expected any) bool { + if absent { + return false + } + return !workflowJSONEqual(resolved, expected) +} + +func workflowCompareContains(resolved any, absent bool, expected any) bool { + if absent { + return false + } + return strings.Contains(workflowTextOf(resolved), workflowTextOf(expected)) +} + +// workflowCompareTextOrder serves both text-position operators: "starts" is prefix, "ends" is +// suffix; any other direction is unreachable. +func workflowCompareTextOrder(direction string) workflowConditionOperator { + return func(resolved any, absent bool, expected any) bool { + if absent { + return false + } + resolvedText, expectedText := workflowTextOf(resolved), workflowTextOf(expected) + switch direction { + case "starts": + return strings.HasPrefix(resolvedText, expectedText) + default: + return strings.HasSuffix(resolvedText, expectedText) + } + } +} + +func workflowCompareEmpty(resolved any, absent bool, expected any) bool { + if absent { + return true // absence is the emptiest a variable can be + } + switch value := resolved.(type) { + case nil: + return true + case string: + return value == "" + case []any: + return len(value) == 0 + case map[string]any: + return len(value) == 0 + default: + return false + } +} + +func workflowCompareExists(resolved any, absent bool, expected any) bool { + return !absent +} + +// workflowCompareNegate flips one operator's verdict. The absent-handling stays the inner +// operator's, so not_exists matches exactly what exists does not. +func workflowCompareNegate(inner workflowConditionOperator) workflowConditionOperator { + return func(resolved any, absent bool, expected any) bool { + return !inner(resolved, absent, expected) + } +} + +// workflowTextOf renders a JSON value as the text a text operator compares. A string is itself; +// anything else is its canonical JSON, so comparing against a number or object stays deterministic. +func workflowTextOf(value any) string { + if text, ok := value.(string); ok { + return text + } + return canonicalJSON(value) +} + +// workflowJSONEqual compares two JSON values by canonical form (encoding/json sorts map keys), so +// object comparison is order-independent and deterministic. A value JSON cannot represent cannot be +// compared — that is a no-match, not a panic. +func workflowJSONEqual(resolved, expected any) bool { + a, b := canonicalJSON(resolved), canonicalJSON(expected) + if a == "" || b == "" { + return false + } + return a == b +} + +// workflowCompareNumeric serves the four ordering operators. Only two numbers are orderable; text +// against a number is a no-match rather than an error. +func workflowCompareNumeric(direction string) workflowConditionOperator { + return func(resolved any, absent bool, expected any) bool { + if absent { + return false + } + resolvedNum, okA := resolved.(float64) + expectedNum, okB := expected.(float64) + if !okA || !okB { + return false + } + switch direction { + case "greater": + return resolvedNum > expectedNum + case "greater_eq": + return resolvedNum >= expectedNum + case "less": + return resolvedNum < expectedNum + default: + return resolvedNum <= expectedNum + } + } +} + +// evaluateWorkflowCondition evaluates a Condition node's cases against the committed node states and +// returns the selected branch id: the first case whose rules match under the case's logic, or `else` +// when none did. +func evaluateWorkflowCondition(node workflowNodeView, states map[string]workflowNodeState) string { + cases, ok := node.Data["cases"].([]any) + if !ok { + return workflowBranchElse + } + for _, raw := range cases { + c, ok := asWorkflowObject(raw) + if !ok { + continue + } + rules, _ := c["conditions"].([]any) + if len(rules) == 0 { + continue // an authored shell with no rule yet falls through + } + matched := false + if c.S("logic") == "or" { + for _, raw := range rules { + if workflowConditionRule(raw, states) { + matched = true + break + } + } + } else { + matched = true + for _, raw := range rules { + if !workflowConditionRule(raw, states) { + matched = false + break + } + } + } + if matched { + return c.S("id") + } + } + return workflowBranchElse +} + +// workflowConditionRule resolves a rule's variableSelector and applies its operator. A malformed +// rule (not an object, malformed selector, unknown operator) is a no-match. +func workflowConditionRule(raw any, states map[string]workflowNodeState) bool { + r, ok := asWorkflowObject(raw) + if !ok { + return false + } + selector, ok := r["variableSelector"].([]any) + if !ok || len(selector) == 0 { + return false + } + nodeID, ok := selector[0].(string) + if !ok || nodeID == "" { + return false + } + state, ok := states[nodeID] + if !ok || state.Status != "succeeded" { + return false + } + resolved, absent := walkWorkflowOutputPath(state.Output, selector[1:]) + operator, ok := workflowConditionOperators[r.S("operator")] + if !ok { + return false + } + return operator(resolved, absent, r["value"]) +} + +// walkWorkflowOutputPath walks a variableSelector's path into one node's output. The first path +// element indexes the output object; deeper elements walk further in. Absent is the honest answer +// for a node whose output is not an object or a path that leaves the object. +func walkWorkflowOutputPath(output Object, path []any) (any, bool) { + if output == nil { + return nil, true + } + current := any(output) + for _, element := range path { + obj, ok := asWorkflowObject(current) + if !ok { + return nil, true + } + key, ok := element.(string) + if !ok { + return nil, true + } + value, present := obj[key] + if !present { + return nil, true + } + current = value + } + return current, false +} diff --git a/internal/core/workflow_scheduler.go b/internal/core/workflow_scheduler.go new file mode 100644 index 000000000..56d53518b --- /dev/null +++ b/internal/core/workflow_scheduler.go @@ -0,0 +1,523 @@ +package core + +import ( + "context" + "time" +) + +// The Cloud-side workflow scheduler: the stateless advance pass that moves a WorkflowRun from +// `pending` into real execution and settles it (specs decisions/cloud/workflow +// 0-cloud-side-scheduler-owns-workflow-execution.md, D1 条件②/④, landing order step 3). +// +// Shape. Every tick recomputes the ready set from committed state only — the node states already +// persisted on the run row plus the frozen snapshot graph — so a replica that crashes mid-pass loses +// nothing and a repeated pass is a no-op. Node results commit as a whole-row compare-and-swap on +// `version` (invariant 3: at most one serial executor per run; late or duplicate advancement is a +// no-op), and a node without a committed result is never recorded as succeeded (invariant 5). +// +// Phase-1 boundary. The executable kinds are start, agent, condition and output. start, condition +// and output commit their results inline: they are pure functions of committed state. An agent node +// needs a run Workspace and a real Node session through the control plane, which lands in the next +// step of the same decision (运行身份与绑定, then the execution wiring); until that step is in place a +// ready agent node settles the run failed with an explicit reason. That is the honest terminal state: +// production runs with agent nodes fail fast with a message that names the gap, and never hang in +// `running`, and never report success for work that did not happen. Dev fixtures keep the simulator, +// which fills the trace at create time, so development behavior is unchanged. + +// workflowAdvanceBatchSize bounds the retry-loop scan so one tick advances at most this many runs. +// Ordering is deterministic (created_at, id) so every tick drains the oldest first; runs beyond the +// bound advance on the next tick. +const workflowAdvanceBatchSize = 100 + +// workflowBranchElse is the branch id a Condition node falls back to when no case matched, and the +// meaning of an outgoing edge with no sourceHandle (desktop branch_projection: an unlabeled edge +// defaults to else so graphs that predate source handles still route somewhere safe). +const workflowBranchElse = "else" + +// asWorkflowObject accepts a JSON document as either the package's Object or a plain +// map[string]any. Object is a named type, so an Object boxed in `any` does not assert to +// map[string]any — and the scheduler reads documents from two worlds: jsonb-decoded rows produce +// plain maps, while hand-built documents (tests, callers passing literals) carry Object. Asserting +// one shape silently no-matches the other, so every document-level assertion goes through here. +func asWorkflowObject(v any) (Object, bool) { + switch o := v.(type) { + case Object: + return o, true + case map[string]any: + return Object(o), true + default: + return nil, false + } +} + +// scanAdvancableWorkflowRuns returns the bounded, deterministically ordered ids of runs the pass can +// advance: `pending` (to claim) and `running` (to progress). Terminal and awaiting runs are never +// rescanned. The scan is read-only and runs outside the advisory lock; claim correctness lives in +// each advance transaction's status CAS. +func (s *Store) scanAdvancableWorkflowRuns(ctx context.Context, limit int) ([]string, error) { + rows, err := s.Pool.QueryContext(ctx, ` + SELECT id FROM workflow_runs + WHERE status IN ('pending','running') AND deleted_at IS NULL + ORDER BY created_at, id + LIMIT $1`, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// AdvanceWorkflowRunsOnce is one bounded retry-loop pass: advance each claimable or in-flight run in +// its own short transaction, with the sleep between ticks owned by the caller. Per-run failures are +// deliberately not surfaced here — the transaction already rolls back and the run is retried on the +// next tick — so the loop must not flood logs because one run is transiently stuck. Only a scan-level +// failure cancels the pass. +func (s *Store) AdvanceWorkflowRunsOnce(ctx context.Context) error { + if ctx.Err() != nil { + return ctx.Err() + } + ids, err := s.scanAdvancableWorkflowRuns(ctx, workflowAdvanceBatchSize) + if err != nil { + return err + } + for _, id := range ids { + if _, err := s.transact(ctx, func(t *transaction) Object { + advanceWorkflowRun(t, id) + return Object{} + }); err != nil { + _ = err + } + } + return nil +} + +// workflowNodeView is one node of the frozen snapshot graph the scheduler executes. +type workflowNodeView struct { + ID string + Kind string + Data Object +} + +// workflowGraphView is the frozen graph reduced to the shapes the scheduler reads. +type workflowGraphView struct { + Nodes []workflowNodeView + Edges []Object + ByNode map[string]workflowNodeView + Start string // the one start node's id, or "" when the graph has none +} + +// readWorkflowGraph reduces a snapshot graph document into its nodes and edges. The document is +// data the scheduler interprets for the first time here, so malformed entries are dropped rather +// than exploding: a graph that reaches a run through a published snapshot was judged at publish, +// and snapshots predating that gate still get an honest run rather than a panic. +func readWorkflowGraph(graph Object) workflowGraphView { + out := workflowGraphView{ByNode: map[string]workflowNodeView{}} + // The comma-ok form keeps the promise above: a document whose nodes are not a list at all is an + // empty graph, not a panic. + nodes, _ := graph["nodes"].([]any) + for _, raw := range nodes { + o, ok := asWorkflowObject(raw) + if !ok { + continue + } + id := o.S("id") + if id == "" { + continue + } + if _, seen := out.ByNode[id]; seen { + continue + } + view := workflowNodeView{ID: id, Kind: o.O("data").S("kind"), Data: o.O("data")} + out.Nodes = append(out.Nodes, view) + out.ByNode[id] = view + if view.Kind == "start" && out.Start == "" { + out.Start = id + } + } + edges, _ := graph["edges"].([]any) + for _, raw := range edges { + e, ok := asWorkflowObject(raw) + if !ok { + continue + } + if e.S("source") == "" || e.S("target") == "" { + continue + } + if _, sourceKnown := out.ByNode[e.S("source")]; !sourceKnown { + continue + } + if _, targetKnown := out.ByNode[e.S("target")]; !targetKnown { + continue + } + out.Edges = append(out.Edges, e) + } + return out +} + +// workflowNodeState is one node's committed execution state on the run row. The shape is the one the +// run viewer already renders (the simulator writes the same keys): status is the shared run/node +// vocabulary, output carries what the node produced, and an inactive state marks a node behind a +// branch that was never taken — it never ran and never will. +type workflowNodeState struct { + Status string + DisplayName string + StartedAt string + FinishedAt string + Output Object +} + +// object renders the state in the shape the run viewer reads (and the simulator writes). +func (s workflowNodeState) object() Object { + out := Object{"status": s.Status, "displayName": s.DisplayName, "startedAt": s.StartedAt, "finishedAt": s.FinishedAt} + if s.Output != nil { + out["output"] = s.Output + } + return out +} + +// workflowStatesFromJSON decodes the committed node_states column. A row the column cannot interpret +// contributes nothing rather than failing the pass: the states are re-derived from whatever is +// readable, and a truncated state only costs a re-execution of the affected node, never a wrong +// success — a node without a committed result is never recorded as succeeded. +func workflowStatesFromJSON(raw Object) map[string]workflowNodeState { + out := map[string]workflowNodeState{} + for id, value := range raw { + node, ok := asWorkflowObject(value) + if !ok { + continue + } + state := workflowNodeState{ + Status: node.S("status"), + DisplayName: node.S("displayName"), + StartedAt: node.S("startedAt"), + FinishedAt: node.S("finishedAt"), + } + if output, ok := asWorkflowObject(node["output"]); ok { + state.Output = output + } + out[id] = state + } + return out +} + +// workflowProjection is one advance pass's view of the run: the frozen graph, the node states +// committed before the pass, and the Ready/Inactive resolution derived from them. It is the Go +// counterpart of desktop's BranchProjection (branch_projection.rs), restricted to the phase-1 +// node kinds — no regions, no iterations, no loops: a run's graph is a DAG. +type workflowProjection struct { + graph workflowGraphView + committed map[string]workflowNodeState + projected map[string]string +} + +// projectWorkflowNodes resolves the projection to fixed point over the committed states: +// Ready/Inactive propagate downstream through each pass until nothing changes, exactly like +// desktop's resolve_projected_nodes. +func projectWorkflowNodes(graph workflowGraphView, committed map[string]workflowNodeState) *workflowProjection { + p := &workflowProjection{ + graph: graph, + committed: committed, + projected: map[string]string{}, + } + for id, state := range committed { + p.projected[id] = state.Status + } + for { + changed := false + for _, node := range graph.Nodes { + if _, known := p.projected[node.ID]; known { + continue + } + computed, resolved := p.nodeState(node.ID) + if !resolved { + continue + } + p.projected[node.ID] = computed + changed = true + } + if !changed { + break + } + } + return p +} + +// readyNode returns the next ready node in document order, or ok=false when the projection has no +// executable node left. This is the scheduler's ready set: recomputed from committed state only +// (D1 条件④), never from anything the pass remembers across transactions. +func (p *workflowProjection) readyNode() (workflowNodeView, bool) { + for _, node := range p.graph.Nodes { + if p.projected[node.ID] == "ready" { + return node, true + } + } + return workflowNodeView{}, false +} + +// hasFailure reports whether any committed node failed. +func (p *workflowProjection) hasFailure() bool { + for _, state := range p.committed { + if state.Status == "failed" { + return true + } + } + return false +} + +// nodeState derives one uncommitted node's projected state from its incoming edges: +// no incoming edges is ready only for the start node (a rootless non-start node never runs); any +// pending edge leaves the node unreached (not resolved); any failed edge fails it; any active edge +// readies it; all-inactive marks it inactive (behind a branch that was not taken). +func (p *workflowProjection) nodeState(nodeID string) (string, bool) { + incoming := p.incomingEdges(nodeID) + if len(incoming) == 0 { + if nodeID == p.graph.Start { + return "ready", true + } + return "inactive", true + } + anyActive, anyFailed, anyPending := false, false, false + for _, edge := range incoming { + switch p.edgeState(edge) { + case "pending": + anyPending = true + case "failed": + anyFailed = true + case "active": + anyActive = true + } + } + if anyPending { + return "", false // an upstream node is still unresolved: wait + } + if anyFailed { + return "failed", true + } + if anyActive { + return "ready", true + } + return "inactive", true +} + +// edgeState resolves one edge from its source node's projected state and, for a Condition, the +// branch it committed. A succeeded non-condition node activates all of its outgoing edges; a +// succeeded Condition activates only the edge whose sourceHandle is the committed branch (an edge +// with no handle is the else edge); a failed source fails the edge; an inactive source stays +// inactive. +func (p *workflowProjection) edgeState(edge Object) string { + switch p.projected[edge.S("source")] { + case "succeeded": + source := p.graph.ByNode[edge.S("source")] + if source.Kind != "condition" { + return "active" + } + selected := p.selectedBranch(source) + handle := edge.S("sourceHandle") + if handle == "" { + handle = workflowBranchElse + } + if handle == selected { + return "active" + } + return "inactive" + case "failed": + return "failed" + case "inactive", "cancelled": + return "inactive" + default: + return "pending" + } +} + +// selectedBranch reads the branch a committed Condition selected from its committed state. A +// condition whose state carries no selection is read as else: the safe default routes downstream +// somewhere rather than dead-ending the branch, and a succeeded condition always committed one. +func (p *workflowProjection) selectedBranch(source workflowNodeView) string { + if state, ok := p.committed[source.ID]; ok { + if branch := state.Output.S("selectedBranch"); branch != "" { + return branch + } + } + return workflowBranchElse +} + +// incomingEdges returns the edges whose target is nodeID, in document order. +func (p *workflowProjection) incomingEdges(nodeID string) []Object { + out := []Object{} + for _, edge := range p.graph.Edges { + if edge.S("target") == nodeID { + out = append(out, edge) + } + } + return out +} + +// advanceWorkflowRun claims one pending run and advances it as far as committed state allows, in the +// caller's transaction. It is idempotent: a run that lost a claim race, was advanced concurrently, or +// is already terminal is left untouched. +func advanceWorkflowRun(t *transaction, runID string) { + run := t.one(` + SELECT r.*, s.graph FROM workflow_runs r + JOIN workflow_snapshots s ON s.id = r.snapshot_id + WHERE r.id=$1 AND r.deleted_at IS NULL`, runID) + if run == nil { + return + } + switch run.S("status") { + case "pending": + // The claim is the CAS: zero affected rows means another pass claimed it first. + if t.execRows(`UPDATE workflow_runs SET status='running', started_at=now(), version=version+1, updated_at=now() + WHERE id=$1 AND status='pending' AND deleted_at IS NULL`, runID) == 0 { + return + } + run = t.one(` + SELECT r.*, s.graph FROM workflow_runs r + JOIN workflow_snapshots s ON s.id = r.snapshot_id + WHERE r.id=$1`, runID) + case "running": + // Recompute from committed state; a crash between ticks resumes exactly here. + default: + return + } + version := run.N("version") + graph := readWorkflowGraph(run.O("graph")) + if graph.Start == "" { + // A snapshot predating the publish gate can lack a start node; there is no defined + // place to begin, so the run settles failed instead of hanging in running. + settleWorkflowRunStates(t, runID, version, "failed", "graph has no start node", nil) + return + } + states := workflowStatesFromJSON(run.O("nodeStates")) + for { + projection := projectWorkflowNodes(graph, states) + node, ready := projection.readyNode() + if !ready { + break + } + if node.Kind == "agent" { + // Phase-1 boundary: agent execution through the control plane is the next step of + // the same decision. Failing the run here is the honest terminal state — the node + // never ran, so it is never recorded as succeeded, and the run names the gap. + states[node.ID] = workflowNodeState{ + Status: "failed", + DisplayName: workflowNodeDisplayName(node), + StartedAt: nowRFC3339(), + FinishedAt: nowRFC3339(), + Output: Object{"reason": "agent_execution_not_wired"}, + } + settleWorkflowRunStates(t, runID, version, "failed", + "agent node "+node.ID+" is not wired to execution yet", states) + return + } + states[node.ID] = executeWorkflowNode(node, run.O("input"), states) + } + // The wave stopped with no executable node left. Persist the projection's inactive nodes — + // behind a branch that was not taken, they never ran and never will — so the run row alone + // tells the viewer the full story (a missing state renders as idle, which would understate a + // branch the run already resolved). Then settle: failed when any committed node failed, + // succeeded otherwise; the failure check reads committed state, never the pass's own reasoning. + states = workflowInactiveStates(graph, states) + if projectWorkflowNodes(graph, states).hasFailure() { + settleWorkflowRunStates(t, runID, version, "failed", workflowRunFailureReason(states), states) + return + } + settleWorkflowRunStates(t, runID, version, "succeeded", "", states) +} + +// workflowInactiveStates adds a committed inactive state for every node the projection resolved as +// inactive but that carries no committed state. Behind a branch that was not taken, such a node +// never ran and never will; persisting it keeps the run row alone telling the viewer the full story. +func workflowInactiveStates(graph workflowGraphView, states map[string]workflowNodeState) map[string]workflowNodeState { + p := projectWorkflowNodes(graph, states) + out := states + for _, node := range graph.Nodes { + if p.projected[node.ID] == "inactive" { + if _, committed := states[node.ID]; !committed { + out[node.ID] = workflowNodeState{Status: "inactive", DisplayName: workflowNodeDisplayName(node)} + } + } + } + return out +} + +// settleWorkflowRunStates persists the pass's results behind the run's version CAS — the +// compare-and-swap that makes a late or duplicate pass a no-op (invariant 3). Zero affected rows +// (a concurrent pass committed first) discards this pass's local view; the next tick recomputes +// from the committed row. +func settleWorkflowRunStates(t *transaction, runID string, version int64, status, runErr string, states map[string]workflowNodeState) { + if states == nil { + t.exec(`UPDATE workflow_runs SET status=$2, error=$3, finished_at=now(), version=version+1, updated_at=now() + WHERE id=$1 AND version=$4 AND deleted_at IS NULL`, runID, status, runErr, version) + return + } + encoded := Object{} + for id, state := range states { + encoded[id] = state.object() + } + t.exec(`UPDATE workflow_runs SET node_states=$2, status=$3, error=$4, finished_at=now(), version=version+1, updated_at=now() + WHERE id=$1 AND version=$5 AND deleted_at IS NULL`, runID, jsonText(encoded), status, runErr, version) +} + +// executeWorkflowNode commits one ready non-agent node's result from committed state. start reports +// what the run was kicked off with; condition evaluates its cases against prior node outputs and +// commits the selected branch (first match wins, else when nothing matched); output is a terminal +// marker whose produce stays in the node state for the run viewer (D7: 成果交付延后). +func executeWorkflowNode(node workflowNodeView, input Object, states map[string]workflowNodeState) workflowNodeState { + now := nowRFC3339() + state := workflowNodeState{ + Status: "succeeded", + DisplayName: workflowNodeDisplayName(node), + StartedAt: now, + FinishedAt: now, + } + switch node.Kind { + case "start": + state.Output = Object{"input": input} + case "condition": + state.Output = Object{"selectedBranch": evaluateWorkflowCondition(node, states)} + case "output": + state.Output = Object{} + default: + // Unreachable: the ready pass only yields executable kinds, and the agent case returned + // earlier. A future kind that reaches here must not be recorded as success. + state.Status = "failed" + state.Output = Object{"reason": "unsupported_node_kind"} + } + return state +} + +// 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 { + if title := node.Data.S("title"); title != "" { + return title + } + return node.Kind +} + +// workflowRunFailureReason names the first committed failure in node order, so the run's error +// points at the node that actually failed rather than at a downstream casualty. +func workflowRunFailureReason(states map[string]workflowNodeState) string { + for _, state := range states { + if state.Status != "failed" { + continue + } + if reason := state.Output.S("reason"); reason != "" { + return reason + } + return "node failed" + } + return "run failed" +} + +func nowRFC3339() string { + return time.Now().UTC().Format(time.RFC3339) +} diff --git a/internal/core/workflow_scheduler_db_test.go b/internal/core/workflow_scheduler_db_test.go new file mode 100644 index 000000000..901cc5d86 --- /dev/null +++ b/internal/core/workflow_scheduler_db_test.go @@ -0,0 +1,156 @@ +package core + +// DB tests for the Cloud-side workflow scheduler's advance pass (workflow_scheduler.go): a pending +// run is claimed and settled from the frozen snapshot's graph, agent nodes fail the run honestly +// until the execution wiring lands, and the pass is idempotent against already-advanced runs. Each +// test is a real isolated schema (dispatcherDB), and the graph reaches the pass the way production +// delivers it — jsonb round-tripped, so every document is plain map[string]any. + +import ( + "context" + "encoding/json" + "strings" + "testing" + + "github.com/google/uuid" +) + +// seedWorkflowRunScene loads one tenant with a workflow, a frozen snapshot and a pending run whose +// 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() + wid, sid, runID := uuid.NewString(), uuid.NewString(), uuid.NewString() + tid, uid := uuid.NewString(), uuid.NewString() + g, err := json.Marshal(graph) + if err != nil { + t.Fatalf("marshal graph: %v", err) + } + tx, err := store.Pool.Begin() + if err != nil { + t.Fatalf("seed begin: %v", err) + } + exec := func(name, q string, args ...any) { + if _, err := tx.Exec(q, args...); err != nil { + t.Fatalf("seed %s: %v", name, err) + } + } + exec("user", `INSERT INTO users(id,display_name,status) VALUES($1,'wf','active')`, uid) + exec("tenant", `INSERT INTO tenants(id,name,status) VALUES($1,'wf','active')`, tid) + 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) + if err := tx.Commit(); err != nil { + t.Fatalf("seed commit: %v", err) + } + return runID +} + +// readWorkflowRun reads the run row back at the storage boundary: the settled status and error, the +// committed node states and the version the CAS bumped. +func readWorkflowRun(t *testing.T, store *Store, runID string) (status, runErr string, states Object, version int64) { + t.Helper() + var raw []byte + if err := store.Pool.QueryRow(`SELECT status, error, node_states, version FROM workflow_runs WHERE id=$1`, runID). + Scan(&status, &runErr, &raw, &version); err != nil { + t.Fatalf("read run: %v", err) + } + if err := json.Unmarshal(raw, &states); err != nil { + t.Fatalf("decode node_states: %v", err) + } + return status, runErr, states, version +} + +// One pass claims the pending run, executes start then output from the frozen graph, and settles the +// run succeeded with every node's committed state on the row. +func TestWorkflowAdvanceSucceedsLinearRun(t *testing.T) { + store := dispatcherDB(t) + runID := seedWorkflowRunScene(t, store, wfGraph( + []Object{wfNode("s", "start", ""), wfNode("o", "output", "done")}, + [][3]string{{"s", "o", ""}}, + )) + + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance: %v", err) + } + status, runErr, states, version := readWorkflowRun(t, store, runID) + if status != "succeeded" || runErr != "" { + t.Fatalf("run settled %q error %q; want succeeded", status, runErr) + } + if version != 3 { // claim +1, settle +1 + t.Fatalf("version = %d; want 3", version) + } + for id, want := range map[string]string{"s": "succeeded", "o": "succeeded"} { + state, ok := asWorkflowObject(states[id]) + if !ok { + t.Fatalf("node %s has no committed state", id) + } + if state.S("status") != want { + 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 report the run input") + } +} + +// A ready agent node settles the run failed with an explicit reason: the node never ran, so it is +// never recorded as succeeded (invariant 5) and nothing downstream of it committed a state. +func TestWorkflowAdvanceFailsRunOnUnwiredAgentNode(t *testing.T) { + store := dispatcherDB(t) + runID := seedWorkflowRunScene(t, store, wfGraph( + []Object{wfNode("s", "start", ""), wfNode("a", "agent", "worker"), wfNode("o", "output", "")}, + [][3]string{{"s", "a", ""}, {"a", "o", ""}}, + )) + + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance: %v", err) + } + status, runErr, states, _ := readWorkflowRun(t, store, runID) + if status != "failed" { + t.Fatalf("run settled %q; want failed", status) + } + if !strings.Contains(runErr, "a") || !strings.Contains(runErr, "not wired") { + t.Fatalf("run error %q does not name the unwired agent node", runErr) + } + agent, ok := asWorkflowObject(states["a"]) + if !ok { + t.Fatal("agent node has no committed state") + } + if agent.S("status") != "failed" { + t.Fatalf("agent node status = %q; want failed", agent.S("status")) + } + if agent.O("output").S("reason") != "agent_execution_not_wired" { + t.Fatalf("agent node reason = %q; want agent_execution_not_wired", agent.O("output").S("reason")) + } + if _, ran := states["o"]; ran { + t.Fatal("output node downstream of the unwired agent committed a state") + } + if _, ran := states["s"]; !ran { + t.Fatal("start node before the agent did not commit its state") + } +} + +// The pass is idempotent: a second tick over a settled run rescans nothing and leaves the committed +// row untouched — no second claim, no second settle, no version drift. +func TestWorkflowAdvanceIsIdempotent(t *testing.T) { + store := dispatcherDB(t) + runID := seedWorkflowRunScene(t, store, wfGraph( + []Object{wfNode("s", "start", ""), wfNode("o", "output", "")}, + [][3]string{{"s", "o", ""}}, + )) + + for i := 0; i < 3; i++ { + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance %d: %v", i, err) + } + } + status, runErr, _, version := readWorkflowRun(t, store, runID) + if status != "succeeded" || runErr != "" { + t.Fatalf("run settled %q error %q; want succeeded", status, runErr) + } + if version != 3 { + t.Fatalf("version = %d after three passes; want 3 (claim + settle only)", version) + } +} diff --git a/internal/core/workflow_scheduler_test.go b/internal/core/workflow_scheduler_test.go new file mode 100644 index 000000000..3c4321718 --- /dev/null +++ b/internal/core/workflow_scheduler_test.go @@ -0,0 +1,279 @@ +package core + +import "testing" + +// The workflow scheduler's pure core: the projection (ready-set recompute from committed state) and +// the branch semantics. These test the parts that decide WHAT happens next; the DB tests +// (workflow_scheduler_db_test.go) cover that the pass commits them. + +// wfNode builds one node document: id, kind, title and the data fields tests need. +func wfNode(id, kind, title string) Object { + data := Object{"kind": kind} + if title != "" { + data["title"] = title + } + return Object{"id": id, "data": data} +} + +// wfGraph builds a graph document from nodes and source→target(→handle) edge triples. +func wfGraph(nodes []Object, edges [][3]string) Object { + nodeList := make([]any, 0, len(nodes)) + for _, node := range nodes { + nodeList = append(nodeList, map[string]any(node)) + } + edgeList := make([]any, 0, len(edges)) + for _, e := range edges { + edge := Object{"source": e[0], "target": e[1]} + if e[2] != "" { + edge["sourceHandle"] = e[2] + } + edgeList = append(edgeList, map[string]any(edge)) + } + return Object{"nodes": nodeList, "edges": edgeList} +} + +// A start→output graph runs to drained in two waves: start readies, then output, then nothing. +func TestWorkflowProjectionLinearGraph(t *testing.T) { + graph := readWorkflowGraph(wfGraph( + []Object{wfNode("s", "start", ""), wfNode("o", "output", "")}, + [][3]string{{"s", "o", ""}}, + )) + states := map[string]workflowNodeState{} + + p := projectWorkflowNodes(graph, states) + node, ok := p.readyNode() + if !ok || node.ID != "s" { + t.Fatalf("first ready node = %q, %v; want start s", node.ID, ok) + } + states["s"] = executeWorkflowNode(node, Object{"flag": true}, states) + + p = projectWorkflowNodes(graph, states) + node, ok = p.readyNode() + if !ok || node.ID != "o" { + t.Fatalf("second ready node = %q, %v; want output o", node.ID, ok) + } + states["o"] = executeWorkflowNode(node, nil, states) + + p = projectWorkflowNodes(graph, states) + if _, ok := p.readyNode(); ok { + t.Fatal("drained linear graph still reports a ready node") + } + if p.hasFailure() { + t.Fatal("drained linear graph reported a failure") + } +} + +// wfConditionNode is a condition node whose cases the test authors directly. +func wfConditionNode(id string, cases any) Object { + node := wfNode(id, "condition", "") + node["data"].(Object)["cases"] = cases + return node +} + +// A condition commits its branch: the matched branch's edge goes active, every other outgoing edge +// goes inactive, and downstream nodes resolve accordingly. +func TestWorkflowProjectionBranchSelection(t *testing.T) { + yes := Object{"id": "yes", "logic": "and", "conditions": []any{ + Object{"variableSelector": []any{"s", "input", "flag"}, "operator": "equals", "value": true}, + }} + graph := readWorkflowGraph(wfGraph( + []Object{wfNode("s", "start", ""), wfConditionNode("c", []any{yes}), wfNode("y", "output", ""), wfNode("e", "output", "")}, + [][3]string{{"s", "c", ""}, {"c", "y", "yes"}, {"c", "e", "else"}}, + )) + states := map[string]workflowNodeState{} + states["s"] = executeWorkflowNode(graph.ByNode["s"], Object{"flag": true}, states) + + branch := evaluateWorkflowCondition(graph.ByNode["c"], states) + if branch != "yes" { + t.Fatalf("condition selected %q; want yes", branch) + } + states["c"] = workflowNodeState{Status: "succeeded", Output: Object{"selectedBranch": branch}} + + p := projectWorkflowNodes(graph, states) + if node, ok := p.readyNode(); !ok || node.ID != "y" { + t.Fatalf("ready node = %q, %v; want taken branch y", node.ID, ok) + } + if p.projected["e"] != "inactive" { + t.Fatalf("untaken branch e = %q; want inactive", p.projected["e"]) + } +} + +// An edge with no sourceHandle is the else edge: when no case matches, it is the one that goes +// active (desktop branch_projection: an unlabeled edge defaults to else). +func TestWorkflowProjectionUnlabeledEdgeIsElse(t *testing.T) { + miss := Object{"id": "yes", "logic": "and", "conditions": []any{ + Object{"variableSelector": []any{"s", "input", "flag"}, "operator": "equals", "value": false}, + }} + graph := readWorkflowGraph(wfGraph( + []Object{wfNode("s", "start", ""), wfConditionNode("c", []any{miss}), wfNode("y", "output", ""), wfNode("e", "output", "")}, + [][3]string{{"s", "c", ""}, {"c", "y", "yes"}, {"c", "e", ""}}, + )) + states := map[string]workflowNodeState{} + states["s"] = executeWorkflowNode(graph.ByNode["s"], Object{"flag": true}, states) + + if branch := evaluateWorkflowCondition(graph.ByNode["c"], states); branch != workflowBranchElse { + t.Fatalf("all-miss condition selected %q; want else", branch) + } + states["c"] = workflowNodeState{Status: "succeeded", Output: Object{"selectedBranch": workflowBranchElse}} + + p := projectWorkflowNodes(graph, states) + if node, ok := p.readyNode(); !ok || node.ID != "e" { + t.Fatalf("ready node = %q, %v; want unlabeled else edge e", node.ID, ok) + } + if p.projected["y"] != "inactive" { + t.Fatalf("labeled branch y = %q; want inactive", p.projected["y"]) + } +} + +// A node with no incoming edges never runs unless it is the start node (desktop branch_projection: +// rootless non-start nodes resolve inactive, so an authoring leftover cannot hijack execution). +func TestWorkflowProjectionRootlessNodeIsInactive(t *testing.T) { + graph := readWorkflowGraph(wfGraph( + []Object{wfNode("s", "start", ""), wfNode("o", "output", ""), wfNode("x", "output", "orphan")}, + [][3]string{{"s", "o", ""}}, + )) + p := projectWorkflowNodes(graph, map[string]workflowNodeState{}) + if node, ok := p.readyNode(); !ok || node.ID != "s" { + t.Fatalf("ready node = %q, %v; want start s", node.ID, ok) + } + if p.projected["x"] != "inactive" { + t.Fatalf("rootless non-start x = %q; want inactive", p.projected["x"]) + } +} + +// An upstream node without a committed result leaves downstream unresolved — not ready, not failed, +// not inactive (invariant 5: nothing resolves on missing state; the pass waits for the commit). +func TestWorkflowProjectionPendingUpstreamWaits(t *testing.T) { + graph := readWorkflowGraph(wfGraph( + []Object{wfNode("s", "start", ""), wfNode("o", "output", "")}, + [][3]string{{"s", "o", ""}}, + )) + states := map[string]workflowNodeState{"s": {Status: "running", DisplayName: "start"}} + p := projectWorkflowNodes(graph, states) + if node, ok := p.readyNode(); ok { + t.Fatalf("downstream of a running node readied %q", node.ID) + } + if resolved, ok := p.projected["o"]; ok { + t.Fatalf("downstream of a running node resolved as %q", resolved) + } +} + +// A failed node fails its downstream edges: the branch resolves failed (not ready, not inactive), +// and hasFailure — read from committed state — is true, which is what settles the run failed. +func TestWorkflowProjectionFailurePropagates(t *testing.T) { + graph := readWorkflowGraph(wfGraph( + []Object{wfNode("s", "start", ""), wfNode("o", "output", "")}, + [][3]string{{"s", "o", ""}}, + )) + states := map[string]workflowNodeState{ + "s": {Status: "failed", DisplayName: "start", Output: Object{"reason": "boom"}}, + } + p := projectWorkflowNodes(graph, states) + if p.projected["o"] != "failed" { + t.Fatalf("downstream of a failed node = %q; want failed", p.projected["o"]) + } + if !p.hasFailure() { + t.Fatal("committed failure not reported") + } +} + +// wfRuleEvaluate runs one condition rule against a start node whose output the test authors +// directly, returning whether the rule matches. +func wfRuleEvaluate(t *testing.T, selector []any, operator string, value any, output Object) bool { + t.Helper() + states := map[string]workflowNodeState{"s": {Status: "succeeded", Output: output}} + return workflowConditionRule(Object{"variableSelector": selector, "operator": operator, "value": value}, states) +} + +// The operator table, evaluated against a start node whose output is +// {"flag": true, "count": 2, "text": "hello world", "list": [], "nothing": null}. +func TestWorkflowConditionOperators(t *testing.T) { + output := Object{"flag": true, "count": 2.0, "text": "hello world", "list": []any{}, "nothing": nil} + cases := []struct { + name string + selector []any + operator string + value any + wantMatch bool + }{ + {"equals match", []any{"s", "flag"}, "equals", true, true}, + {"is is equals", []any{"s", "flag"}, "is", true, true}, + {"equals mismatch", []any{"s", "flag"}, "equals", false, false}, + {"not_equals mismatch", []any{"s", "flag"}, "not_equals", true, false}, + {"not_equals match", []any{"s", "flag"}, "not_equals", false, true}, + {"contains match", []any{"s", "text"}, "contains", "lo wo", true}, + {"not_contains match", []any{"s", "text"}, "not_contains", "zz", true}, + {"starts_with match", []any{"s", "text"}, "starts_with", "hello", true}, + {"ends_with match", []any{"s", "text"}, "ends_with", "world", true}, + {"empty on empty list", []any{"s", "list"}, "empty", nil, true}, + {"empty on null", []any{"s", "nothing"}, "empty", nil, true}, + {"not_empty on text", []any{"s", "text"}, "not_empty", nil, true}, + {"exists on present", []any{"s", "flag"}, "exists", nil, true}, + {"not_exists on absent", []any{"s", "missing"}, "not_exists", nil, true}, + {"exists on absent", []any{"s", "missing"}, "exists", nil, false}, + {"empty on absent", []any{"s", "missing"}, "empty", nil, true}, + {"not_empty on absent", []any{"s", "missing"}, "not_empty", nil, false}, + {"comparator on absent", []any{"s", "missing"}, "equals", nil, false}, + {"greater_than match", []any{"s", "count"}, "greater_than", 1.0, true}, + {"greater_than no match", []any{"s", "count"}, "greater_than", 2.0, false}, + {"greater_than_or_equal boundary", []any{"s", "count"}, "greater_than_or_equal", 2.0, true}, + {"less_than match", []any{"s", "count"}, "less_than", 3.0, true}, + {"less_than_or_equal boundary", []any{"s", "count"}, "less_than_or_equal", 2.0, true}, + {"numeric against text", []any{"s", "text"}, "greater_than", 1.0, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := wfRuleEvaluate(t, tc.selector, tc.operator, tc.value, output) + if got != tc.wantMatch { + t.Fatalf("rule %q %v got %v; want %v", tc.operator, tc.value, got, tc.wantMatch) + } + }) + } +} + +// Case logic: "and" requires every rule, "or" any rule, the first matching case wins; a case with +// no rules is an authored shell and falls through; a malformed rule, an unknown operator and a +// selector naming a node that never produced output are all no-matches. +func TestWorkflowConditionCaseLogic(t *testing.T) { + states := map[string]workflowNodeState{"s": {Status: "succeeded", Output: Object{"flag": true, "count": 2.0}}} + pick := func(cases any) string { + return evaluateWorkflowCondition(workflowNodeView{ID: "c", Kind: "condition", Data: Object{"cases": cases}}, states) + } + rule := func(key string, value any) Object { + return Object{"variableSelector": []any{"s", key}, "operator": "equals", "value": value} + } + and := []any{Object{"id": "both", "logic": "and", "conditions": []any{rule("flag", true), rule("count", 3.0)}}} + if got := pick(and); got != workflowBranchElse { + t.Fatalf("and-case with one failed rule selected %q; want else", got) + } + or := []any{Object{"id": "either", "logic": "or", "conditions": []any{rule("flag", false), rule("count", 2.0)}}} + if got := pick(or); got != "either" { + t.Fatalf("or-case with one matching rule selected %q; want either", got) + } + first := []any{ + Object{"id": "first", "logic": "or", "conditions": []any{rule("flag", true)}}, + Object{"id": "second", "logic": "or", "conditions": []any{rule("count", 2.0)}}, + } + if got := pick(first); got != "first" { + t.Fatalf("first matching case selected %q; want first", got) + } + if got := pick([]any{Object{"id": "shell", "logic": "and"}}); got != workflowBranchElse { + t.Fatalf("case with no rules selected %q; want else", got) + } + malformed := []any{Object{"id": "bad", "logic": "or", "conditions": []any{Object{"operator": "equals"}}}} + if got := pick(malformed); got != workflowBranchElse { + t.Fatalf("malformed rule selected %q; want else", got) + } + badOp := []any{Object{"id": "unk", "logic": "or", "conditions": []any{ + Object{"variableSelector": []any{"s", "flag"}, "operator": "frobnicate", "value": true}, + }}} + if got := pick(badOp); got != workflowBranchElse { + t.Fatalf("unknown operator selected %q; want else", got) + } + ghost := []any{Object{"id": "ghost", "logic": "or", "conditions": []any{ + Object{"variableSelector": []any{"nope", "flag"}, "operator": "exists", "value": nil}, + }}} + if got := pick(ghost); got != workflowBranchElse { + t.Fatalf("rule on a node that never ran selected %q; want else", got) + } +} From 0d9363d1d207f345e384375b4bbf96059ce960ff Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 14:31:54 +0800 Subject: [PATCH 2/6] fix(core): name a run's failure reason in node order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `workflowRunFailureReason` read the reason out of the committed state map. Go randomizes map iteration, so a run with more than one failed node recorded whichever failure the iteration happened to reach first: the same trace could record a different reason on every replay, which makes the field useless as evidence of anything. Walk the graph's nodes instead. Every committed state belongs to a graph node — the scheduler writes them from graph nodes and the graph is frozen at creation — so node order reaches them all and is stable across replays. The regression test asserts the first-in-order reason across 200 calls: against the old implementation it failed within the first 15 iterations, so the determinism is pinned rather than assumed. Co-Authored-By: Claude Code --- internal/core/workflow_scheduler.go | 13 +++++--- internal/core/workflow_scheduler_test.go | 40 ++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/internal/core/workflow_scheduler.go b/internal/core/workflow_scheduler.go index 56d53518b..900e2f6fa 100644 --- a/internal/core/workflow_scheduler.go +++ b/internal/core/workflow_scheduler.go @@ -426,7 +426,7 @@ func advanceWorkflowRun(t *transaction, runID string) { // succeeded otherwise; the failure check reads committed state, never the pass's own reasoning. states = workflowInactiveStates(graph, states) if projectWorkflowNodes(graph, states).hasFailure() { - settleWorkflowRunStates(t, runID, version, "failed", workflowRunFailureReason(states), states) + settleWorkflowRunStates(t, runID, version, "failed", workflowRunFailureReason(graph, states), states) return } settleWorkflowRunStates(t, runID, version, "succeeded", "", states) @@ -504,9 +504,14 @@ func workflowNodeDisplayName(node workflowNodeView) string { } // workflowRunFailureReason names the first committed failure in node order, so the run's error -// points at the node that actually failed rather than at a downstream casualty. -func workflowRunFailureReason(states map[string]workflowNodeState) string { - for _, state := range states { +// points at the node that actually failed rather than at a downstream casualty. The order has to +// come from the graph: the states are a map, Go randomizes map iteration, and reading the reason +// out of it would let the same trace record a different reason on every replay. Every committed +// state belongs to a graph node — the scheduler writes them from graph nodes, and the graph is +// frozen at creation — so walking the graph reaches them all. +func workflowRunFailureReason(graph workflowGraphView, states map[string]workflowNodeState) string { + for _, node := range graph.Nodes { + state := states[node.ID] if state.Status != "failed" { continue } diff --git a/internal/core/workflow_scheduler_test.go b/internal/core/workflow_scheduler_test.go index 3c4321718..76a07586c 100644 --- a/internal/core/workflow_scheduler_test.go +++ b/internal/core/workflow_scheduler_test.go @@ -177,6 +177,46 @@ func TestWorkflowProjectionFailurePropagates(t *testing.T) { } } +// A run's recorded failure reason must come from the first failure in node order, not from +// whichever failure the map happened to yield first: the states are a map, Go randomizes map +// iteration, and a reason that changes between two replays of the same trace is not evidence of +// anything. One iteration would pass half the time under the old implementation, so the loop turns +// that coin flip into a near-certain failure. +func TestWorkflowRunFailureReasonFollowsNodeOrder(t *testing.T) { + graph := readWorkflowGraph(wfGraph( + []Object{wfNode("first", "agent", ""), wfNode("second", "agent", "")}, + nil, + )) + states := map[string]workflowNodeState{ + "first": {Status: "failed", Output: Object{"reason": "first failed"}}, + "second": {Status: "failed", Output: Object{"reason": "second failed"}}, + } + for i := 0; i < 200; i++ { + if reason := workflowRunFailureReason(graph, states); reason != "first failed" { + t.Fatalf("iteration %d: reason = %q; want the first failure in node order", i, reason) + } + } +} + +// A failed node that carries no reason of its own still has to say something, and a run whose only +// committed failures are non-failing states reports the generic phrase rather than an empty one. +func TestWorkflowRunFailureReasonFallbacks(t *testing.T) { + graph := readWorkflowGraph(wfGraph( + []Object{wfNode("a", "agent", ""), wfNode("b", "agent", "")}, + nil, + )) + if reason := workflowRunFailureReason(graph, map[string]workflowNodeState{ + "a": {Status: "failed"}, + }); reason != "node failed" { + t.Fatalf("reasonless failure = %q; want node failed", reason) + } + if reason := workflowRunFailureReason(graph, map[string]workflowNodeState{ + "a": {Status: "succeeded", Output: Object{"reason": "not a failure"}}, + }); reason != "run failed" { + t.Fatalf("no committed failure = %q; want run failed", reason) + } +} + // wfRuleEvaluate runs one condition rule against a start node whose output the test authors // directly, returning whether the rule matches. func wfRuleEvaluate(t *testing.T, selector []any, operator string, value any, output Object) bool { From a0af7aa0acf86b1d41015d0706cefda75c5e3a0f Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 16:49:49 +0800 Subject: [PATCH 3/6] fix(core): read an empty object as empty and keep not_contains off absence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two Condition-node bugs found in review, both reachable from a legal `start -> condition` graph: - `empty` / `not_empty` type-switched on `map[string]any`, which a named `Object` never matches. A start node's output nests the run input as an `Object`, so a selector walking `[startId, "input"]` resolved to an `Object` and fell through to the default arm — `empty` answered false on `{}` and the run took the branch desktop's is_empty would not have taken, with nothing in the run to say so. The object arm now goes through `asWorkflowObject`, the helper the rest of the document reading already uses for exactly this trap. - `not_contains` was `negate(contains)`, so it answered true on an absent variable while every other comparator in the file — `not_equals` included — answers false. The file's own rule is that a comparator on an absent variable does not match; the negation belongs to the comparison, not to the classification of absence. `not_contains` gets its own operator that keeps absence a no-match. Both are pinned end to end by new DB tests driving a real start -> condition -> {case, else} graph through the advance pass: an empty run input must select the empty case, `not_empty` on the same input must not, and `not_contains` must not select on a path no node produced. Co-Authored-By: Claude Code --- internal/core/workflow_condition.go | 27 +++++- internal/core/workflow_scheduler_db_test.go | 98 +++++++++++++++++++++ internal/core/workflow_scheduler_test.go | 24 ++++- 3 files changed, 143 insertions(+), 6 deletions(-) diff --git a/internal/core/workflow_condition.go b/internal/core/workflow_condition.go index ef0d0c0b0..087a1c9dd 100644 --- a/internal/core/workflow_condition.go +++ b/internal/core/workflow_condition.go @@ -27,7 +27,7 @@ var workflowConditionOperators = map[string]workflowConditionOperator{ "not_equals": workflowCompareNotEquals, "is_not": workflowCompareNotEquals, "contains": workflowCompareContains, - "not_contains": workflowCompareNegate(workflowCompareContains), + "not_contains": workflowCompareNotContains, "starts_with": workflowCompareTextOrder("starts"), "ends_with": workflowCompareTextOrder("ends"), "empty": workflowCompareEmpty, @@ -61,6 +61,19 @@ func workflowCompareContains(resolved any, absent bool, expected any) bool { return strings.Contains(workflowTextOf(resolved), workflowTextOf(expected)) } +// workflowCompareNotContains is contains' negation with one deliberate exception: absence. Every +// comparator in this file treats an absent variable as a no-match (see the file comment), so +// not_contains has to answer false there too. Negating the inner verdict instead would answer +// true, which puts not_contains and not_equals on opposite branches of the same absent variable — +// not_contains would select its case while not_equals fell through to else. The negation belongs to +// the comparison, not to the classification of absence. +func workflowCompareNotContains(resolved any, absent bool, expected any) bool { + if absent { + return false + } + return !workflowCompareContains(resolved, absent, expected) +} + // workflowCompareTextOrder serves both text-position operators: "starts" is prefix, "ends" is // suffix; any other direction is unreachable. func workflowCompareTextOrder(direction string) workflowConditionOperator { @@ -89,10 +102,16 @@ func workflowCompareEmpty(resolved any, absent bool, expected any) bool { return value == "" case []any: return len(value) == 0 - case map[string]any: - return len(value) == 0 default: - return false + // The object case goes through asWorkflowObject rather than a `case map[string]any` arm: + // Object is a named type, so an Object boxed in `any` — which is exactly what a start node's + // output is, since it wraps the run input — never matches that arm and would fall through to + // "not empty". Reading a nested object out of a node output hits the same trap one level in. + obj, ok := asWorkflowObject(value) + if !ok { + return false + } + return len(obj) == 0 } } diff --git a/internal/core/workflow_scheduler_db_test.go b/internal/core/workflow_scheduler_db_test.go index 901cc5d86..08789dd64 100644 --- a/internal/core/workflow_scheduler_db_test.go +++ b/internal/core/workflow_scheduler_db_test.go @@ -154,3 +154,101 @@ func TestWorkflowAdvanceIsIdempotent(t *testing.T) { t.Fatalf("version = %d after three passes; want 3 (claim + settle only)", version) } } + +// wfOneRuleNode is a Condition node with exactly one case, matching when `operator` holds for the +// variable at `selector`. It exists so a DB test can name the operator under test without spelling +// out the whole case document. +func wfOneRuleNode(id, caseID string, selector []any, operator string) Object { + return wfConditionNode(id, []any{ + Object{"id": caseID, "logic": "and", "conditions": []any{ + Object{"variableSelector": selector, "operator": operator, "value": nil}, + }}, + }) +} + +// 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. +func wfConditionScene(t *testing.T, store *Store, caseID string, selector []any, operator string) string { + t.Helper() + return seedWorkflowRunScene(t, store, wfGraph( + []Object{ + wfNode("s", "start", ""), + wfOneRuleNode("c", caseID, selector, operator), + wfNode("e", "output", ""), + wfNode("f", "output", ""), + }, + [][3]string{{"s", "c", ""}, {"c", "e", caseID}, {"c", "f", workflowBranchElse}}, + )) +} + +// wfSelectedBranch advances one pass and returns the branch the condition node committed. +func wfSelectedBranch(t *testing.T, store *Store, runID string) string { + t.Helper() + if err := store.AdvanceWorkflowRunsOnce(context.Background()); err != nil { + t.Fatalf("advance: %v", err) + } + _, _, states, _ := readWorkflowRun(t, store, runID) + cond, ok := asWorkflowObject(states["c"]) + if !ok { + t.Fatal("condition node has no committed state") + } + return cond.O("output").S("selectedBranch") +} + +// wfNodeStatus reads one committed node's status, failing the test if the node has no state at all. +func wfNodeStatus(t *testing.T, states Object, nodeID string) string { + t.Helper() + state, ok := asWorkflowObject(states[nodeID]) + if !ok { + t.Fatalf("node %s has no committed state", nodeID) + } + 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) { + 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) + } + _, _, states, _ := readWorkflowRun(t, store, runID) + if got := wfNodeStatus(t, states, "e"); got != "succeeded" { + t.Fatalf("taken branch status = %q; want succeeded", got) + } + // 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) + } +} + +// 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) { + 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) + } +} + +// not_contains must not match an absent variable, and the run has to prove it end to end: the +// selector names a path the start node never produced, and the case is written with not_contains, the +// one negated comparator whose absence handling is not simply the inner verdict flipped. +func TestWorkflowAdvanceConditionNotContainsOnAbsentVariable(t *testing.T) { + store := dispatcherDB(t) + runID := wfConditionScene(t, store, "absent", []any{"s", "never_written"}, "not_contains") + + if branch := wfSelectedBranch(t, store, runID); branch != workflowBranchElse { + t.Fatalf("condition selected %q; want else — not_contains matched an absent variable", branch) + } +} diff --git a/internal/core/workflow_scheduler_test.go b/internal/core/workflow_scheduler_test.go index 76a07586c..4f6a40f89 100644 --- a/internal/core/workflow_scheduler_test.go +++ b/internal/core/workflow_scheduler_test.go @@ -226,9 +226,15 @@ func wfRuleEvaluate(t *testing.T, selector []any, operator string, value any, ou } // The operator table, evaluated against a start node whose output is -// {"flag": true, "count": 2, "text": "hello world", "list": [], "nothing": null}. +// {"flag": true, "count": 2, "text": "hello world", "list": [], "nothing": null, "emptyObj": {}, +// "fullObj": {"a": 1}, "plainObj": {}}. The three object keys are deliberately three different Go +// types for the same JSON: a start node's output nests the run input as the package's named Object, +// while a jsonb-decoded row nests a plain map. Both must classify as empty the same way. func TestWorkflowConditionOperators(t *testing.T) { - output := Object{"flag": true, "count": 2.0, "text": "hello world", "list": []any{}, "nothing": nil} + output := Object{ + "flag": true, "count": 2.0, "text": "hello world", "list": []any{}, "nothing": nil, + "emptyObj": Object{}, "fullObj": Object{"a": 1.0}, "plainObj": map[string]any{}, + } cases := []struct { name string selector []any @@ -248,12 +254,26 @@ func TestWorkflowConditionOperators(t *testing.T) { {"empty on empty list", []any{"s", "list"}, "empty", nil, true}, {"empty on null", []any{"s", "nothing"}, "empty", nil, true}, {"not_empty on text", []any{"s", "text"}, "not_empty", nil, true}, + // An empty object is empty whichever Go type carries it. Before asWorkflowObject was used + // here, the named Object fell through the type switch and an empty run input read as + // not-empty, so a `start -> condition` graph on input {} took the opposite branch from + // desktop's is_empty. + {"empty on empty Object", []any{"s", "emptyObj"}, "empty", nil, true}, + {"not_empty on empty Object", []any{"s", "emptyObj"}, "not_empty", nil, false}, + {"not_empty on full Object", []any{"s", "fullObj"}, "not_empty", nil, true}, + {"empty on full Object", []any{"s", "fullObj"}, "empty", nil, false}, + {"empty on empty plain map", []any{"s", "plainObj"}, "empty", nil, true}, {"exists on present", []any{"s", "flag"}, "exists", nil, true}, {"not_exists on absent", []any{"s", "missing"}, "not_exists", nil, true}, {"exists on absent", []any{"s", "missing"}, "exists", nil, false}, {"empty on absent", []any{"s", "missing"}, "empty", nil, true}, {"not_empty on absent", []any{"s", "missing"}, "not_empty", nil, false}, {"comparator on absent", []any{"s", "missing"}, "equals", nil, false}, + // Absence is a no-match for every comparator, negated ones included: not_contains must not + // select its branch on a variable no prior node ever produced, or it disagrees with + // not_equals about the same absent variable. + {"not_contains on absent", []any{"s", "missing"}, "not_contains", "zz", false}, + {"not_contains mismatch", []any{"s", "text"}, "not_contains", "hello", false}, {"greater_than match", []any{"s", "count"}, "greater_than", 1.0, true}, {"greater_than no match", []any{"s", "count"}, "greater_than", 2.0, false}, {"greater_than_or_equal boundary", []any{"s", "count"}, "greater_than_or_equal", 2.0, true}, From eac61097c5b31de3c9a5dce9151b928db1820c73 Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 17:23:31 +0800 Subject: [PATCH 4/6] test(core): assert the unwired-node run error instead of a substring of it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round on #67, non-blocking item. `strings.Contains(runErr, "a")` cannot fail: the node it claims to look for is named "a", so every non-empty string satisfies it, and the check that actually did work was the one for "not wired". The test therefore asserted less than its message said. The unwired-agent path is also the one failure that writes a run error of its own — `agent node is not wired to execution yet` — rather than reusing the node's reason the way workflowRunFailureReason does everywhere else, so naming the node is a property worth pinning. It is pinned as the whole sentence now. Co-Authored-By: Claude Code --- internal/core/workflow_scheduler_db_test.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/internal/core/workflow_scheduler_db_test.go b/internal/core/workflow_scheduler_db_test.go index 08789dd64..7defa254f 100644 --- a/internal/core/workflow_scheduler_db_test.go +++ b/internal/core/workflow_scheduler_db_test.go @@ -9,7 +9,6 @@ package core import ( "context" "encoding/json" - "strings" "testing" "github.com/google/uuid" @@ -111,8 +110,12 @@ func TestWorkflowAdvanceFailsRunOnUnwiredAgentNode(t *testing.T) { if status != "failed" { t.Fatalf("run settled %q; want failed", status) } - if !strings.Contains(runErr, "a") || !strings.Contains(runErr, "not wired") { - t.Fatalf("run error %q does not name the unwired agent node", runErr) + // The unwired-agent path is the one failure that writes a run error of its own rather than + // reusing the node's reason, and the point of it is that the run names the node. The assertion + // is the whole sentence: a substring check on the id would assert nothing here, because the node + // is named "a" and every non-empty string contains it. + if runErr != "agent node a is not wired to execution yet" { + t.Fatalf("run error %q; want it to name the unwired agent node", runErr) } agent, ok := asWorkflowObject(states["a"]) if !ok { From 7abd979c37eb040b470af248e204d2794223bcc8 Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 17:44:53 +0800 Subject: [PATCH 5/6] fix(core): refuse condition configs the first phase cannot evaluate like desktop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The publish gate validated condition nodes only for the fields every node shares, so a document naming an operator Cloud does not implement, a logic neither side defines, a case with no id, a one-segment selector or an expected value carrying `{{#...#}}` published cleanly and then answered something at run time that nothing in the document asked for — a no-match, an `and`, the else branch, the whole node output, a literal comparison. desktop's condition.rs answers differently in each of those places, so the graph was accepted under one semantics and evaluated under another. This is the gate half of circus#202's owner decision A condition 2: the list of configurations the first phase does not support now lives in workflowUnsupportedConditionConfigs, publishing refuses each with a problem that names the node and the field, and the test checks the list against the cases in both directions so the ADR's list, the PR's list and the gate cannot drift apart. An unfinished node is deliberately not refused: no cases, or a case with no rules, falls through to else on both sides already. The differences that only show up on a *resolved* value — `contains` on an array or object, a non-string `starts_with`/`ends_with`, an expected-value template that resolves — cannot be decided from the document and stay with the parity work in circus#210. Co-Authored-By: Claude Code --- internal/core/workflow_graph.go | 109 +++++++++++++++++++++ internal/core/workflow_graph_test.go | 140 +++++++++++++++++++++++++++ 2 files changed, 249 insertions(+) diff --git a/internal/core/workflow_graph.go b/internal/core/workflow_graph.go index 1f798d0e7..67c798ec9 100644 --- a/internal/core/workflow_graph.go +++ b/internal/core/workflow_graph.go @@ -173,6 +173,8 @@ func validateWorkflowNodes(nodes []any) []workflowGraphProblem { // and reporting its config too would ask the author to fix a field of a node they have to // delete anyway. problems = append(problems, validateAgentConfig(id, o.O("data"))...) + case kind == "condition": + problems = append(problems, validateConditionConfig(id, o.O("data"))...) } } if starts != 1 { @@ -203,6 +205,113 @@ func validateAgentConfig(nodeID string, data Object) []workflowGraphProblem { return problems } +// workflowUnsupportedConditionConfigs is the list of condition-node configuration Cloud cannot +// evaluate the way desktop's condition.rs does, and which publishing therefore refuses. It is +// written out rather than derived because it is a statement about the two implementations, not about +// Cloud's own code: each entry is a place where the evaluator would otherwise answer silently — +// an unknown operator is a no-match, an unknown logic falls back to `and`, a missing case id becomes +// the else branch, a one-segment selector becomes the whole node output, and an expected value +// carrying `{{#selector#}}` is compared as literal text where desktop resolves it and keeps the +// bound type (circus#202 A 条件 2, #141 结论 5). Every one of them is decidable from the document +// alone, which is what makes refusing at publish possible; the differences that depend on a +// *resolved* value's type at run time are not, and are named in the ADR instead. +// +// A condition node with no cases, or a case with no rules, is deliberately NOT on this list: falling +// through to else is what an unfinished node is meant to do, and both sides already agree on it. +// +// When the parity work lands (circus#210, split out by circus#202 A 条件 3) these refusals come off +// one at a time, each with the cross-check case that proves the two sides now answer the same thing. +var workflowUnsupportedConditionConfigs = []string{ + "condition_cases_not_a_list", + "condition_case_not_an_object", + "condition_case_id_missing", + "condition_logic_unsupported", + "condition_rules_not_a_list", + "condition_rule_not_an_object", + "condition_selector_unsupported", + "condition_operator_unsupported", + "condition_value_template_unsupported", +} + +// validateConditionConfig refuses the condition configuration the first phase does not evaluate the +// way desktop does. Every problem names the node and the field, so the author is told which branch +// of which node to fix rather than that the graph is invalid somewhere. +func validateConditionConfig(nodeID string, data Object) []workflowGraphProblem { + problems := []workflowGraphProblem{} + raw, present := data["cases"] + if !present || raw == nil { + return problems + } + cases, ok := raw.([]any) + if !ok { + return append(problems, workflowGraphProblem{Code: "condition_cases_not_a_list", NodeID: nodeID, Field: "cases"}) + } + for _, item := range cases { + c, ok := asWorkflowObject(item) + if !ok { + problems = append(problems, workflowGraphProblem{Code: "condition_case_not_an_object", NodeID: nodeID, Field: "cases"}) + continue + } + if c.S("id") == "" { + problems = append(problems, workflowGraphProblem{Code: "condition_case_id_missing", NodeID: nodeID, Field: "cases.id"}) + } + // An absent logic is `and`, which is what the evaluator applies and what the editor omits; + // only a value neither side recognises is refused. + switch c.S("logic") { + case "", "and", "or": + default: + problems = append(problems, workflowGraphProblem{Code: "condition_logic_unsupported", NodeID: nodeID, Field: "cases.logic"}) + } + rules, ok := c["conditions"].([]any) + if !ok { + // An absent or null rule list is the same unfinished case as an empty one, and is left + // to fall through; a list of some other type is a document this gate cannot read. + if value, present := c["conditions"]; present && value != nil { + problems = append(problems, workflowGraphProblem{Code: "condition_rules_not_a_list", NodeID: nodeID, Field: "cases.conditions"}) + } + continue + } + for _, rawRule := range rules { + problems = append(problems, validateConditionRule(nodeID, rawRule)...) + } + } + return problems +} + +// validateConditionRule checks one rule of one case. The operator set is read from the evaluator's +// own table rather than restated, so the gate cannot admit an operator the evaluator would then +// no-match — the two lists are the same list. +func validateConditionRule(nodeID string, raw any) []workflowGraphProblem { + problems := []workflowGraphProblem{} + r, ok := asWorkflowObject(raw) + if !ok { + return append(problems, workflowGraphProblem{Code: "condition_rule_not_an_object", NodeID: nodeID, Field: "cases.conditions"}) + } + selector, ok := r["variableSelector"].([]any) + if !ok { + return append(problems, workflowGraphProblem{Code: "condition_selector_unsupported", NodeID: nodeID, Field: "cases.conditions.variableSelector"}) + } + // The selector needs the node AND a path into its output. The editor hands a new comparison an + // empty list, and one element names the whole output, which desktop refuses and Cloud would read + // as everything the node produced. + if len(selector) < 2 { + problems = append(problems, workflowGraphProblem{Code: "condition_selector_unsupported", NodeID: nodeID, Field: "cases.conditions.variableSelector"}) + } + for _, element := range selector { + if text, ok := element.(string); !ok || text == "" { + problems = append(problems, workflowGraphProblem{Code: "condition_selector_unsupported", NodeID: nodeID, Field: "cases.conditions.variableSelector"}) + break + } + } + if _, ok := workflowConditionOperators[r.S("operator")]; !ok { + problems = append(problems, workflowGraphProblem{Code: "condition_operator_unsupported", NodeID: nodeID, Field: "cases.conditions.operator"}) + } + if text, ok := r["value"].(string); ok && strings.Contains(text, "{{#") && strings.Contains(text, "#}}") { + problems = append(problems, workflowGraphProblem{Code: "condition_value_template_unsupported", NodeID: nodeID, Field: "cases.conditions.value"}) + } + return problems +} + // validateWorkflowEdges checks that every edge joins two nodes that exist. An edge to a node the // document does not contain is a dangling reference the run would have to guess at, and the editor // only produces one when the document was edited outside it. diff --git a/internal/core/workflow_graph_test.go b/internal/core/workflow_graph_test.go index 18cdb34a1..d1a090a6f 100644 --- a/internal/core/workflow_graph_test.go +++ b/internal/core/workflow_graph_test.go @@ -83,6 +83,21 @@ func graphSlot(name, description string) Object { return Object{"name": name, "description": description} } +// graphCondition builds a condition node whose cases the test authors directly. +func graphCondition(id string, cases any) Object { + return Object{"id": id, "type": "workflow", "data": Object{"kind": "condition", "cases": cases}} +} + +// graphConditionCase is one branch: its id, its logic and its comparison rules. +func graphConditionCase(id, logic string, rules []any) Object { + return Object{"id": id, "logic": logic, "conditions": rules} +} + +// graphRule is one comparison inside a case. +func graphRule(selector []any, operator string, value any) Object { + return Object{"variableSelector": selector, "operator": operator, "value": value} +} + // workflowGraph assembles a document. A nil `slots` leaves `executorSlots` off entirely, the way a // graph authored before slots existed is stored. func workflowGraph(nodes, edges, slots []any) Object { @@ -348,6 +363,131 @@ func TestValidateWorkflowGraph(t *testing.T) { } } +// The publish gate refuses every condition configuration Cloud cannot evaluate the way desktop's +// condition.rs does (circus#202 A 条件 2). Each one is a place the evaluator would otherwise answer +// silently — a no-match, an `and`, an else branch, a whole node output, a literal where desktop +// resolves — so a graph would publish and then take a branch nothing in the document asked for. +// +// The cases and workflowUnsupportedConditionConfigs are checked against each other in both +// directions: a declared code with no case here fails, and a case producing a code the list does not +// declare fails too. That is what keeps the ADR's list, the PR's list and the gate's behaviour the +// same list, and what makes circus#210's job — deleting one entry at a time as each difference +// is closed — a change this test refuses to let pass unnoticed. +func TestWorkflowPublishRefusesUnsupportedConditionConfigs(t *testing.T) { + selector := []any{"s", "input", "flag"} + filled := []any{graphRule(selector, "equals", true)} + cases := []struct { + name string + node Object + want []any + }{ + { + name: "cases of the wrong shape", + node: graphCondition("c", "not a list"), + want: []any{Object{"code": "condition_cases_not_a_list", "nodeId": "c", "field": "cases"}}, + }, + { + name: "a case that is not an object", + node: graphCondition("c", []any{"nope"}), + want: []any{Object{"code": "condition_case_not_an_object", "nodeId": "c", "field": "cases"}}, + }, + { + name: "a case with no id", + node: graphCondition("c", []any{graphConditionCase("", "and", filled)}), + want: []any{Object{"code": "condition_case_id_missing", "nodeId": "c", "field": "cases.id"}}, + }, + { + name: "a logic neither side knows", + node: graphCondition("c", []any{graphConditionCase("yes", "xor", filled)}), + want: []any{Object{"code": "condition_logic_unsupported", "nodeId": "c", "field": "cases.logic"}}, + }, + { + name: "rules of the wrong shape", + node: graphCondition("c", []any{Object{"id": "yes", "logic": "and", "conditions": "nope"}}), + want: []any{Object{"code": "condition_rules_not_a_list", "nodeId": "c", "field": "cases.conditions"}}, + }, + { + name: "a rule that is not an object", + node: graphCondition("c", []any{graphConditionCase("yes", "and", []any{"nope"})}), + want: []any{Object{"code": "condition_rule_not_an_object", "nodeId": "c", "field": "cases.conditions"}}, + }, + { + name: "a selector that is not a list", + node: graphCondition("c", []any{graphConditionCase("yes", "and", []any{Object{"variableSelector": "s.input.flag", "operator": "equals"}})}), + want: []any{Object{"code": "condition_selector_unsupported", "nodeId": "c", "field": "cases.conditions.variableSelector"}}, + }, + { + // The editor hands a new comparison an empty selector; this is that state reaching publish. + name: "a selector with no path", + node: graphCondition("c", []any{graphConditionCase("yes", "and", []any{graphRule([]any{}, "equals", true)})}), + want: []any{Object{"code": "condition_selector_unsupported", "nodeId": "c", "field": "cases.conditions.variableSelector"}}, + }, + { + // One element names the whole node output on Cloud and is refused by desktop, so the two + // sides disagree about what the rule even reads. + name: "a single-segment selector", + node: graphCondition("c", []any{graphConditionCase("yes", "and", []any{graphRule([]any{"s"}, "equals", true)})}), + want: []any{Object{"code": "condition_selector_unsupported", "nodeId": "c", "field": "cases.conditions.variableSelector"}}, + }, + { + name: "a path element that is not a name", + node: graphCondition("c", []any{graphConditionCase("yes", "and", []any{graphRule([]any{"s", 0}, "equals", true)})}), + want: []any{Object{"code": "condition_selector_unsupported", "nodeId": "c", "field": "cases.conditions.variableSelector"}}, + }, + { + name: "an operator neither side knows", + node: graphCondition("c", []any{graphConditionCase("yes", "and", []any{graphRule(selector, "matches", true)})}), + want: []any{Object{"code": "condition_operator_unsupported", "nodeId": "c", "field": "cases.conditions.operator"}}, + }, + { + name: "an operator the editor has not filled in yet", + node: graphCondition("c", []any{graphConditionCase("yes", "and", []any{graphRule(selector, "", true)})}), + want: []any{Object{"code": "condition_operator_unsupported", "nodeId": "c", "field": "cases.conditions.operator"}}, + }, + { + // desktop resolves this and keeps the bound type; Cloud would compare the literal text. + name: "an expected value carrying a selector template", + node: graphCondition("c", []any{graphConditionCase("yes", "and", []any{graphRule(selector, "equals", "{{#s.input.name#}}")})}), + want: []any{Object{"code": "condition_value_template_unsupported", "nodeId": "c", "field": "cases.conditions.value"}}, + }, + } + + exercised := map[string]bool{} + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + graph := workflowGraph([]any{graphStart("s"), tc.node}, []any{graphEdge("s", "c")}, nil) + got := problemObjects(validateWorkflowGraph(decodeGraph(t, graph))) + if !reflect.DeepEqual(got, tc.want) { + t.Fatalf("problems = %v, want %v", got, tc.want) + } + for _, problem := range got { + exercised[problem.(Object)["code"].(string)] = true + } + }) + } + declared := map[string]bool{} + for _, code := range workflowUnsupportedConditionConfigs { + declared[code] = true + } + if !reflect.DeepEqual(exercised, declared) { + t.Fatalf("these cases exercise %v; the declared list is %v — the two must name the same set", exercised, declared) + } + + // Refusing is not the same as refusing everything. The two documents the editor legitimately + // publishes — a filled-in node, and one the author has added but not configured yet, which falls + // through to else — must both pass. + for name, node := range map[string]Object{ + "a filled-in condition node": graphCondition("c", []any{graphConditionCase("yes", "and", filled)}), + "a condition node with no cases yet": graphCondition("c", []any{}), + "a condition node with an unfinished case": graphCondition("c", []any{graphConditionCase("yes", "and", []any{})}), + } { + got := problemObjects(validateWorkflowGraph(decodeGraph(t, workflowGraph([]any{graphStart("s"), node}, []any{graphEdge("s", "c")}, nil)))) + if len(got) != 0 { + t.Fatalf("%s was refused: %v", name, got) + } + } +} + // TestRequireValidWorkflowGraphIsOneStableFault covers the wire contract of the publish refusal: one // code whatever went wrong, and every problem carried as a parameter so the editor can mark each // node. A client that branched on a per-violation code would break the day a new rule was added. From 241b2944d4294bf8bc5ed1a997736f3d9dc75b6b Mon Sep 17 00:00:00 2001 From: t Date: Sat, 17 Oct 2026 18:19:30 +0800 Subject: [PATCH 6/6] fix(core): spell the two comments the way the linter reads English golangci-lint's misspell check rejected `recognises` and `behaviour` in the new comments, which is the only thing standing between this branch and a green CI run. No code changes. Co-Authored-By: Claude Code --- internal/core/workflow_graph.go | 2 +- internal/core/workflow_graph_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/core/workflow_graph.go b/internal/core/workflow_graph.go index 67c798ec9..3861d1f47 100644 --- a/internal/core/workflow_graph.go +++ b/internal/core/workflow_graph.go @@ -256,7 +256,7 @@ func validateConditionConfig(nodeID string, data Object) []workflowGraphProblem problems = append(problems, workflowGraphProblem{Code: "condition_case_id_missing", NodeID: nodeID, Field: "cases.id"}) } // An absent logic is `and`, which is what the evaluator applies and what the editor omits; - // only a value neither side recognises is refused. + // only a value neither side recognizes is refused. switch c.S("logic") { case "", "and", "or": default: diff --git a/internal/core/workflow_graph_test.go b/internal/core/workflow_graph_test.go index d1a090a6f..dad47157b 100644 --- a/internal/core/workflow_graph_test.go +++ b/internal/core/workflow_graph_test.go @@ -370,7 +370,7 @@ func TestValidateWorkflowGraph(t *testing.T) { // // The cases and workflowUnsupportedConditionConfigs are checked against each other in both // directions: a declared code with no case here fails, and a case producing a code the list does not -// declare fails too. That is what keeps the ADR's list, the PR's list and the gate's behaviour the +// declare fails too. That is what keeps the ADR's list, the PR's list and the gate's behavior the // same list, and what makes circus#210's job — deleting one entry at a time as each difference // is closed — a change this test refuses to let pass unnoticed. func TestWorkflowPublishRefusesUnsupportedConditionConfigs(t *testing.T) {