feat: add audited task review and linked manual retries

2026-09-17 21:57:53 +08:00
parent 571107f96d
commit a50d5faac7
43 changed files with 2096 additions and 57 deletions
--- a/internal/storage/mysql/application_resources_upgrade_integration_test.go
+++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go
@@ -64,6 +64,7 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) {
 	retryFinish(t, g, finished, "SUCCEEDED", false)
 	assertApplicationCounts(t, f, 4, 2, 1)
 	// 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。
+	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)`)
 	f.exec(`DROP TABLE orbit_application_resources`)
 	f.exec(`DROP TABLE orbit_result_events,orbit_event_streams`)
 	f.exec(`ALTER TABLE orbit_subscriptions DROP CHECK ck_subscription_window, DROP COLUMN queue_id, DROP COLUMN cursor_sequence, DROP COLUMN pending_count, DROP COLUMN max_pending, DROP COLUMN max_attempts`)
--- a/internal/storage/mysql/bootstrap.go
+++ b/internal/storage/mysql/bootstrap.go
@@ -14,6 +14,7 @@ import (

 // BootstrapInput 仅供持有数据库管理权限的初始化命令使用，不暴露为公共 HTTP。
 type BootstrapInput struct {
+	RecoveryAccess        bool    // RecoveryAccess 显式授予任务核查、人工重试及操作审计读取权限。
 	EventAccess           bool    // EventAccess 显式增加结果订阅查询、管理和消费权限。
 	Application           string  // Application 新应用名称，与 ExistingApplicationID 二选一。
 	Namespace             string  // Namespace 命名空间名称。
@@ -84,6 +85,9 @@ func Bootstrap(ctx context.Context, db *sql.DB, in BootstrapInput, deliver func(
 	if in.EventAccess {
 		permissions.Actions = append(permissions.Actions, "events:read", "events:manage", "events:consume")
 	}
+	if in.RecoveryAccess {
+		permissions.Actions = append(permissions.Actions, "tasks:review", "tasks:retry", "tasks:audit:read")
+	}
 	raw, err := json.Marshal(permissions)
 	if err != nil {
 		return err
--- a/internal/storage/mysql/migrations/000015_task_manual_recovery.up.sql
+++ b/internal/storage/mysql/migrations/000015_task_manual_recovery.up.sql
@@ -0,0 +1,7 @@
+-- 人工重试每个原任务只允许直接派生一个后继，后续重试沿新任务继续；不增加外键。
+-- fk_pulse_tasks_3 是历史外键已删除后留下的普通非唯一索引；以下唯一索引覆盖相同列。
+ALTER TABLE pulse_tasks
+    MODIFY COLUMN retry_of BINARY(16) NULL COMMENT '人工重试的原任务标识，同命名空间每个原任务最多一个直接后继；空表示非人工派生',
+    DROP INDEX fk_pulse_tasks_3,
+    ADD UNIQUE KEY uq_task_retry_parent(namespace_id,retry_of) COMMENT '防止不同操作身份并发重新发起同一原任务',
+    ADD CONSTRAINT ck_task_retry_not_self CHECK (retry_of IS NULL OR retry_of<>id);
--- a/internal/storage/mysql/no_foreign_keys_integration_test.go
+++ b/internal/storage/mysql/no_foreign_keys_integration_test.go
@@ -88,7 +88,8 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
 	}
 	indexes := func() []string {
 		t.Helper()
-		rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
+		// v15 将原 retry_of 非唯一索引替换为相同列的唯一索引，下方单独验证其定义。
+		rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -167,7 +168,28 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
 	if got := foreignKeys(); got != 0 {
 		t.Fatalf("remaining foreign keys=%d", got)
 	}
-	if !reflect.DeepEqual(beforeIndexes, indexes()) {
+	var retryIndex string
+	if err := db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(CONCAT(COLUMN_NAME,':',NON_UNIQUE) ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND INDEX_NAME='uq_task_retry_parent'`).Scan(&retryIndex); err != nil || retryIndex != "namespace_id:0,retry_of:0" {
+		t.Fatalf("retry uniqueness replacement=%q err=%v", retryIndex, err)
+	}
+	if afterIndexes := indexes(); !reflect.DeepEqual(beforeIndexes, afterIndexes) {
+		before, after := map[string]bool{}, map[string]bool{}
+		for _, index := range beforeIndexes {
+			before[index] = true
+		}
+		for _, index := range afterIndexes {
+			after[index] = true
+		}
+		for index := range before {
+			if !after[index] {
+				t.Errorf("upgrade removed index component: %s", index)
+			}
+		}
+		for index := range after {
+			if !before[index] {
+				t.Errorf("upgrade added unexpected index component: %s", index)
+			}
+		}
 		t.Fatal("upgrade changed index definitions")
 	}
 	if !reflect.DeepEqual(beforeTask, taskSnapshot()) {
--- a/internal/storage/mysql/task_history.go
+++ b/internal/storage/mysql/task_history.go
@@ -29,6 +29,14 @@ type historyCursor struct {
 }

 func decodeHistoryCursor(value string, ns, task, attempt core.ID) (historyCursor, error) {
+	kind := "attempts"
+	if attempt != "" {
+		kind = "evidence"
+	}
+	return decodeHistoryCursorKind(value, ns, task, attempt, kind)
+}
+
+func decodeHistoryCursorKind(value string, ns, task, attempt core.ID, kind string) (historyCursor, error) {
 	var c historyCursor
 	raw, err := base64.RawURLEncoding.Strict().DecodeString(value)
 	if err != nil || len(raw) > 768 {
@@ -53,12 +61,12 @@ func decodeHistoryCursor(value string, ns, task, attempt core.ID) (historyCursor
 	if d.Decode(&c) != nil || c.Version != 1 || c.Namespace != ns || c.Task != task || c.Attempt != attempt {
 		return c, fault.Invalid("cursor does not match this history query")
 	}
-	if attempt == "" {
+	if kind == "attempts" {
 		if c.Kind != "attempts" || c.Epoch == 0 || c.ID != "" || c.Created != nil {
 			return c, fault.Invalid("invalid attempts cursor")
 		}
 	} else {
-		if c.Kind != "evidence" || c.Epoch != 0 || c.ID.Validate() != nil || c.Created == nil || c.Created.Year() < 2000 || c.Created.Year() > 9998 || c.Created.Nanosecond()%1000 != 0 {
+		if c.Kind != kind || c.Epoch != 0 || c.ID.Validate() != nil || c.Created == nil || c.Created.Year() < 2000 || c.Created.Year() > 9998 || c.Created.Nanosecond()%1000 != 0 {
 			return c, fault.Invalid("invalid evidence cursor")
 		}
 	}
--- a/internal/storage/mysql/task_operations.go
+++ b/internal/storage/mysql/task_operations.go
@@ -0,0 +1,79 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+
+	"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"
+)
+
+const taskOperationColumns = `id,resource_id,operation_id,actor_id,action,created_at`
+
+func scanTaskOperation(row scanner) (service.TaskOperationSummary, error) {
+	var v service.TaskOperationSummary
+	err := row.Scan(&v.ID, &v.TaskID, &v.OperationID, &v.ActorID, &v.Action, &v.CreatedAt)
+	return v, err
+}
+func (s *TaskStore) ListTaskOperations(ctx context.Context, t auth.Token, id core.ID, in service.HistoryPageInput) (service.TaskOperationPage, error) {
+	out := service.TaskOperationPage{Items: []service.TaskOperationSummary{}}
+	err := s.transaction(ctx, t, "tasks:audit:read", func(tx *sql.Tx, p auth.Permissions) error {
+		if err := authorizeHistoryTask(ctx, tx, t.NamespaceID, id, p); err != nil {
+			return err
+		}
+		query := `SELECT ` + taskOperationColumns + ` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='task' AND resource_id=?`
+		args := []any{t.NamespaceID, id}
+		if in.Cursor != "" {
+			c, err := decodeHistoryCursorKind(in.Cursor, t.NamespaceID, id, "", "operations")
+			if err != nil {
+				return err
+			}
+			query += ` AND (created_at<? OR (created_at=? AND id<?))`
+			args = append(args, *c.Created, *c.Created, c.ID)
+		}
+		query += ` ORDER BY created_at DESC,id DESC LIMIT ?`
+		args = append(args, in.Limit+1)
+		rows, err := tx.QueryContext(ctx, query, args...)
+		if err != nil {
+			return err
+		}
+		defer rows.Close()
+		for rows.Next() {
+			v, err := scanTaskOperation(rows)
+			if err != nil {
+				return err
+			}
+			out.Items = append(out.Items, v)
+		}
+		if err := rows.Err(); err != nil {
+			return err
+		}
+		if len(out.Items) > in.Limit {
+			out.Items = out.Items[:in.Limit]
+			last := out.Items[len(out.Items)-1]
+			out.NextCursor, err = encodeHistoryCursor(historyCursor{Version: 1, Kind: "operations", Namespace: t.NamespaceID, Task: id, ID: last.ID, Created: &last.CreatedAt})
+		}
+		return err
+	})
+	return out, err
+}
+
+func (s *TaskStore) GetTaskOperation(ctx context.Context, t auth.Token, id, audit core.ID) (service.TaskOperationView, error) {
+	var out service.TaskOperationView
+	err := s.transaction(ctx, t, "tasks:audit:read", func(tx *sql.Tx, p auth.Permissions) error {
+		if err := authorizeHistoryTask(ctx, tx, t.NamespaceID, id, p); err != nil {
+			return err
+		}
+		var before, after []byte
+		err := tx.QueryRowContext(ctx, `SELECT `+taskOperationColumns+",reason,COALESCE(evidence_ref,''),`before`,`after` FROM orbit_audit_events WHERE namespace_id=? AND resource_kind='task' AND resource_id=? AND id=?", t.NamespaceID, id, audit).Scan(&out.ID, &out.TaskID, &out.OperationID, &out.ActorID, &out.Action, &out.CreatedAt, &out.Reason, &out.EvidenceRef, &before, &after)
+		if errors.Is(err, sql.ErrNoRows) {
+			return fault.NotFound
+		}
+		out.Before, out.After = before, after
+		return err
+	})
+	return out, err
+}
--- a/internal/storage/mysql/task_recovery.go
+++ b/internal/storage/mysql/task_recovery.go
@@ -0,0 +1,185 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+
+	"git.yuntuo.hk/k2safe/orbit/internal/fault"
+	core "git.yuntuo.hk/k2safe/orbit/model"
+	"git.yuntuo.hk/k2safe/orbit/platform/auth"
+	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
+	"git.yuntuo.hk/k2safe/orbit/pulse/service"
+)
+
+// recoveryTaskIdentity 认证后按应用额度→管理身份→队列→任务锁定。
+// 回执核对不重新执行核查/重试，仍检查当前凭据与队列范围。
+func recoveryTaskIdentity(ctx context.Context, tx *sql.Tx, t auth.Token, p auth.Permissions, id core.ID, operation, key string, in any) (executionTask, core.ID, []byte, error) {
+	v, err := readTask(ctx, tx, t.NamespaceID, id, false)
+	if err != nil {
+		return executionTask{}, "", nil, err
+	}
+	if !p.AllowsQueue(v.QueueID) {
+		return executionTask{}, "", nil, fault.NotFound
+	}
+	if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil {
+		return executionTask{}, "", nil, err
+	}
+	identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, operation, string(id), key, "task", id, in)
+	if err != nil || cached != nil {
+		return executionTask{}, identity, cached, err
+	}
+	if _, _, err := lockQueueAfterApplication(ctx, tx, t.NamespaceID, v.QueueID); err != nil {
+		return executionTask{}, "", nil, err
+	}
+	task, err := readExecutionTask(ctx, tx, t.NamespaceID, id)
+	return task, identity, nil, err
+}
+
+func allowTaskRecovery(ctx context.Context, tx *sql.Tx, ns, id core.ID, allowSchedule bool) error {
+	var owner sql.NullString
+	var ownerID, gate *core.ID
+	if err := tx.QueryRowContext(ctx, `SELECT owner_kind,owner_id,gate_id FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, id).Scan(&owner, &ownerID, &gate); err != nil {
+		return err
+	}
+	if gate != nil {
+		return fault.Conflict
+	}
+	if owner.Valid {
+		if !allowSchedule || owner.String != "schedule" || ownerID == nil {
+			return fault.Conflict
+		}
+		var matched int
+		if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_schedules s JOIN pulse_schedule_occurrences o ON o.namespace_id=s.namespace_id AND o.schedule_id=s.id WHERE s.namespace_id=? AND s.id=? AND o.task_id=?`, ns, *ownerID, id).Scan(&matched); err != nil {
+			return err
+		}
+		if matched != 1 {
+			return fault.Conflict
+		}
+	} else if ownerID != nil {
+		return fault.Conflict
+	}
+	var active int
+	if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_attempts WHERE namespace_id=? AND task_id=? AND state IN ('LEASED','RUNNING')`, ns, id).Scan(&active); err != nil {
+		return err
+	}
+	if active != 0 {
+		return fault.Conflict
+	}
+	if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_permits p JOIN pulse_attempts a ON a.namespace_id=p.namespace_id AND a.id=p.attempt_id WHERE a.namespace_id=? AND a.task_id=? AND p.released_at IS NULL`, ns, id).Scan(&active); err != nil {
+		return err
+	}
+	if active != 0 {
+		return fault.Conflict
+	}
+	return nil
+}
+
+func (s *TaskStore) ReviewTask(ctx context.Context, t auth.Token, id core.ID, key string, in service.ReviewInput) (service.ReviewResult, error) {
+	var out service.ReviewResult
+	err := s.transaction(ctx, t, "tasks:review", func(tx *sql.Tx, p auth.Permissions) error {
+		task, identity, cached, err := recoveryTaskIdentity(ctx, tx, t, p, id, "task.review", key, in)
+		if err != nil {
+			return err
+		}
+		if cached != nil {
+			if err := json.Unmarshal(cached, &out); err != nil {
+				return err
+			}
+			out.Deduplicated = true
+			return nil
+		}
+		v := task.view
+		if v.State != pulse.TaskNeedsReview || v.Revision != in.ExpectedRevision || task.lease != nil {
+			return fault.Conflict
+		}
+		if err := allowTaskRecovery(ctx, tx, t.NamespaceID, id, true); err != nil {
+			return err
+		}
+		now, err := dbNow(ctx, tx)
+		if err != nil {
+			return err
+		}
+		changed, err := tx.ExecContext(ctx, `UPDATE pulse_tasks SET state=?,result=?,effect_uncertain=FALSE,completed_at=?,updated_at=?,revision=revision+1 WHERE namespace_id=? AND id=? AND revision=? AND state='NEEDS_REVIEW'`, in.Outcome, string(in.Result), now, now, t.NamespaceID, id, in.ExpectedRevision)
+		if err := exactlyOne(changed, err); err != nil {
+			return err
+		}
+		if err := releaseQueueTask(ctx, tx, t.NamespaceID, v.QueueID); err != nil {
+			return err
+		}
+		out.Task, err = readTask(ctx, tx, t.NamespaceID, id, false)
+		if err != nil {
+			return err
+		}
+		digest, err := requestDigest(in.Result)
+		if err != nil {
+			return err
+		}
+		after := map[string]any{"state": in.Outcome, "revision": fmt.Sprint(out.Task.Revision), "effect_uncertain": false, "execution_settled": true, "result_digest": hex.EncodeToString(digest[:])}
+		out.AuditID, err = appendAuditWithEvidence(ctx, tx, t, identity, id, "task", "task.review", in.Reason, in.EvidenceRef, map[string]any{"state": v.State, "revision": fmt.Sprint(v.Revision), "effect_uncertain": v.EffectUncertain}, after)
+		if err != nil {
+			return err
+		}
+		if err := saveIdentity(ctx, tx, t.NamespaceID, identity, out); err != nil {
+			return err
+		}
+		// 结果流锁最后取得；冻结新的终态事件，历史NEEDS_REVIEW事件/尝试保持原样。
+		return emitEvent(ctx, tx, t.NamespaceID, id, out.Task.Revision, "task", resultEvent(in.Outcome), map[string]any{"task_id": id, "state": in.Outcome, "effect_uncertain": false, "decision": "MANUAL_REVIEW", "audit_id": out.AuditID})
+	})
+	return out, err
+}
+
+func (s *TaskStore) RetryTask(ctx context.Context, t auth.Token, id core.ID, key string, in service.ManualRetryInput) (service.ManualRetryResult, error) {
+	var out service.ManualRetryResult
+	err := s.transaction(ctx, t, "tasks:retry", func(tx *sql.Tx, p auth.Permissions) error {
+		task, identity, cached, err := recoveryTaskIdentity(ctx, tx, t, p, id, "task.retry", key, in)
+		if err != nil {
+			return err
+		}
+		if cached != nil {
+			if err := json.Unmarshal(cached, &out); err != nil {
+				return err
+			}
+			out.Deduplicated = true
+			return nil
+		}
+		v := task.view
+		if (v.State != pulse.TaskFailed && v.State != pulse.TaskCancelled) || v.EffectUncertain || v.Revision != in.ExpectedRevision || task.lease != nil {
+			return fault.Conflict
+		}
+		if err := allowTaskRecovery(ctx, tx, t.NamespaceID, id, false); err != nil {
+			return err
+		}
+		var children int
+		if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM pulse_tasks WHERE namespace_id=? AND retry_of=?`, t.NamespaceID, id).Scan(&children); err != nil {
+			return err
+		}
+		if children != 0 {
+			return fault.Conflict
+		}
+		input, err := service.NormalizeSubmit(service.SubmitInput{QueueID: v.QueueID, Handler: v.Handler, HandlerVersion: v.HandlerVersion, PayloadVersion: v.PayloadVersion, Payload: v.Payload, BusinessKey: v.BusinessKey, ConcurrencyKey: v.ConcurrencyKey, Priority: v.Priority, RecoverySafety: v.RecoverySafety, Retry: v.Retry, AvailableAt: in.AvailableAt, StartDeadlineAt: in.StartDeadlineAt, TaskTimeoutMS: in.TaskTimeoutMS})
+		if err != nil {
+			return err
+		}
+		newID, err := core.NewID()
+		if err != nil {
+			return err
+		}
+		out.SourceTaskID = id
+		out.Task, err = insertAcceptedTask(ctx, tx, t.NamespaceID, newID, input, &id)
+		if err != nil {
+			return err
+		}
+		out.AuditID, err = appendAuditWithEvidence(ctx, tx, t, identity, id, "task", "task.retry", in.Reason, in.EvidenceRef, map[string]any{"state": v.State, "revision": fmt.Sprint(v.Revision)}, map[string]any{"source_task_id": id, "retry_task_id": newID, "safe_to_retry": true, "task_timeout_ms": in.TaskTimeoutMS})
+		if err != nil {
+			return err
+		}
+		if err := emitEvent(ctx, tx, t.NamespaceID, newID, 1, "task", "task.accepted", map[string]any{"task_id": newID, "state": pulse.TaskWaiting, "retry_of": id, "audit_id": out.AuditID}); err != nil {
+			return err
+		}
+		return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
+	})
+	return out, err
+}
--- a/internal/storage/mysql/task_recovery_integration_test.go
+++ b/internal/storage/mysql/task_recovery_integration_test.go
@@ -0,0 +1,345 @@
+package mysql
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"reflect"
+	"sync"
+	"testing"
+	"time"
+
+	"git.yuntuo.hk/k2safe/orbit/internal/fault"
+	"git.yuntuo.hk/k2safe/orbit/platform/events"
+	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
+	"git.yuntuo.hk/k2safe/orbit/pulse/service"
+	"git.yuntuo.hk/k2safe/orbit/pulse/worker"
+)
+
+func grantTaskRecovery(f *executionFixture) {
+	f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.actions',JSON_ARRAY_APPEND(JSON_EXTRACT(permissions,'$.actions'),'$','tasks:review','$','tasks:retry','$','tasks:audit:read')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
+}
+func unknownTask(t *testing.T, f *executionFixture) (service.TaskView, worker.AttemptRef) {
+	t.Helper()
+	v := f.submit(1)
+	ref := retryStart(t, f)
+	retryFinish(t, f, ref, "UNKNOWN", false)
+	v = f.get(v.ID)
+	if v.State != pulse.TaskNeedsReview || !v.EffectUncertain {
+		t.Fatal("fixture did not reach uncertainty")
+	}
+	return v, ref
+}
+func reviewInput(v service.TaskView, outcome pulse.TaskState) service.ReviewInput {
+	return service.ReviewInput{ExpectedRevision: v.Revision, Reason: "operator verified execution and external result", EvidenceRef: "evidence:reconciliation-001", ExecutionSettled: true, Outcome: outcome, Result: json.RawMessage(`{"verified":true,"n":9007199254740993}`)}
+}
+func manualRetryInput(v service.TaskView) service.ManualRetryInput {
+	return service.ManualRetryInput{ExpectedRevision: v.Revision, Reason: "operator confirmed safe same-operation retry", EvidenceRef: "evidence:safe-retry-001", SafeToRetry: true, TaskTimeoutMS: 600000}
+}
+
+func TestIntegrationTaskRecovery(t *testing.T) {
+	dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
+	if dsn == "" {
+		t.Skip("requires disposable MySQL")
+	}
+	ctx, stop := context.WithTimeout(context.Background(), 90*time.Second)
+	defer stop()
+	db, err := Open(ctx, dsn, 32, 8)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if err := Migrate(ctx, db); err != nil {
+		t.Fatal(err)
+	}
+	fixture := func(t *testing.T) *executionFixture {
+		f := newExecutionFixture(t, ctx, db)
+		grantTaskRecovery(f)
+		return f
+	}
+	t.Run("review_preserves_attempts_and_publishes_new_authoritative_result", func(t *testing.T) {
+		f := newEventFixture(t, ctx, db)
+		grantTaskRecovery(f.executionFixture)
+		sub := f.create(events.CreateInput{})
+		v, ref := unknownTask(t, f.executionFixture)
+		before, err := f.tasks.GetTaskAttempt(ctx, f.token, v.ID, ref.AttemptID)
+		if err != nil {
+			t.Fatal(err)
+		}
+		input := reviewInput(v, pulse.TaskSucceeded)
+		out, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", input)
+		if err != nil || out.Task.State != pulse.TaskSucceeded || out.Task.EffectUncertain || out.Task.AttemptCount != v.AttemptCount || out.Task.Revision != v.Revision+1 || string(out.Task.Result) != string(input.Result) {
+			t.Fatal(out, err)
+		}
+		after, err := f.tasks.GetTaskAttempt(ctx, f.token, v.ID, ref.AttemptID)
+		if err != nil || !reflect.DeepEqual(before, after) {
+			t.Fatal("review rewrote worker history", err)
+		}
+		assertQueueCount(t, f.executionFixture, 0)
+		assertApplicationCounts(t, f.executionFixture, 0, 0, 0)
+		replay, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", input)
+		if err != nil || !replay.Deduplicated || replay.AuditID != out.AuditID || replay.Task.Revision != out.Task.Revision {
+			t.Fatal("review identity changed", err)
+		}
+		input.Outcome = pulse.TaskFailed
+		_, err = f.tasks.ReviewTask(ctx, f.token, v.ID, "review", input)
+		expectWorkerCode(t, err, "IDEMPOTENCY_CONFLICT")
+		audit, err := f.tasks.GetTaskOperation(ctx, f.token, v.ID, out.AuditID)
+		if err != nil || audit.EvidenceRef != "evidence:reconciliation-001" || audit.ActorID != string(f.token.CredentialID) || audit.Action != "task.review" {
+			t.Fatal(audit, err)
+		}
+		// 旧UNKNOWN结果仍作为原快照交付；核查终态成为下一条独立不可变事件。
+		g := f.grant(sub.ID)
+		if g.Event.Type != "task.needs_review" || !g.Event.Snapshot.EffectUncertain {
+			t.Fatal("old snapshot was rewritten", g.Event)
+		}
+		f.ack(sub.ID, g)
+		g = f.grant(sub.ID)
+		if g.Event.Type != "task.succeeded" || g.Event.Snapshot.EffectUncertain || g.Event.Snapshot.Revision != out.Task.Revision {
+			t.Fatal("review result not journaled", g.Event)
+		}
+		f.ack(sub.ID, g)
+		late := retryFinish(t, f.executionFixture, ref, "SUCCEEDED", false)
+		if late.Receipt.Disposition != "EVIDENCE_ONLY" || f.get(v.ID).Revision != out.Task.Revision {
+			t.Fatal("late worker result overwrote review")
+		}
+		if f.count("orbit_audit_events", "resource_id=? AND action='task.review'", v.ID) != 1 {
+			t.Fatal("duplicate review audit")
+		}
+	})
+	t.Run("review_authority_is_explicit_and_history_is_scoped", func(t *testing.T) {
+		f := newExecutionFixture(t, ctx, db)
+		v, _ := unknownTask(t, f)
+		_, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		expectWorkerCode(t, err, "PERMISSION_DENIED")
+		_, err = f.tasks.ListTaskOperations(ctx, f.token, v.ID, service.HistoryPageInput{})
+		expectWorkerCode(t, err, "PERMISSION_DENIED")
+		grantTaskRecovery(f)
+		out, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		if err != nil {
+			t.Fatal(err)
+		}
+		g := fixture(t)
+		_, err = g.tasks.GetTaskOperation(ctx, g.token, v.ID, out.AuditID)
+		expectWorkerCode(t, err, "NOT_FOUND")
+		other := f.submit(1)
+		_, err = f.tasks.GetTaskOperation(ctx, f.token, other.ID, out.AuditID)
+		expectWorkerCode(t, err, "NOT_FOUND")
+		page, err := f.tasks.ListTaskOperations(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 1})
+		if err != nil || len(page.Items) != 1 || page.NextCursor == "" {
+			t.Fatal(page, err)
+		}
+		next, err := f.tasks.ListTaskOperations(ctx, f.token, v.ID, service.HistoryPageInput{Limit: 1, Cursor: page.NextCursor})
+		if err != nil || len(next.Items) != 1 || next.Items[0].ID == page.Items[0].ID {
+			t.Fatal(next, err)
+		}
+		_, err = f.tasks.ListTaskOperations(ctx, f.token, other.ID, service.HistoryPageInput{Cursor: page.NextCursor})
+		expectWorkerCode(t, err, "INVALID_ARGUMENT")
+		_, err = f.tasks.ListTaskAttempts(ctx, f.token, v.ID, service.HistoryPageInput{Cursor: page.NextCursor})
+		expectWorkerCode(t, err, "INVALID_ARGUMENT")
+		f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(g.resources.QueueID), f.resources.NamespaceID, f.resources.CredentialID)
+		_, err = f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		expectWorkerCode(t, err, "NOT_FOUND")
+	})
+	t.Run("review_cas_competition_and_atomic_audit_failure", func(t *testing.T) {
+		f := fixture(t)
+		v, _ := unknownTask(t, f)
+		input := reviewInput(v, pulse.TaskFailed)
+		trigger := "review_audit_" + string(executionID(t))
+		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled review audit failure'")
+		_, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", input)
+		f.exec("DROP TRIGGER " + trigger)
+		if err == nil || !reflect.DeepEqual(v, f.get(v.ID)) {
+			t.Fatal("failed audit did not roll back review", err)
+		}
+		assertQueueCount(t, f, 1)
+		assertApplicationCounts(t, f, 1, 0, 0)
+		if f.count("orbit_idempotency_keys", "operation='task.review'") != 0 || f.count("orbit_result_events", "1=1") != 1 {
+			t.Fatal("review leaked identity or result event")
+		}
+		keys := []string{"review", "competing-review"}
+		ch := make(chan error, 2)
+		var wg sync.WaitGroup
+		for _, key := range keys {
+			wg.Add(1)
+			go func() { defer wg.Done(); _, err := f.tasks.ReviewTask(ctx, f.token, v.ID, key, input); ch <- err }()
+		}
+		wg.Wait()
+		close(ch)
+		success, conflicts := 0, 0
+		for err := range ch {
+			if err == nil {
+				success++
+			} else if fault.Code(err) == "STATE_CONFLICT" {
+				conflicts++
+			} else {
+				t.Fatal(err)
+			}
+		}
+		if success != 1 || conflicts != 1 {
+			t.Fatal(success, conflicts)
+		}
+		assertQueueCount(t, f, 0)
+	})
+	t.Run("retry_new_identity_preserves_source_and_respects_backpressure", func(t *testing.T) {
+		f := fixture(t)
+		configureResources(t, f, 10, 4, 1, false)
+		v, oldRef := unknownTask(t, f)
+		_, err := f.tasks.RetryTask(ctx, f.token, v.ID, "retry", manualRetryInput(v))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		review, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		if err != nil {
+			t.Fatal(err)
+		}
+		v = review.Task
+		occupant := f.submit(1)
+		in := manualRetryInput(v)
+		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "retry", in)
+		expectWorkerCode(t, err, "QUEUE_FULL")
+		if f.count("pulse_tasks", "retry_of=?", v.ID) != 0 || f.count("orbit_idempotency_keys", "operation='task.retry'") != 0 {
+			t.Fatal("quota rejection left a child")
+		}
+		if _, err := f.tasks.CancelTask(ctx, f.token, occupant.ID, "free", service.CancelInput{ExpectedRevision: occupant.Revision, Reason: "free retry quota"}); err != nil {
+			t.Fatal(err)
+		}
+		out, err := f.tasks.RetryTask(ctx, f.token, v.ID, "retry", in)
+		if err != nil || out.Task.ID == v.ID || out.Task.RetryOf == nil || *out.Task.RetryOf != v.ID || out.Task.AttemptCount != 0 || out.Task.Revision != 1 || out.Task.Retry != v.Retry || string(out.Task.Payload) != string(v.Payload) || !reflect.DeepEqual(v, f.get(v.ID)) {
+			t.Fatal(out, err)
+		}
+		if !out.Task.DeadlineAt.After(out.Task.CreatedAt) || out.Task.DeadlineAt.Sub(out.Task.CreatedAt) != 10*time.Minute {
+			t.Fatal("new budget incorrect")
+		}
+		replay, err := f.tasks.RetryTask(ctx, f.token, v.ID, "retry", in)
+		if err != nil || !replay.Deduplicated || replay.Task.ID != out.Task.ID || replay.AuditID != out.AuditID {
+			t.Fatal("retry replay changed identity", err)
+		}
+		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "another", in)
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		late := retryFinish(t, f, oldRef, "SUCCEEDED", false)
+		if late.Receipt.Disposition != "EVIDENCE_ONLY" || !reflect.DeepEqual(v, f.get(v.ID)) {
+			t.Fatal("old worker changed resolved task")
+		}
+		assignment, ref := f.claim()
+		if assignment.Task.ID != out.Task.ID || assignment.Task.RetryOf == nil || *assignment.Task.RetryOf != v.ID {
+			t.Fatal("worker lost retry lineage")
+		}
+		if _, err := f.a.Start(ctx, f.token, f.binding, executionID(t), ref); err != nil {
+			t.Fatal(err)
+		}
+		retryFinish(t, f, ref, "SUCCEEDED", false)
+		assertQueueCount(t, f, 0)
+	})
+	t.Run("retry_competing_keys_and_audit_rollback", func(t *testing.T) {
+		f := fixture(t)
+		v := f.submit(1)
+		cancelled, err := f.tasks.CancelTask(ctx, f.token, v.ID, "cancel", service.CancelInput{ExpectedRevision: v.Revision, Reason: "stop original"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		v = cancelled.Task
+		trigger := "retry_audit_" + string(executionID(t))
+		f.exec("CREATE TRIGGER " + trigger + " BEFORE INSERT ON orbit_audit_events FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled retry audit failure'")
+		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "retry", manualRetryInput(v))
+		f.exec("DROP TRIGGER " + trigger)
+		if err == nil || f.count("pulse_tasks", "retry_of=?", v.ID) != 0 {
+			t.Fatal("retry audit failure leaked child", err)
+		}
+		assertQueueCount(t, f, 0)
+		ch := make(chan error, 8)
+		var wg sync.WaitGroup
+		for i := 0; i < 8; i++ {
+			key := string(executionID(t))
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				_, err := f.tasks.RetryTask(ctx, f.token, v.ID, key, manualRetryInput(v))
+				ch <- err
+			}()
+		}
+		wg.Wait()
+		close(ch)
+		success := 0
+		for err := range ch {
+			if err == nil {
+				success++
+			} else if fault.Code(err) != "STATE_CONFLICT" {
+				t.Fatal(err)
+			}
+		}
+		if success != 1 || f.count("pulse_tasks", "retry_of=?", v.ID) != 1 || f.count("orbit_audit_events", "action='task.retry'") != 1 {
+			t.Fatal("multiple manual children", success)
+		}
+		assertApplicationCounts(t, f, 1, 0, 0)
+	})
+	t.Run("review_unblocks_serial_schedule_without_manual_child", func(t *testing.T) {
+		f := newScheduleFixture(t, newExecutionFixture(t, ctx, db))
+		grantTaskRecovery(f.executionFixture)
+		in := f.definition()
+		in.Policy.Overlap = "SERIAL"
+		plan := f.create(in)
+		f.due(plan, 2*time.Second)
+		f.tick()
+		items := f.occurrences(plan.ID)
+		if len(items) != 1 || items[0].TaskID == nil {
+			t.Fatal("missing scheduled task")
+		}
+		ref := retryStart(t, f.executionFixture)
+		retryFinish(t, f.executionFixture, ref, "UNKNOWN", false)
+		v := f.get(*items[0].TaskID)
+		f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, plan.ID)
+		f.tick()
+		if len(f.occurrences(plan.ID)) != 1 {
+			t.Fatal("unknown execution did not block serial schedule")
+		}
+		out, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		if err != nil {
+			t.Fatal(err)
+		}
+		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "retry", manualRetryInput(out.Task))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, plan.ID)
+		f.tick()
+		if len(f.occurrences(plan.ID)) != 2 {
+			t.Fatal("review did not release serial schedule")
+		}
+	})
+	t.Run("recovery_rejects_gates_bad_owner_active_and_success", func(t *testing.T) {
+		f := fixture(t)
+		waiting := f.submit(1)
+		_, err := f.tasks.ReviewTask(ctx, f.token, waiting.ID, "waiting", reviewInput(waiting, pulse.TaskFailed))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		ref := retryStart(t, f)
+		active := f.get(waiting.ID)
+		_, err = f.tasks.ReviewTask(ctx, f.token, active.ID, "active", reviewInput(active, pulse.TaskFailed))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		retryFinish(t, f, ref, "SUCCEEDED", false)
+		success := f.get(waiting.ID)
+		_, err = f.tasks.RetryTask(ctx, f.token, success.ID, "retry", manualRetryInput(success))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		v, _ := unknownTask(t, f)
+		f.exec(`UPDATE pulse_tasks SET gate_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, v.ID)
+		_, err = f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		f.exec(`UPDATE pulse_tasks SET gate_id=NULL,owner_kind='schedule',owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, v.ID)
+		_, err = f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		if f.get(v.ID).Revision != v.Revision {
+			t.Fatal("rejected recovery changed task")
+		}
+	})
+	t.Run("managed_tasks_cannot_bypass_recovery_owner", func(t *testing.T) {
+		f := fixture(t)
+		v, _ := unknownTask(t, f)
+		f.exec(`UPDATE pulse_tasks SET owner_kind='workflow',owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, v.ID)
+		_, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+		f.exec(`UPDATE pulse_tasks SET owner_kind=NULL,owner_id=NULL WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
+		out, err := f.tasks.ReviewTask(ctx, f.token, v.ID, "review", reviewInput(v, pulse.TaskFailed))
+		if err != nil {
+			t.Fatal(err)
+		}
+		f.exec(`UPDATE pulse_tasks SET owner_kind='schedule',owner_id=? WHERE namespace_id=? AND id=?`, executionID(t), f.resources.NamespaceID, v.ID)
+		_, err = f.tasks.RetryTask(ctx, f.token, v.ID, "retry", manualRetryInput(out.Task))
+		expectWorkerCode(t, err, "STATE_CONFLICT")
+	})
+}
--- a/internal/storage/mysql/task_store.go
+++ b/internal/storage/mysql/task_store.go
@@ -137,20 +137,25 @@ func emitEvent(ctx context.Context, tx *sql.Tx, ns, resource core.ID, revision u
 	return err
 }
 func appendAudit(ctx context.Context, tx *sql.Tx, token auth.Token, operation, resource core.ID, kind, action, reason string, before, after any) error {
+	_, err := appendAuditWithEvidence(ctx, tx, token, operation, resource, kind, action, reason, "", before, after)
+	return err
+}
+
+func appendAuditWithEvidence(ctx context.Context, tx *sql.Tx, token auth.Token, operation, resource core.ID, kind, action, reason, evidence string, before, after any) (core.ID, error) {
 	id, err := core.NewID()
 	if err != nil {
-		return err
+		return "", err
 	}
 	old, err := json.Marshal(before)
 	if err != nil {
-		return err
+		return "", err
 	}
 	current, err := json.Marshal(after)
 	if err != nil {
-		return err
+		return "", err
 	}
-	_, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,`before`,`after`) VALUES (?,?,?,?,?,?,?,?,?,?)", token.NamespaceID, id, operation, string(token.CredentialID), action, kind, resource, reason, string(old), string(current))
-	return err
+	_, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,evidence_ref,`before`,`after`) VALUES (?,?,?,?,?,?,?,?,NULLIF(?,''),?,?)", token.NamespaceID, id, operation, string(token.CredentialID), action, kind, resource, reason, evidence, string(old), string(current))
+	return id, err
 }
 func dbNow(ctx context.Context, tx *sql.Tx) (time.Time, error) {
 	var now time.Time
@@ -188,34 +193,7 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in
 		if _, _, err := lockQueueAfterApplication(ctx, tx, token.NamespaceID, in.QueueID); err != nil {
 			return err
 		}
-		now, err := dbNow(ctx, tx)
-		if err != nil {
-			return err
-		}
-		available := now
-		if in.AvailableAt != nil {
-			available = *in.AvailableAt
-		}
-		deadline := now.Add(time.Duration(in.TaskTimeoutMS) * time.Millisecond)
-		if available.Before(now.Add(-time.Second)) || !available.Before(deadline) {
-			return fault.Invalid("available_at must be current/future and before the task deadline")
-		}
-		if in.StartDeadlineAt != nil && (!in.StartDeadlineAt.After(now) || in.StartDeadlineAt.Before(available) || in.StartDeadlineAt.After(deadline)) {
-			return fault.Invalid("start_deadline_at must be after now, at or after available_at, and no later than the task deadline")
-		}
-		policy, err := json.Marshal(in.Retry)
-		if err != nil {
-			return err
-		}
-		if err := reserveQueueTask(ctx, tx, token.NamespaceID, in.QueueID); err != nil {
-			return err
-		}
-		_, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,concurrency_key,concurrency_key_digest,priority,available_at,start_deadline_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at)
-            VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),?,?,?,?,?,?,?,?,?,?)`, token.NamespaceID, taskID, in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, available, in.StartDeadlineAt, deadline, in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now)
-		if err != nil {
-			return err
-		}
-		out.Task, err = readTask(ctx, tx, token.NamespaceID, taskID, false)
+		out.Task, err = insertAcceptedTask(ctx, tx, token.NamespaceID, taskID, in, nil)
 		if err != nil {
 			return err
 		}
@@ -230,14 +208,46 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in
 	return out, err
 }

-const taskColumns = `id,namespace_id,queue_id,state,revision,handler,handler_version,payload_version,payload,result,COALESCE(business_key,''),COALESCE(concurrency_key,''),priority,available_at,start_deadline_at,deadline_at,recovery_safety,policy_snapshot,attempt_count,cancel_requested_at,completed_at,created_at,updated_at,effect_uncertain`
+// insertAcceptedTask 要求调用方已持有应用额度和队列锁，并将身份、事件及审计同事务提交。
+func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in service.SubmitInput, retryOf *core.ID) (service.TaskView, error) {
+	now, err := dbNow(ctx, tx)
+	if err != nil {
+		return service.TaskView{}, err
+	}
+	available := now
+	if in.AvailableAt != nil {
+		available = *in.AvailableAt
+	}
+	deadline := now.Add(time.Duration(in.TaskTimeoutMS) * time.Millisecond)
+	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")
+	}
+	if in.StartDeadlineAt != nil && (!in.StartDeadlineAt.After(now) || in.StartDeadlineAt.Before(available) || in.StartDeadlineAt.After(deadline)) {
+		return service.TaskView{}, fault.Invalid("start_deadline_at must be after now, at or after available_at, and no later than the task deadline")
+	}
+	policy, err := json.Marshal(in.Retry)
+	if err != nil {
+		return service.TaskView{}, err
+	}
+	if err := reserveQueueTask(ctx, tx, ns, in.QueueID); err != nil {
+		return service.TaskView{}, err
+	}
+	_, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,concurrency_key,concurrency_key_digest,priority,available_at,start_deadline_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at,retry_of)
+            VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),?,?,?,?,?,?,?,?,?,?,?)`, ns, taskID, in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, available, in.StartDeadlineAt, deadline, in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now, retryOf)
+	if err != nil {
+		return service.TaskView{}, err
+	}
+	return readTask(ctx, tx, ns, taskID, false)
+}
+
+const taskColumns = `id,namespace_id,queue_id,state,revision,handler,handler_version,payload_version,payload,result,COALESCE(business_key,''),COALESCE(concurrency_key,''),priority,available_at,start_deadline_at,deadline_at,recovery_safety,policy_snapshot,attempt_count,cancel_requested_at,completed_at,created_at,updated_at,effect_uncertain,retry_of`

 type scanner interface{ Scan(...any) error }

 func scanTask(row scanner) (service.TaskView, error) {
 	var v service.TaskView
 	var policy, payload, result []byte
-	err := row.Scan(&v.ID, &v.NamespaceID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.PayloadVersion, &payload, &result, &v.BusinessKey, &v.ConcurrencyKey, &v.Priority, &v.AvailableAt, &v.StartDeadlineAt, &v.DeadlineAt, &v.RecoverySafety, &policy, &v.AttemptCount, &v.CancelRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain)
+	err := row.Scan(&v.ID, &v.NamespaceID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.PayloadVersion, &payload, &result, &v.BusinessKey, &v.ConcurrencyKey, &v.Priority, &v.AvailableAt, &v.StartDeadlineAt, &v.DeadlineAt, &v.RecoverySafety, &policy, &v.AttemptCount, &v.CancelRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain, &v.RetryOf)
 	if err != nil {
 		return v, err
 	}