feat: persist workflow admission and initial execution graph

2026-09-18 01:43:01 +08:00
parent 0f7ac5203b
commit 553a0bf522
19 changed files with 1381 additions and 10 deletions
--- a/internal/storage/mysql/application_resources_upgrade_integration_test.go
+++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go
@@ -64,6 +64,8 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) {
 	retryFinish(t, g, finished, "SUCCEEDED", false)
 	assertApplicationCounts(t, f, 4, 2, 1)
 	// 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。
+	f.exec(`ALTER TABLE weave_waits DROP COLUMN signal_name, DROP COLUMN payload_version, MODIFY COLUMN correlation_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_as_cs NOT NULL COMMENT '外部业务操作关联键，防止旧信号误用于新工作'`)
+	f.exec(`ALTER TABLE weave_instances MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_as_cs NULL COMMENT '业务侧关联键，用于检索追踪，不自动提供业务幂等'`)
 	f.exec(`ALTER TABLE weave_inbox DROP CHECK ck_weave_inbox_source, DROP INDEX uq_weave_task_result, DROP COLUMN source_task_id, DROP COLUMN source_activation_id, DROP COLUMN source_revision, DROP COLUMN snapshot`)
 	f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_queue_business_created, DROP INDEX ix_task_queue_business_state_created, MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_as_cs NULL COMMENT '业务侧关联键，用于检索追踪，不自动提供业务幂等'`)
 	f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_retry_not_self, DROP INDEX uq_task_retry_parent, ADD INDEX fk_pulse_tasks_3(namespace_id,retry_of)`)
--- a/internal/storage/mysql/migrations/000018_workflow_wait_identity.up.sql
+++ b/internal/storage/mysql/migrations/000018_workflow_wait_identity.up.sql
@@ -0,0 +1,7 @@
+-- 冻结等待的信号身份和契约；历史NULL记录必须经固定定义核验，不能默认解释为新等待。
+ALTER TABLE weave_waits
+ ADD COLUMN signal_name VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NULL COMMENT '冻结信号名称；TIMER及待核实历史记录为空',
+ ADD COLUMN payload_version VARCHAR(32) CHARACTER SET ascii COLLATE ascii_bin NULL COMMENT '冻结信号内容契约版本；TIMER及待核实历史记录为空',
+ MODIFY COLUMN correlation_key VARBINARY(764) NOT NULL COMMENT '精确UTF8业务关联且保留尾部空格；新受理最多512字节，兼容历史191字符最多764字节，TIMER为空串';
+
+ALTER TABLE weave_instances MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NULL COMMENT '精确业务追踪键，保留尾部空格，不提供业务幂等';
--- a/internal/storage/mysql/task_store.go
+++ b/internal/storage/mysql/task_store.go
@@ -217,6 +217,11 @@ func businessKeyValue(key string) *string {

 // insertAcceptedTask 要求调用方已持有应用额度和队列锁，并将身份、事件及审计同事务提交。
 func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in service.SubmitInput, retryOf *core.ID) (service.TaskView, error) {
+	return insertAcceptedTaskUntil(ctx, tx, ns, taskID, in, retryOf, nil)
+}
+
+// insertAcceptedTaskUntil 在原任务时限基础上收紧绝对父截止，不重写冻结重试策略。
+func insertAcceptedTaskUntil(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in service.SubmitInput, retryOf *core.ID, parentDeadline *time.Time) (service.TaskView, error) {
 	now, err := dbNow(ctx, tx)
 	if err != nil {
 		return service.TaskView{}, err
@@ -227,6 +232,9 @@ func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in
 		available = *in.AvailableAt
 	}
 	deadline := now.Add(time.Duration(in.TaskTimeoutMS) * time.Millisecond)
+	if parentDeadline != nil {
+		deadline = minimumTime(deadline, *parentDeadline)
+	}
 	if available.Before(now.Add(-time.Second)) || !available.Before(deadline) {
 		return service.TaskView{}, fault.Invalid("available_at must be current/future and before the task deadline")
 	}
--- a/internal/storage/mysql/workflow_apply.go
+++ b/internal/storage/mysql/workflow_apply.go
@@ -0,0 +1,272 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"time"
+
+	core "git.yuntuo.hk/k2safe/orbit/model"
+	"git.yuntuo.hk/k2safe/orbit/weave/definition"
+	"git.yuntuo.hk/k2safe/orbit/weave/engine"
+	"git.yuntuo.hk/k2safe/orbit/weave/instance"
+	weave "git.yuntuo.hk/k2safe/orbit/weave/model"
+)
+
+func createWorkflowSteps(ctx context.Context, tx *sql.Tx, g *workflowGraph, now time.Time) error {
+	create := func(key, kind string) error {
+		id, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, "INSERT INTO weave_steps(namespace_id,id,workflow_id,`key`,kind,created_at,updated_at) VALUES (?,?,?,?,?,?,?)", g.view.NamespaceID, id, g.view.ID, key, kind, now, now); err != nil {
+			return err
+		}
+		g.steps[key] = &workflowStep{view: instance.StepView{ID: id, Key: key, Kind: weave.NodeKind(kind), State: weave.StepBlocked, Revision: 1}}
+		return nil
+	}
+	var visit func(definition.Block) error
+	visit = func(b definition.Block) error {
+		for _, s := range b.Steps {
+			if err := create(s.Key, s.Kind); err != nil {
+				return err
+			}
+			if s.Branch != nil {
+				for _, c := range s.Branch.Cases {
+					if err := visit(c.Body); err != nil {
+						return err
+					}
+				}
+				if err := visit(*s.Branch.Default); err != nil {
+					return err
+				}
+			}
+			if s.Parallel != nil {
+				for _, branch := range s.Parallel.Branches {
+					if err := visit(branch.Body); err != nil {
+						return err
+					}
+				}
+			}
+		}
+		return nil
+	}
+	if err := visit(g.version.Specification.Body); err != nil {
+		return err
+	}
+	if err := create(workflowRootKey, "JOIN"); err != nil {
+		return err
+	}
+	root := g.steps[workflowRootKey]
+	activation, err := core.NewID()
+	if err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,input,created_at,updated_at) VALUES (?,?,?,?,1,'ACTIVE',?,?,?)`, g.view.NamespaceID, activation, g.view.ID, root.view.ID, string(g.view.Input), now, now); err != nil {
+		return err
+	}
+	r, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state='ACTIVE',current_activation_id=? WHERE namespace_id=? AND id=?`, activation, g.view.NamespaceID, root.view.ID)
+	if err := exactlyOne(r, err); err != nil {
+		return err
+	}
+	root.view.State, root.view.ActivationID, root.input = weave.StepActive, &activation, g.view.Input
+	return nil
+}
+
+// applyWorkflowDecision 调用方必须已按应用→排序队列→实例持锁，结果和待续状态一并提交。
+// 初始受理的实例行尚未对其他事务可见，同样由应用锁保护，不在函数内重排父级锁。
+func applyWorkflowDecision(ctx context.Context, tx *sql.Tx, g *workflowGraph, d engine.Decision, now time.Time) error {
+	if len(d.Changes) > 32 {
+		return errors.New("workflow change budget exceeded")
+	}
+	for _, c := range d.Changes {
+		row, ok := g.steps[c.Key]
+		if !ok || string(row.view.Kind) != c.Kind {
+			return errors.New("workflow decision step mismatch")
+		}
+		switch c.Operation {
+		case "ACTIVATE":
+			if row.view.State != weave.StepBlocked || row.view.ActivationID != nil {
+				return errors.New("workflow step already activated")
+			}
+			if err := activateWorkflowStep(ctx, tx, g, row, c, now); err != nil {
+				return err
+			}
+		case "SKIP":
+			if row.view.State != weave.StepBlocked || row.view.ActivationID != nil || c.State.Status != weave.StepSkipped {
+				return errors.New("workflow skip state invalid")
+			}
+		case "COMPLETE":
+			if row.view.ActivationID == nil || (row.view.State != weave.StepActive && row.view.State != weave.StepWaiting) || c.State.Status != weave.StepSucceeded {
+				return errors.New("workflow completion state invalid")
+			}
+			r, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state='SUCCEEDED',output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state IN ('ACTIVE','WAITING')`, string(c.State.Output), now, now, g.view.NamespaceID, row.view.ActivationID, g.view.ID, row.view.ID)
+			if err := exactlyOne(r, err); err != nil {
+				return err
+			}
+		default:
+			return errors.New("unsupported workflow decision")
+		}
+		r, err := tx.ExecContext(ctx, `UPDATE weave_steps SET state=?,current_activation_id=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND revision=?`, c.State.Status, row.view.ActivationID, now, g.view.NamespaceID, row.view.ID, g.view.ID, row.view.Revision)
+		if err := exactlyOne(r, err); err != nil {
+			return err
+		}
+		row.view.State, row.input, row.output = c.State.Status, c.State.Input, c.State.Output
+		row.view.Revision++
+		if err := emitEvent(ctx, tx, g.view.NamespaceID, row.view.ID, row.view.Revision, "workflow_step", "workflow.step.changed", map[string]any{"workflow_id": g.view.ID, "key": c.Key, "state": c.State.Status, "activation_id": row.view.ActivationID}); err != nil {
+			return err
+		}
+	}
+	root := g.steps[workflowRootKey]
+	if root == nil || root.view.ActivationID == nil {
+		return errors.New("workflow root missing")
+	}
+	state := weave.InstanceRunning
+	var completed *time.Time
+	var output any
+	rootState := weave.StepActive
+	if d.Complete {
+		if d.More {
+			return errors.New("workflow complete decision has pending work")
+		}
+		for key, row := range g.steps {
+			if key != workflowRootKey && row.view.State != weave.StepSucceeded && row.view.State != weave.StepSkipped {
+				return errors.New("workflow completion has unfinished steps")
+			}
+		}
+		state, rootState, completed, output = weave.InstanceSucceeded, weave.StepSucceeded, &now, string(d.Output)
+	} else if !d.More {
+		waiting, active := false, false
+		for key, row := range g.steps {
+			if key == workflowRootKey {
+				continue
+			}
+			if row.view.State == weave.StepWaiting {
+				waiting = true
+			}
+			if row.view.State == weave.StepActive && (row.view.Kind == weave.NodeTask || row.view.Kind == weave.NodePoll) {
+				active = true
+			}
+		}
+		if waiting && !active {
+			state = weave.InstanceWaiting
+		}
+	}
+	r, err := tx.ExecContext(ctx, `UPDATE weave_activations SET state=?,output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND workflow_id=? AND step_id=? AND state='ACTIVE'`, rootState, output, completed, now, g.view.NamespaceID, root.view.ActivationID, g.view.ID, root.view.ID)
+	if err := exactlyOne(r, err); err != nil {
+		return err
+	}
+	r, err = tx.ExecContext(ctx, `UPDATE weave_steps SET state=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, rootState, now, g.view.NamespaceID, root.view.ID, root.view.Revision)
+	if err := exactlyOne(r, err); err != nil {
+		return err
+	}
+	root.view.Revision++
+	root.view.State = rootState
+	// 包括首批决策在内，每次持久化推进递增实例版本；不能以未改变字段的affected_rows判断存在性。
+	r, err = tx.ExecContext(ctx, `UPDATE weave_instances SET state=?,output=?,completed_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=? AND revision=?`, state, output, completed, now, g.view.NamespaceID, g.view.ID, g.view.Revision)
+	if err := exactlyOne(r, err); err != nil {
+		return err
+	}
+	if d.Complete {
+		if _, err := tx.ExecContext(ctx, `UPDATE pulse_execution_gates SET paused=TRUE,revision=revision+1,updated_at=? WHERE namespace_id=? AND id=?`, now, g.view.NamespaceID, g.gate); err != nil {
+			return err
+		}
+	}
+	if d.More {
+		id, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		_, err = tx.ExecContext(ctx, `INSERT INTO weave_pending_advances(namespace_id,id,workflow_id,step_id,expected_step_revision,checkpoint,available_at,created_at,updated_at) VALUES (?,?,?,?,?,'{"schema_version":1,"kind":"ROOT"}',?,?,?) ON DUPLICATE KEY UPDATE expected_step_revision=VALUES(expected_step_revision),checkpoint=VALUES(checkpoint),available_at=VALUES(available_at),revision=revision+1,updated_at=VALUES(updated_at)`, g.view.NamespaceID, id, g.view.ID, root.view.ID, root.view.Revision, now, now, now)
+		if err != nil {
+			return err
+		}
+	} else {
+		if _, err := tx.ExecContext(ctx, `DELETE FROM weave_pending_advances WHERE namespace_id=? AND workflow_id=? AND step_id=?`, g.view.NamespaceID, g.view.ID, root.view.ID); err != nil {
+			return err
+		}
+	}
+	g.view.State = state
+	g.view.Revision++
+	g.view.CompletedAt = completed
+	g.view.Output = d.Output
+	return nil
+}
+
+func activateWorkflowStep(ctx context.Context, tx *sql.Tx, g *workflowGraph, row *workflowStep, c engine.Change, now time.Time) error {
+	id, err := core.NewID()
+	if err != nil {
+		return err
+	}
+	var taskID *core.ID
+	if c.Task != nil {
+		v, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		taskID = &v
+	}
+	var direct *core.ID
+	if c.Kind == "TASK" {
+		direct = taskID
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO weave_activations(namespace_id,id,workflow_id,step_id,sequence,state,task_id,input,created_at,updated_at) VALUES (?,?,?,?,1,?,?,?,?,?)`, g.view.NamespaceID, id, g.view.ID, row.view.ID, c.State.Status, direct, string(c.State.Input), now, now); err != nil {
+		return err
+	}
+	row.view.ActivationID = &id
+	row.view.TaskID = direct
+	deadline := g.view.DeadlineAt
+	if c.Kind == "POLL" {
+		node, ok := g.program.Step(c.Key)
+		if !ok || node.Poll == nil || taskID == nil {
+			return errors.New("workflow poll decision incomplete")
+		}
+		p := node.Poll
+		deadline = minimumTime(deadline, now.Add(time.Duration(p.TimeoutMS)*time.Millisecond))
+		pollID, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_poll_states(namespace_id,id,workflow_id,activation_id,max_rounds,used_rounds,max_attempts_per_round,interval_ms,deadline_at,current_task_id,created_at,updated_at) VALUES (?,?,?,?,?,1,?,?,?,?,?,?)`, g.view.NamespaceID, pollID, g.view.ID, id, p.MaxRounds, p.Task.Retry.MaxAttempts, p.IntervalMS, deadline, taskID, now, now); err != nil {
+			return err
+		}
+		roundID, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, "INSERT INTO weave_poll_rounds(namespace_id,id,workflow_id,activation_id,`round`,task_id,created_at,updated_at) VALUES (?,?,?,?,1,?,?,?)", g.view.NamespaceID, roundID, g.view.ID, id, taskID, now, now); err != nil {
+			return err
+		}
+	}
+	if c.Wait != nil {
+		waitID, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		until := minimumTime(deadline, now.Add(time.Duration(c.Wait.TimeoutMS)*time.Millisecond))
+		var name, version any
+		if c.Wait.Kind != "TIMER" {
+			name, version = c.Wait.Name, c.Wait.PayloadVersion
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at,signal_name,payload_version,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, g.view.NamespaceID, waitID, g.view.ID, id, c.Wait.Kind, []byte(c.Wait.Correlation), until, name, version, now, now); err != nil {
+			return err
+		}
+	}
+	if c.Task != nil {
+		if taskID == nil {
+			return errors.New("workflow task identity missing")
+		}
+		task, err := insertAcceptedTaskUntil(ctx, tx, g.view.NamespaceID, *taskID, *c.Task, nil, &deadline)
+		if err != nil {
+			return err
+		}
+		r, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET owner_kind=?,owner_id=?,gate_id=? WHERE namespace_id=? AND id=? AND owner_kind IS NULL`, engine.TaskOwnerKind, id, g.gate, g.view.NamespaceID, *taskID)
+		if err := exactlyOne(r, err); err != nil {
+			return err
+		}
+		if err := emitEvent(ctx, tx, g.view.NamespaceID, *taskID, task.Revision, "task", "task.accepted", map[string]any{"task_id": task.ID, "state": task.State, "workflow_id": g.view.ID, "activation_id": id}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
--- a/internal/storage/mysql/workflow_instances.go
+++ b/internal/storage/mysql/workflow_instances.go
@@ -0,0 +1,292 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"git.yuntuo.hk/k2safe/orbit/internal/fault"
+	"git.yuntuo.hk/k2safe/orbit/internal/jsonwire"
+	core "git.yuntuo.hk/k2safe/orbit/model"
+	"git.yuntuo.hk/k2safe/orbit/platform/auth"
+	"git.yuntuo.hk/k2safe/orbit/weave/definition"
+	"git.yuntuo.hk/k2safe/orbit/weave/engine"
+	"git.yuntuo.hk/k2safe/orbit/weave/instance"
+)
+
+// WorkflowInstanceStore 以应用额度锁串行受理，在一个事务内建立全部归属及初始工作。
+// 当前为内部存储适配层，公共HTTP和SDK在消费运行器接通后开放。
+type WorkflowInstanceStore struct{ *TaskStore }
+
+func NewWorkflowInstanceStore(db *sql.DB) *WorkflowInstanceStore {
+	return &WorkflowInstanceStore{NewTaskStore(db)}
+}
+
+var _ instance.Store = (*WorkflowInstanceStore)(nil)
+
+// 根步骤是实例级有限推进的检查点；$不属于合法的用户步骤字符，不能与定义冲突。
+const workflowRootKey = "$root"
+
+type workflowGraph struct {
+	view    instance.View
+	gate    core.ID
+	version definition.VersionView
+	program *engine.Program
+	steps   map[string]*workflowStep
+}
+type workflowStep struct {
+	view   instance.StepView
+	input  json.RawMessage
+	output json.RawMessage
+}
+type workflowPolicy struct {
+	SchemaVersion   uint32  `json:"schema_version"`    // SchemaVersion 冻结策略快照的结构版本，当前为1。
+	VersionID       core.ID `json:"version_id"`        // VersionID 固定的不可变定义版本，读取时核对实例归属。
+	ContentDigest   string  `json:"content_digest"`    // ContentDigest 固定定义的SHA-256摘要，防止策略和规则错配。
+	FailurePolicy   string  `json:"failure_policy"`    // FailurePolicy 原定义的失败结清策略，不随新发布版本变化。
+	MaxTaskAttempts uint32  `json:"max_task_attempts"` // MaxTaskAttempts 所有路径的静态尝试上界，实际预算由任务与轮次独立保存。
+}
+
+func readWorkflowInstance(ctx context.Context, tx *sql.Tx, ns, id core.ID, lock bool) (instance.View, core.ID, error) {
+	var v instance.View
+	var gate core.ID
+	q := `SELECT i.id,i.namespace_id,i.definition_id,i.definition_version_id,v.version,i.gate_id,i.state,i.revision,i.paused,i.control_revision,i.input,i.output,COALESCE(i.business_key,''),i.deadline_at,i.cancel_requested_at,i.completed_at,i.created_at,i.updated_at
+ FROM weave_instances i JOIN weave_definition_versions v ON v.namespace_id=i.namespace_id AND v.definition_id=i.definition_id AND v.id=i.definition_version_id
+ JOIN weave_definitions d ON d.namespace_id=i.namespace_id AND d.id=i.definition_id
+ JOIN pulse_execution_gates g ON g.namespace_id=i.namespace_id AND g.id=i.gate_id
+ WHERE i.namespace_id=? AND i.id=?`
+	if lock {
+		q += ` FOR SHARE`
+	}
+	var input, output []byte
+	err := tx.QueryRowContext(ctx, q, ns, id).Scan(&v.ID, &v.NamespaceID, &v.DefinitionID, &v.DefinitionVersionID, &v.Version, &gate, &v.State, &v.Revision, &v.Paused, &v.ControlRevision, &input, &output, &v.BusinessKey, &v.DeadlineAt, &v.CancelRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt)
+	v.Input, v.Output = input, output
+	if errors.Is(err, sql.ErrNoRows) {
+		err = fault.NotFound
+	}
+	return v, gate, err
+}
+func instanceVersion(ctx context.Context, tx *sql.Tx, v instance.View) (definition.VersionView, error) {
+	ver, err := readWorkflowVersion(ctx, tx, v.NamespaceID, v.DefinitionID, v.Version)
+	if err != nil {
+		return ver, err
+	}
+	if ver.ID != v.DefinitionVersionID {
+		return ver, errors.New("workflow fixed version mismatch")
+	}
+	if definition.ValidateData(ver.Specification.InputType, v.Input) != nil {
+		return ver, errors.New("workflow stored input invalid")
+	}
+	if v.State == "SUCCEEDED" && definition.ValidateData(ver.Specification.Body.OutputType, v.Output) != nil {
+		return ver, errors.New("workflow stored output invalid")
+	}
+	var raw []byte
+	if err := tx.QueryRowContext(ctx, `SELECT policy_snapshot FROM weave_instances WHERE namespace_id=? AND id=?`, v.NamespaceID, v.ID).Scan(&raw); err != nil {
+		return ver, err
+	}
+	var policy workflowPolicy
+	if json.Unmarshal(raw, &policy) != nil || policy != (workflowPolicy{1, ver.ID, ver.ContentDigest, ver.Specification.FailurePolicy, ver.Analysis.MaxTaskAttempts}) {
+		return ver, errors.New("workflow frozen policy mismatch")
+	}
+	return ver, nil
+}
+func lockWorkflowQueues(ctx context.Context, tx *sql.Tx, ns core.ID, p auth.Permissions, queues []core.ID) error {
+	// Checked.Queues始终排序去重。这里直接取排他锁，禁止先共享后升级。
+	for _, id := range queues {
+		if !p.AllowsQueue(id) {
+			return fault.NotFound
+		}
+		if _, _, err := lockQueueAfterApplication(ctx, tx, ns, id); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *WorkflowInstanceStore) Start(ctx context.Context, t auth.Token, key string, in instance.StartInput) (instance.StartResult, error) {
+	var out instance.StartResult
+	err := s.transaction(ctx, t, instance.StartAction, func(tx *sql.Tx, p auth.Permissions) error {
+		if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil {
+			return err
+		}
+		id, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "workflow.start", "", key, "workflow", id, in)
+		if err != nil {
+			return err
+		}
+		if _, err := readWorkflowDefinition(ctx, tx, t.NamespaceID, in.DefinitionID, false); err != nil {
+			return err
+		}
+		version, err := readWorkflowVersion(ctx, tx, t.NamespaceID, in.DefinitionID, in.Version)
+		if err != nil {
+			return err
+		}
+		if len(version.Analysis.Queues) > 0 && !p.Allows("tasks:submit") {
+			return fault.Forbidden
+		}
+		if err := lockWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues); err != nil {
+			return err
+		}
+		if cached != nil {
+			if err := json.Unmarshal(cached, &out); err != nil {
+				return err
+			}
+			current, _, err := readWorkflowInstance(ctx, tx, t.NamespaceID, out.Instance.ID, false)
+			if err != nil {
+				return err
+			}
+			if current.DefinitionVersionID != version.ID || out.Instance.DefinitionVersionID != version.ID || out.Instance.NamespaceID != t.NamespaceID || !current.CreatedAt.Equal(out.Instance.CreatedAt) {
+				return errors.New("workflow start receipt mismatch")
+			}
+			out.Deduplicated = true
+			return nil
+		}
+		if err := definition.ValidateData(version.Specification.InputType, in.Input); err != nil {
+			return fault.Invalid("workflow input does not match fixed version")
+		}
+		var count int
+		if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM (SELECT id FROM weave_instances WHERE namespace_id=? AND state IN ('RUNNING','WAITING','CANCELLING','NEEDS_REVIEW') LIMIT ?) active_workflows`, t.NamespaceID, instance.MaxUnfinished).Scan(&count); err != nil {
+			return err
+		}
+		if count >= instance.MaxUnfinished {
+			return fault.New("WORKFLOW_INSTANCE_LIMIT", "namespace unfinished workflow limit reached")
+		}
+		program, err := engine.Compile(version.Specification)
+		if err != nil {
+			return err
+		}
+		decision, err := program.Plan(engine.Snapshot{WorkflowID: id, BusinessKey: in.BusinessKey, Input: in.Input, Steps: map[string]engine.State{}, AllowNew: true}, 32)
+		if err != nil {
+			return fault.Invalid("workflow initial step input is invalid")
+		}
+		now, err := dbNow(ctx, tx)
+		if err != nil {
+			return err
+		}
+		gate, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		deadline := now.Add(time.Duration(version.Specification.TimeoutMS) * time.Millisecond)
+		policy, err := jsonwire.Marshal(workflowPolicy{1, version.ID, version.ContentDigest, version.Specification.FailurePolicy, version.Analysis.MaxTaskAttempts})
+		if err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO pulse_execution_gates(namespace_id,id,created_at,updated_at) VALUES (?,?,?,?)`, t.NamespaceID, gate, now, now); err != nil {
+			return err
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,input,policy_snapshot,business_key,deadline_at,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)`, t.NamespaceID, id, in.DefinitionID, version.ID, gate, string(in.Input), string(policy), businessKeyValue(in.BusinessKey), deadline, now, now); err != nil {
+			return err
+		}
+		graph := workflowGraph{gate: gate, version: version, program: program, steps: map[string]*workflowStep{}}
+		graph.view, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
+		if err != nil {
+			return err
+		}
+		if err := createWorkflowSteps(ctx, tx, &graph, now); err != nil {
+			return err
+		}
+		if err := applyWorkflowDecision(ctx, tx, &graph, decision, now); err != nil {
+			return err
+		}
+		out.Instance, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
+		if err != nil {
+			return err
+		}
+		meta := map[string]any{"definition_id": in.DefinitionID, "definition_version_id": version.ID, "state": out.Instance.State, "deadline_at": deadline, "steps": version.Analysis.Steps}
+		if err := emitEvent(ctx, tx, t.NamespaceID, id, out.Instance.Revision, "workflow", "workflow.started", meta); err != nil {
+			return err
+		}
+		if err := appendAudit(ctx, tx, t, identity, id, "workflow", "workflow.start", "start fixed workflow version", nil, meta); err != nil {
+			return err
+		}
+		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
+	})
+	return out, err
+}
+
+func (s *WorkflowInstanceStore) Get(ctx context.Context, t auth.Token, id core.ID) (instance.View, error) {
+	var out instance.View
+	err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
+		var err error
+		out, _, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
+		if err != nil {
+			return err
+		}
+		version, err := instanceVersion(ctx, tx, out)
+		if err != nil {
+			return err
+		}
+		return authorizeWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues)
+	})
+	return out, err
+}
+func (s *WorkflowInstanceStore) Steps(ctx context.Context, t auth.Token, id core.ID) (instance.StepsView, error) {
+	out := instance.StepsView{InstanceID: id, Steps: []instance.StepView{}}
+	err := s.transaction(ctx, t, instance.ReadAction, func(tx *sql.Tx, p auth.Permissions) error {
+		// 先核对队列，后锁实例；未来运行器遵守应用→队列→实例，读取不得逆序锁队列。
+		v, _, err := readWorkflowInstance(ctx, tx, t.NamespaceID, id, false)
+		if err != nil {
+			return err
+		}
+		version, err := instanceVersion(ctx, tx, v)
+		if err != nil {
+			return err
+		}
+		if err := authorizeWorkflowQueues(ctx, tx, t.NamespaceID, p, version.Analysis.Queues); err != nil {
+			return err
+		}
+		var gate core.ID
+		v, gate, err = readWorkflowInstance(ctx, tx, t.NamespaceID, id, true)
+		if err != nil {
+			return err
+		}
+		out.Revision = v.Revision
+		program, err := engine.Compile(version.Specification)
+		if err != nil {
+			return err
+		}
+		rows, err := tx.QueryContext(ctx, "SELECT s.id,s.`key`,s.kind,s.state,s.revision,s.current_activation_id,a.task_id,a.id,a.state,task.id,task.queue_id FROM weave_steps s LEFT JOIN weave_activations a ON a.namespace_id=s.namespace_id AND a.id=s.current_activation_id AND a.workflow_id=s.workflow_id AND a.step_id=s.id LEFT JOIN pulse_tasks task ON task.namespace_id=s.namespace_id AND task.id=a.task_id AND task.owner_kind='workflow_activation' AND task.owner_id=a.id AND task.gate_id=? WHERE s.namespace_id=? AND s.workflow_id=? AND s.`key`<>? ORDER BY s.`key` LIMIT 129", gate, t.NamespaceID, id, workflowRootKey)
+		if err != nil {
+			return err
+		}
+		defer rows.Close()
+		for rows.Next() {
+			var step instance.StepView
+			var activation *core.ID
+			var task, queue *core.ID
+			var state sql.NullString
+			if err := rows.Scan(&step.ID, &step.Key, &step.Kind, &step.State, &step.Revision, &step.ActivationID, &step.TaskID, &activation, &state, &task, &queue); err != nil {
+				return err
+			}
+			node, ok := program.Step(step.Key)
+			if !ok || node.Kind != string(step.Kind) {
+				return errors.New("workflow step definition mismatch")
+			}
+			inactive := step.State == "BLOCKED" || step.State == "SKIPPED"
+			if inactive && step.ActivationID != nil || !inactive && (step.ActivationID == nil || activation == nil || *activation != *step.ActivationID || state.String != string(step.State)) {
+				return errors.New("workflow current activation mismatch")
+			}
+			if !inactive && (step.Kind == "TASK") != (step.TaskID != nil) {
+				return errors.New("workflow direct task ownership mismatch")
+			}
+			if step.TaskID != nil && (task == nil || queue == nil || *task != *step.TaskID || node.Task == nil || *queue != node.Task.QueueID) {
+				return errors.New("workflow task association invalid")
+			}
+			out.Steps = append(out.Steps, step)
+		}
+		if err := rows.Err(); err != nil {
+			return err
+		}
+		if len(out.Steps) != int(version.Analysis.Steps) {
+			return errors.New("workflow step set incomplete")
+		}
+		return nil
+	})
+	return out, err
+}
--- a/internal/storage/mysql/workflow_instances_integration_test.go
+++ b/internal/storage/mysql/workflow_instances_integration_test.go
@@ -0,0 +1,378 @@
+package mysql
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"git.yuntuo.hk/k2safe/orbit/internal/fault"
+	core "git.yuntuo.hk/k2safe/orbit/model"
+	"git.yuntuo.hk/k2safe/orbit/platform/auth"
+	"git.yuntuo.hk/k2safe/orbit/pulse/service"
+	"git.yuntuo.hk/k2safe/orbit/weave/definition"
+	"git.yuntuo.hk/k2safe/orbit/weave/instance"
+)
+
+func instanceCount(f *executionFixture, q string, args ...any) int {
+	f.t.Helper()
+	var n int
+	if err := f.db.QueryRowContext(f.ctx, q, args...).Scan(&n); err != nil {
+		f.t.Fatal(err)
+	}
+	return n
+}
+
+func grantWorkflowInstance(t *testing.T, f *executionFixture, queues ...string) {
+	t.Helper()
+	if len(queues) == 0 {
+		queues = []string{"*"}
+	}
+	p := auth.Permissions{Actions: []string{instance.StartAction, instance.ReadAction, definition.ReadAction, definition.PublishAction, "tasks:submit", "tasks:read", "tasks:cancel", "workers:connect", "workers:consume"}, Queues: queues}
+	raw, err := json.Marshal(p)
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), f.resources.NamespaceID, f.resources.CredentialID)
+}
+func publishInstanceSpec(t *testing.T, f *executionFixture, spec definition.Specification) instance.StartInput {
+	t.Helper()
+	api := definition.New(NewWorkflowDefinitionStore(f.db))
+	c, err := api.Create(f.ctx, f.token, string(executionID(t)), definition.CreateInput{Name: "instance-" + string(executionID(t))})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := api.Publish(f.ctx, f.token, c.Definition.ID, "v1", definition.PublishInput{Version: "v1", Specification: spec, Reason: "verify instance admission"}); err != nil {
+		t.Fatal(err)
+	}
+	return instance.StartInput{DefinitionID: c.Definition.ID, Version: "v1", Input: json.RawMessage(`{"n":9007199254740993}`), BusinessKey: "作业关联 "}
+}
+func allInitialNodes(queue core.ID) definition.Specification {
+	spec := workflowTestSpec(queue)
+	typ := spec.InputType
+	work := *spec.Body.Steps[0].Task
+	work.Input = definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "input"}}
+	signal := definition.Signal{Name: strings.Repeat("s", 128), PayloadVersion: "v1", Correlation: definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`"` + strings.Repeat("x", 511) + ` "`)}, PayloadType: typ, TimeoutMS: 4000}
+	pollTask := work
+	pollTask.ResultType = definition.Type{Kind: "STRING", Enum: []string{"PENDING", "DONE"}}
+	pollSignal := signal
+	pollSignal.PayloadType = pollTask.ResultType
+	pollSignal.TimeoutMS = 5000
+	poll := definition.Poll{Task: pollTask, MaxRounds: 3, TimeoutMS: 5000, IntervalMS: 1000, Pending: []string{"PENDING"}, Succeeded: []string{"DONE"}, FinalSignal: &pollSignal}
+	defaultBlock := definition.Block{Steps: []definition.Step{{Key: "chosen", Kind: "TASK", Task: &work}}, OutputType: typ, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "chosen"}}}
+	left, right := definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`0`)}, definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`1`)}
+	branch := definition.Branch{Cases: []definition.BranchCase{{When: definition.Predicate{Op: "EQ", Left: &left, Right: &right}, Body: definition.Block{Steps: []definition.Step{{Key: "skipped", Kind: "TASK", Task: &work}}, OutputType: typ, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "skipped"}}}}}, Default: &defaultBlock}
+	steps := []definition.Step{{Key: "job", Kind: "TASK", Task: &work}, {Key: "query", Kind: "POLL", Poll: &poll}, {Key: "signal", Kind: "SIGNAL", Signal: &signal}, {Key: "timer", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 5000}}, {Key: "choice", Kind: "BRANCH", Branch: &branch}}
+	parallel := definition.Parallel{}
+	out := definition.Type{Kind: "OBJECT", Fields: map[string]definition.Type{}}
+	for _, step := range steps {
+		result := typ
+		if step.Kind == "POLL" {
+			result = pollTask.ResultType
+		}
+		if step.Kind == "TIMER" {
+			result = definition.Type{Kind: "NULL"}
+		}
+		parallel.Branches = append(parallel.Branches, definition.ParallelBranch{Name: step.Key, Body: definition.Block{Steps: []definition.Step{step}, OutputType: result, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: step.Key}}}})
+		out.Fields[step.Key] = result
+	}
+	spec.Body = definition.Block{Steps: []definition.Step{{Key: "join", Kind: "PARALLEL", Parallel: &parallel}}, OutputType: out, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "join"}}}
+	return spec
+}
+
+func TestIntegrationWorkflowInstanceAdmission(t *testing.T) {
+	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
+	if dsn == "" {
+		t.Skip("requires disposable MySQL")
+	}
+	ctx, stop := context.WithTimeout(context.Background(), 120*time.Second)
+	defer stop()
+	db, err := Open(ctx, dsn, 24, 8)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if err := Migrate(ctx, db); err != nil {
+		t.Fatal(err)
+	}
+	api := instance.New(NewWorkflowInstanceStore(db))
+	t.Run("concurrent_identity_real_worker_and_fixed_version", func(t *testing.T) {
+		f := newExecutionFixture(t, ctx, db)
+		grantWorkflowInstance(t, f)
+		in := publishInstanceSpec(t, f, workflowTestSpec(f.resources.QueueID))
+		type result struct {
+			out instance.StartResult
+			err error
+		}
+		ch := make(chan result, 12)
+		var wg sync.WaitGroup
+		for i := 0; i < 12; i++ {
+			wg.Add(1)
+			go func() { defer wg.Done(); v, e := api.Start(ctx, f.token, "one-start", in); ch <- result{v, e} }()
+		}
+		wg.Wait()
+		close(ch)
+		var original instance.View
+		fresh := 0
+		for r := range ch {
+			if r.err != nil {
+				t.Fatal(r.err)
+			}
+			if !r.out.Deduplicated {
+				fresh++
+				original = r.out.Instance
+			}
+		}
+		if fresh != 1 || original.ID == "" || string(original.Input) != string(in.Input) || original.BusinessKey != in.BusinessKey {
+			t.Fatal("start identity or precision", fresh, original)
+		}
+		if instanceCount(f, `SELECT COUNT(*) FROM weave_instances WHERE namespace_id=?`, f.resources.NamespaceID) != 1 || instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 1 {
+			t.Fatal("duplicate admission")
+		}
+		steps, err := api.Steps(ctx, f.token, original.ID)
+		if err != nil || len(steps.Steps) != 1 || steps.Steps[0].TaskID == nil {
+			t.Fatal(steps, err)
+		}
+		taskID := *steps.Steps[0].TaskID
+		if _, err := f.tasks.CancelTask(ctx, f.token, taskID, "direct-cancel", service.CancelInput{ExpectedRevision: 1, Reason: "owned task"}); fault.Code(err) != "STATE_CONFLICT" {
+			t.Fatal("workflow ownership bypassed", err)
+		}
+		ref := retryStart(t, f)
+		if ref.TaskID != taskID {
+			t.Fatal("wrong claimed task")
+		}
+		retryFinish(t, f, ref, "SUCCEEDED", false)
+		if instanceCount(f, `SELECT COUNT(*) FROM weave_inbox WHERE namespace_id=? AND workflow_id=? AND source_task_id=? AND state='PENDING'`, f.resources.NamespaceID, original.ID, taskID) != 1 {
+			t.Fatal("real instance result not durably delivered")
+		}
+		// 查询和重建存储对象不能隐式消费；下一阶段运行器必须消费已持久化的事件。
+		after, err := instance.New(NewWorkflowInstanceStore(db)).Get(ctx, f.token, original.ID)
+		if err != nil || after.State != "RUNNING" {
+			t.Fatal(after, err)
+		}
+		changed := workflowTestSpec(f.resources.QueueID)
+		changed.TimeoutMS = 60000
+		if _, err := definition.New(NewWorkflowDefinitionStore(db)).Publish(ctx, f.token, in.DefinitionID, "v2", definition.PublishInput{Version: "v2", Specification: changed, Reason: "new release"}); err != nil {
+			t.Fatal(err)
+		}
+		dupe, err := api.Start(ctx, f.token, "one-start", in)
+		if err != nil || !dupe.Deduplicated || dupe.Instance.ID != original.ID || !dupe.Instance.DeadlineAt.Equal(original.DeadlineAt) || dupe.Instance.DefinitionVersionID != original.DefinitionVersionID {
+			t.Fatal("replay changed fixed admission", err)
+		}
+		in.Input = json.RawMessage(`{"n":9007199254740992}`)
+		if _, err := api.Start(ctx, f.token, "one-start", in); fault.Code(err) != "IDEMPOTENCY_CONFLICT" {
+			t.Fatal(err)
+		}
+	})
+	t.Run("all_node_kinds_freeze_waits_and_poll_deadline", func(t *testing.T) {
+		f := newExecutionFixture(t, ctx, db)
+		grantWorkflowInstance(t, f)
+		in := publishInstanceSpec(t, f, allInitialNodes(f.resources.QueueID))
+		started, err := api.Start(ctx, f.token, "mixed", in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		steps, err := api.Steps(ctx, f.token, started.Instance.ID)
+		if err != nil || len(steps.Steps) != 8 {
+			t.Fatal(steps, err)
+		}
+		if instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 3 || instanceCount(f, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=?`, f.resources.NamespaceID) != 3 {
+			t.Fatal("task/wait counts")
+		}
+		if instanceCount(f, `SELECT COUNT(*) FROM weave_waits WHERE namespace_id=? AND kind IN ('SIGNAL','POLL') AND OCTET_LENGTH(correlation_key)=512 AND OCTET_LENGTH(signal_name)=128 AND payload_version='v1'`, f.resources.NamespaceID) != 2 {
+			t.Fatal("signal metadata truncated")
+		}
+		if instanceCount(f, `SELECT COUNT(*) FROM weave_steps WHERE namespace_id=? AND state='SKIPPED' AND current_activation_id IS NULL`, f.resources.NamespaceID) != 1 {
+			t.Fatal("unselected path activated")
+		}
+		var used, max uint32
+		var deadline, taskDeadline, waitDeadline time.Time
+		var direct *core.ID
+		if err := db.QueryRowContext(ctx, `SELECT p.used_rounds,p.max_rounds,p.deadline_at,t.deadline_at,w.deadline_at,a.task_id FROM weave_poll_states p JOIN pulse_tasks t ON t.namespace_id=p.namespace_id AND t.id=p.current_task_id JOIN weave_activations a ON a.namespace_id=p.namespace_id AND a.id=p.activation_id JOIN weave_waits w ON w.namespace_id=p.namespace_id AND w.activation_id=p.activation_id WHERE p.namespace_id=?`, f.resources.NamespaceID).Scan(&used, &max, &deadline, &taskDeadline, &waitDeadline, &direct); err != nil {
+			t.Fatal(err)
+		}
+		if used != 1 || max != 3 || direct != nil || !deadline.Equal(taskDeadline) || !deadline.Equal(waitDeadline) || !deadline.Before(started.Instance.DeadlineAt) {
+			t.Fatal("poll budget or parent deadline not frozen")
+		}
+	})
+	t.Run("waiting_and_empty_completion", func(t *testing.T) {
+		for _, empty := range []bool{false, true} {
+			f := newExecutionFixture(t, ctx, db)
+			grantWorkflowInstance(t, f)
+			spec := workflowTestSpec(f.resources.QueueID)
+			spec.Body.Steps = nil
+			spec.Body.Output = definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "input"}}
+			if !empty {
+				spec.Body.Steps = []definition.Step{{Key: "wait", Kind: "TIMER", Timer: &definition.Timer{DelayMS: 1000}}}
+			}
+			in := publishInstanceSpec(t, f, spec)
+			v, err := api.Start(ctx, f.token, "start", in)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if empty {
+				if v.Instance.State != "SUCCEEDED" || v.Instance.CompletedAt == nil || string(v.Instance.Output) != string(in.Input) {
+					t.Fatal(v)
+				}
+			} else if v.Instance.State != "WAITING" || v.Instance.CompletedAt != nil {
+				t.Fatal(v)
+			}
+			if instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 0 {
+				t.Fatal("wait allocated worker task")
+			}
+		}
+	})
+	t.Run("bounded_initial_decision_persists_root_checkpoint", func(t *testing.T) {
+		f := newExecutionFixture(t, ctx, db)
+		grantWorkflowInstance(t, f)
+		spec := workflowTestSpec(f.resources.QueueID)
+		work := *spec.Body.Steps[0].Task
+		left, right := definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`0`)}, definition.Value{Kind: "LITERAL", Literal: json.RawMessage(`1`)}
+		branch := definition.Branch{Default: &spec.Body}
+		for i := 0; i < 16; i++ {
+			body := definition.Block{OutputType: spec.InputType, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "input"}}}
+			for j := 0; j < 7; j++ {
+				body.Steps = append(body.Steps, definition.Step{Key: fmt.Sprintf("unused_%d_%d", i, j), Kind: "TASK", Task: &work})
+			}
+			branch.Cases = append(branch.Cases, definition.BranchCase{When: definition.Predicate{Op: "EQ", Left: &left, Right: &right}, Body: body})
+		}
+		// 默认块单独保存，避免构造时让指针指向随后被替换的根块。
+		fallback := spec.Body
+		branch.Default = &fallback
+		spec.Body = definition.Block{Steps: []definition.Step{{Key: "choose", Kind: "BRANCH", Branch: &branch}}, OutputType: spec.InputType, Output: definition.Value{Kind: "REFERENCE", Reference: &definition.Reference{Source: "choose"}}}
+		in := publishInstanceSpec(t, f, spec)
+		out, err := api.Start(ctx, f.token, "budget", in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if instanceCount(f, `SELECT COUNT(*) FROM weave_steps WHERE namespace_id=? AND state='SKIPPED'`, f.resources.NamespaceID) != 31 || instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 0 {
+			t.Fatal("initial decision exceeded budget")
+		}
+		if instanceCount(f, "SELECT COUNT(*) FROM weave_pending_advances p JOIN weave_steps s ON s.namespace_id=p.namespace_id AND s.workflow_id=p.workflow_id AND s.id=p.step_id WHERE p.namespace_id=? AND s.`key`='$root' AND p.expected_step_revision=s.revision", f.resources.NamespaceID) != 1 {
+			t.Fatal("durable checkpoint missing")
+		}
+		steps, err := instance.New(NewWorkflowInstanceStore(db)).Steps(ctx, f.token, out.Instance.ID)
+		if err != nil || len(steps.Steps) != 114 {
+			t.Fatal(steps, err)
+		}
+		if _, err := api.Start(ctx, f.token, "budget", in); err != nil {
+			t.Fatal(err)
+		}
+		if instanceCount(f, `SELECT COUNT(*) FROM weave_pending_advances WHERE namespace_id=?`, f.resources.NamespaceID) != 1 {
+			t.Fatal("replay changed checkpoint")
+		}
+	})
+	t.Run("backpressure_rolls_back_complete_graph", func(t *testing.T) {
+		f := newExecutionFixture(t, ctx, db)
+		grantWorkflowInstance(t, f)
+		in := publishInstanceSpec(t, f, allInitialNodes(f.resources.QueueID))
+		f.exec(`UPDATE pulse_queues SET max_unfinished_tasks=2 WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.QueueID)
+		if _, err := api.Start(ctx, f.token, "capacity", in); fault.Code(err) != "QUEUE_FULL" {
+			t.Fatal(err)
+		}
+		for _, table := range []string{"weave_instances", "weave_steps", "weave_activations", "weave_waits", "weave_poll_states", "weave_poll_rounds", "pulse_tasks", "pulse_execution_gates"} {
+			if instanceCount(f, "SELECT COUNT(*) FROM "+table+" WHERE namespace_id=?", f.resources.NamespaceID) != 0 {
+				t.Fatal("partial graph", table)
+			}
+		}
+		assertApplicationCounts(t, f, 0, 0, 0)
+		if instanceCount(f, `SELECT COUNT(*) FROM orbit_idempotency_keys WHERE namespace_id=? AND operation='workflow.start'`, f.resources.NamespaceID) != 0 {
+			t.Fatal("failed start receipt committed")
+		}
+		f.exec(`UPDATE pulse_queues SET max_unfinished_tasks=3 WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.QueueID)
+		v, err := api.Start(ctx, f.token, "capacity", in)
+		if err != nil || v.Deduplicated {
+			t.Fatal(v, err)
+		}
+		dupe, err := api.Start(ctx, f.token, "capacity", in)
+		if err != nil || !dupe.Deduplicated {
+			t.Fatal("receipt blocked at capacity", err)
+		}
+		assertApplicationCounts(t, f, 3, 0, 0)
+	})
+	for _, table := range []string{"weave_steps", "weave_waits", "pulse_tasks", "orbit_audit_events"} {
+		t.Run("rollback_"+table, func(t *testing.T) {
+			f := newExecutionFixture(t, ctx, db)
+			grantWorkflowInstance(t, f)
+			in := publishInstanceSpec(t, f, allInitialNodes(f.resources.QueueID))
+			f.exec("CREATE TRIGGER test_reject_workflow_start BEFORE INSERT ON " + table + " FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='injected workflow admission failure'")
+			_, err := api.Start(ctx, f.token, "rollback", in)
+			f.exec(`DROP TRIGGER test_reject_workflow_start`)
+			if err == nil {
+				t.Fatal("injected write accepted")
+			}
+			if instanceCount(f, `SELECT COUNT(*) FROM weave_instances WHERE namespace_id=?`, f.resources.NamespaceID) != 0 || instanceCount(f, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=?`, f.resources.NamespaceID) != 0 {
+				t.Fatal("rollback left resources")
+			}
+			assertApplicationCounts(t, f, 0, 0, 0)
+			if _, err := api.Start(ctx, f.token, "rollback", in); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+	t.Run("current_authority_and_foreign_associations", func(t *testing.T) {
+		f := newExecutionFixture(t, ctx, db)
+		grantWorkflowInstance(t, f)
+		in := publishInstanceSpec(t, f, workflowTestSpec(f.resources.QueueID))
+		v, err := api.Start(ctx, f.token, "authorized", in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		grantWorkflowInstance(t, f, string(executionID(t)))
+		for _, run := range []func() error{func() error { _, e := api.Start(ctx, f.token, "authorized", in); return e }, func() error { _, e := api.Get(ctx, f.token, v.Instance.ID); return e }, func() error { _, e := api.Steps(ctx, f.token, v.Instance.ID); return e }} {
+			if fault.Code(run()) != "NOT_FOUND" {
+				t.Fatal("scope revoked but data disclosed")
+			}
+		}
+		grantWorkflowInstance(t, f)
+		in.Input = json.RawMessage(`{"n":"bad"}`)
+		if _, err := api.Start(ctx, f.token, "invalid", in); fault.Code(err) != "INVALID_ARGUMENT" {
+			t.Fatal(err)
+		}
+		other := newExecutionFixture(t, ctx, db)
+		grantWorkflowInstance(t, other)
+		if _, err := api.Get(ctx, other.token, v.Instance.ID); fault.Code(err) != "NOT_FOUND" {
+			t.Fatal(err)
+		}
+		if _, err := api.Start(ctx, other.token, "cross-namespace", in); fault.Code(err) != "NOT_FOUND" {
+			t.Fatal(err)
+		}
+	})
+	t.Run("query_rejects_broken_current_activation_and_policy", func(t *testing.T) {
+		f := newExecutionFixture(t, ctx, db)
+		grantWorkflowInstance(t, f)
+		in := publishInstanceSpec(t, f, workflowTestSpec(f.resources.QueueID))
+		v, err := api.Start(ctx, f.token, "query", in)
+		if err != nil {
+			t.Fatal(err)
+		}
+		steps, err := api.Steps(ctx, f.token, v.Instance.ID)
+		if err != nil {
+			t.Fatal(err)
+		}
+		step := steps.Steps[0]
+		f.exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, step.ID)
+		if _, err := api.Steps(ctx, f.token, v.Instance.ID); err == nil {
+			t.Fatal("missing activation accepted")
+		}
+		f.exec(`UPDATE weave_steps SET current_activation_id=? WHERE namespace_id=? AND id=?`, step.ActivationID, f.resources.NamespaceID, step.ID)
+		f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, step.TaskID)
+		if _, err := api.Steps(ctx, f.token, v.Instance.ID); err == nil {
+			t.Fatal("wrong task owner accepted")
+		}
+		f.exec(`UPDATE pulse_tasks SET owner_id=? WHERE namespace_id=? AND id=?`, step.ActivationID, f.resources.NamespaceID, step.TaskID)
+		f.exec(`UPDATE pulse_tasks SET queue_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, step.TaskID)
+		if _, err := api.Steps(ctx, f.token, v.Instance.ID); err == nil {
+			t.Fatal("wrong task queue accepted")
+		}
+		f.exec(`UPDATE pulse_tasks SET queue_id=? WHERE namespace_id=? AND id=?`, f.resources.QueueID, f.resources.NamespaceID, step.TaskID)
+		f.exec(`UPDATE weave_instances SET policy_snapshot='{}' WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.Instance.ID)
+		if _, err := api.Get(ctx, f.token, v.Instance.ID); err == nil {
+			t.Fatal("missing fixed policy accepted")
+		}
+	})
+}
--- a/internal/storage/mysql/workflow_wait_upgrade_integration_test.go
+++ b/internal/storage/mysql/workflow_wait_upgrade_integration_test.go
@@ -0,0 +1,106 @@
+package mysql
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	core "git.yuntuo.hk/k2safe/orbit/model"
+	driver "github.com/go-sql-driver/mysql"
+)
+
+func TestIntegrationWorkflowWaitUpgrade(t *testing.T) {
+	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
+	if dsn == "" {
+		t.Skip("requires disposable MySQL")
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	admin, err := Open(ctx, dsn, 2, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer admin.Close()
+	name := "orbit_workflow_wait_upgrade_" + string(executionID(t))
+	if _, err := admin.ExecContext(ctx, "CREATE DATABASE `"+name+"`"); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		cleanup, stop := context.WithTimeout(context.Background(), 10*time.Second)
+		defer stop()
+		if _, err := admin.ExecContext(cleanup, "DROP DATABASE `"+name+"`"); err != nil {
+			t.Error(err)
+		}
+	}()
+	cfg, err := driver.ParseDSN(dsn)
+	if err != nil {
+		t.Fatal("invalid test DSN")
+	}
+	cfg.DBName = name
+	db, err := Open(ctx, cfg.FormatDSN(), 4, 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	exec := func(q string, args ...any) {
+		t.Helper()
+		if _, err := db.ExecContext(ctx, q, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	exec(`CREATE TABLE orbit_schema_migrations (version BIGINT UNSIGNED PRIMARY KEY,name VARCHAR(191) NOT NULL,checksum CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,dirty BOOLEAN NOT NULL,applied_at DATETIME(6) NULL) ENGINE=InnoDB`)
+	migrations, err := Migrations()
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, m := range migrations {
+		if m.Version > 17 {
+			break
+		}
+		statements, err := splitSQL(m.SQL)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, q := range statements {
+			exec(q)
+		}
+		exec(`INSERT INTO orbit_schema_migrations(version,name,checksum,dirty,applied_at) VALUES (?,?,?,FALSE,UTC_TIMESTAMP(6))`, m.Version, m.Name, m.Checksum)
+	}
+	ns, workflow := executionID(t), executionID(t)
+	records := map[core.ID]string{}
+	for _, value := range []string{"关联é ", strings.Repeat("𠮷", 191)} {
+		id := executionID(t)
+		records[id] = value
+		exec(`INSERT INTO weave_waits(namespace_id,id,workflow_id,activation_id,kind,correlation_key,deadline_at) VALUES (?,?,?,?,'SIGNAL',?,DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR))`, ns, id, workflow, executionID(t), value)
+	}
+	exec(`INSERT INTO weave_instances(namespace_id,id,definition_id,definition_version_id,gate_id,input,policy_snapshot,business_key,deadline_at) VALUES (?,?,?,?,?,'{}','{}',?,DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR))`, ns, workflow, executionID(t), executionID(t), executionID(t), "关联é ")
+	if err := Migrate(ctx, db); err != nil {
+		t.Fatal(err)
+	}
+	if err := CheckSchema(ctx, db); err != nil {
+		t.Fatal(err)
+	}
+	for id, want := range records {
+		var got, state string
+		var name, version *string
+		if err := db.QueryRowContext(ctx, `SELECT correlation_key,state,signal_name,payload_version FROM weave_waits WHERE namespace_id=? AND id=?`, ns, id).Scan(&got, &state, &name, &version); err != nil {
+			t.Fatal(err)
+		}
+		if got != want || state != "WAITING" || name != nil || version != nil {
+			t.Fatal("legacy wait identity rewritten or truncated")
+		}
+	}
+	var key, collation string
+	if err := db.QueryRowContext(ctx, `SELECT business_key FROM weave_instances WHERE namespace_id=? AND id=?`, ns, workflow).Scan(&key); err != nil || key != "关联é " {
+		t.Fatal("business key changed", err)
+	}
+	if err := db.QueryRowContext(ctx, `SELECT COLLATION_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='weave_instances' AND COLUMN_NAME='business_key'`).Scan(&collation); err != nil || collation != "utf8mb4_0900_bin" {
+		t.Fatal("incorrect key collation", err)
+	}
+	if err := Migrate(ctx, db); err != nil {
+		t.Fatal(err)
+	}
+	t.Log("v17 wait identity, 764-byte legacy correlation and trailing spaces preserved; new signal metadata remains unverified NULL; schema has zero foreign keys")
+}