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..087a1c9dd
--- /dev/null
+++ b/internal/core/workflow_condition.go
@@ -0,0 +1,267 @@
+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":          workflowCompareNotContains,
+	"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))
+}
+
+// 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 {
+	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
+	default:
+		// 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
+	}
+}
+
+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_graph.go b/internal/core/workflow_graph.go
index 1f798d0e7..3861d1f47 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 recognizes 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..dad47157b 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 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) {
+	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.
diff --git a/internal/core/workflow_scheduler.go b/internal/core/workflow_scheduler.go
new file mode 100644
index 000000000..900e2f6fa
--- /dev/null
+++ b/internal/core/workflow_scheduler.go
@@ -0,0 +1,528 @@
+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(graph, 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. 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
+		}
+		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..7defa254f
--- /dev/null
+++ b/internal/core/workflow_scheduler_db_test.go
@@ -0,0 +1,257 @@
+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"
+	"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)
+	}
+	// 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 {
+		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)
+	}
+}
+
+// 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
new file mode 100644
index 000000000..4f6a40f89
--- /dev/null
+++ b/internal/core/workflow_scheduler_test.go
@@ -0,0 +1,339 @@
+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")
+	}
+}
+
+// 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 {
+	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, "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,
+		"emptyObj": Object{}, "fullObj": Object{"a": 1.0}, "plainObj": map[string]any{},
+	}
+	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},
+		// 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},
+		{"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)
+	}
+}