HH-768: implement phase A offline workflow (#10)

2026-08-28 11:20:00 +08:00
parent 14c8ce691f
commit 829521f5bc
12 changed files with 1432 additions and 26 deletions
@@ -9,6 +9,7 @@ FROM golang:1.26-alpine@sha256:28d89ee9cc0ff9fec75c82ca201e6bf7fdf9a679d4b7b24df
 WORKDIR /src
 COPY go.mod go.sum ./
 COPY cmd/ ./cmd/
+COPY internal/ ./internal/
 RUN CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/control-plane ./cmd/control-plane \
    && CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/docker-gateway ./cmd/docker-gateway

@@ -4,14 +4,14 @@

 ## 当前阶段

-项目处于阶段 A 技术验证。首个可运行切片提供浏览器容器的创建、列表、启停和回收，功能参考公开上游项目独立实现，不直接复制其代码。
+项目处于阶段 A 技术验证。当前可运行切片提供浏览器容器控制，以及单用户、Mock 连接器的离线业务闭环，功能参考公开上游项目独立实现，不直接复制其代码。

 详细的五领域优先级、系统边界、合规知识库与分阶段验证方案见
 [《合规产品规划与上游调研》](docs/product/compliance-product-plan.md)。

 ## 本地运行

-需要 Docker Compose。控制面仅绑定本机；浏览器网络在阶段 A 默认禁止外网访问：
+需要 Docker Compose。控制面仅绑定本机；PostgreSQL 仅在内部 control 网络可见，浏览器网络在阶段 A 默认禁止外网访问：

 ```bash
 docker pull git.ipao.vip/rogee/fingerprint-chromium@sha256:b9f23b8e3ac640174db0dfa49e9095fe7eb06f5db55a4e7550d979b35ff3a1b7
@@ -15,6 +15,7 @@ import (
 	"syscall"
 	"time"

+	"git.ipao.vip/rogee/creator-hub/internal/phasea"
 	"github.com/gofiber/fiber/v3"
 	"github.com/gofiber/fiber/v3/middleware/adaptor"
 	"github.com/sirupsen/logrus"
@@ -23,22 +24,27 @@ import (
 )

 type config struct {
-	listenAddr string
-	gateway    *url.URL
-	webDir     string
-	logLevel   logrus.Level
+	listenAddr  string
+	gateway     *url.URL
+	webDir      string
+	databaseURL string
+	logLevel    logrus.Level
 }

 func main() {
 	logrus.SetFormatter(&logrus.JSONFormatter{})
-	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
-	defer stop()
-	if err := newCommand().ExecuteContext(ctx); err != nil {
+	if err := execute(newCommand()); err != nil {
 		logrus.WithField("service", "control-plane").WithError(err).Error("service stopped")
 		os.Exit(1)
 	}
 }

+func execute(command *cobra.Command) error {
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	return command.ExecuteContext(ctx)
+}
+
 func newCommand() *cobra.Command {
 	command := &cobra.Command{
 		Use:           "control-plane",
@@ -52,12 +58,13 @@ func newCommand() *cobra.Command {
 				return err
 			}
 			logrus.SetLevel(cfg.logLevel)
-			logrus.WithFields(logrus.Fields{
-				"service":        "control-plane",
-				"listen_addr":    cfg.listenAddr,
-				"docker_gateway": cfg.gateway.Scheme + "://" + cfg.gateway.Host,
-			}).Info("service starting")
-			return newHandler(cfg.gateway, cfg.webDir).Listen(cfg.listenAddr, fiber.ListenConfig{
+			store, err := phasea.Open(command.Context(), cfg.databaseURL)
+			if err != nil {
+				return err
+			}
+			defer store.Close()
+			logStartup(cfg)
+			return newHandlerWithPhaseA(cfg.gateway, cfg.webDir, store).Listen(cfg.listenAddr, fiber.ListenConfig{
 				GracefulContext:       command.Context(),
 				DisableStartupMessage: true,
 			})
@@ -66,15 +73,25 @@ func newCommand() *cobra.Command {
 	return command
 }

+func logStartup(cfg config) {
+	logrus.WithFields(logrus.Fields{
+		"service":        "control-plane",
+		"listen_addr":    cfg.listenAddr,
+		"docker_gateway": cfg.gateway.Scheme + "://" + cfg.gateway.Host,
+	}).Info("service starting")
+}
+
 func loadConfig() (config, error) {
 	v := viper.New()
 	v.SetDefault("listen_addr", ":8080")
 	v.SetDefault("docker_gateway_url", "http://127.0.0.1:8081")
 	v.SetDefault("web_dir", "web/dist")
+	v.SetDefault("database_url", "postgres://creatorhub@127.0.0.1:5432/creatorhub?sslmode=disable")
 	v.SetDefault("log_level", "info")
 	_ = v.BindEnv("listen_addr", "LISTEN_ADDR")
 	_ = v.BindEnv("docker_gateway_url", "DOCKER_GATEWAY_URL")
 	_ = v.BindEnv("web_dir", "WEB_DIR")
+	_ = v.BindEnv("database_url", "DATABASE_URL")
 	_ = v.BindEnv("log_level", "LOG_LEVEL")

 	gateway, err := url.Parse(v.GetString("docker_gateway_url"))
@@ -86,10 +103,11 @@ func loadConfig() (config, error) {
 		return config{}, errors.New("LOG_LEVEL must be panic, fatal, error, warn, info, debug, or trace")
 	}
 	cfg := config{
-		listenAddr: strings.TrimSpace(v.GetString("listen_addr")),
-		gateway:    gateway,
-		webDir:     strings.TrimSpace(v.GetString("web_dir")),
-		logLevel:   level,
+		listenAddr:  strings.TrimSpace(v.GetString("listen_addr")),
+		gateway:     gateway,
+		webDir:      strings.TrimSpace(v.GetString("web_dir")),
+		databaseURL: strings.TrimSpace(v.GetString("database_url")),
+		logLevel:    level,
 	}
 	if cfg.listenAddr == "" {
 		return config{}, errors.New("LISTEN_ADDR must not be empty")
@@ -100,6 +118,11 @@ func loadConfig() (config, error) {
 	if cfg.webDir == "" {
 		return config{}, errors.New("WEB_DIR must not be empty")
 	}
+	databaseURL, err := url.Parse(cfg.databaseURL)
+	if err != nil || databaseURL.Host == "" || databaseURL.Path == "" || databaseURL.Path == "/" ||
+		(databaseURL.Scheme != "postgres" && databaseURL.Scheme != "postgresql") {
+		return config{}, errors.New("DATABASE_URL must be a postgres URL with a host")
+	}
 	return cfg, nil
 }

@@ -116,6 +139,10 @@ func validateListenAddr(addr string) error {
 }

 func newHandler(gateway *url.URL, webDirectory string) *fiber.App {
+	return newHandlerWithPhaseA(gateway, webDirectory, nil)
+}
+
+func newHandlerWithPhaseA(gateway *url.URL, webDirectory string, store *phasea.Store) *fiber.App {
 	proxy := httputil.NewSingleHostReverseProxy(gateway)
 	originalDirector := proxy.Director
 	proxy.Director = func(request *http.Request) {
@@ -127,6 +154,7 @@ func newHandler(gateway *url.URL, webDirectory string) *fiber.App {

 	app := fiber.New(fiber.Config{
 		AppName:     "CreatorHub control plane",
+		BodyLimit:   1 << 20,
 		ReadTimeout: 5 * time.Second,
 		IdleTimeout: 60 * time.Second,
 	})
@@ -136,6 +164,9 @@ func newHandler(gateway *url.URL, webDirectory string) *fiber.App {
 	})
 	app.All("/api/browsers", proxyHandler)
 	app.All("/api/browsers/*", proxyHandler)
+	if store != nil {
+		registerPhaseA(app, store)
+	}
 	app.Get("/*", spaHandler(webDirectory))
 	return app
 }
@@ -1,13 +1,26 @@
 package main

 import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
 	"io"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"os"
+	"os/exec"
 	"path/filepath"
+	"strings"
+	"syscall"
 	"testing"
+	"time"
+
+	"github.com/gofiber/fiber/v3"
+	"github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
 )

 func TestAPIProxyPreservesMethodPathAndStatus(t *testing.T) {
@@ -45,6 +58,122 @@ func TestLoadConfigRejectsInvalidGateway(t *testing.T) {
 	}
 }

+func TestLoadConfigRejectsInvalidDatabase(t *testing.T) {
+	t.Setenv("DATABASE_URL", "file:///tmp/creatorhub.db")
+	if _, err := loadConfig(); err == nil {
+		t.Fatal("expected non-Postgres database URL to be rejected")
+	}
+}
+
+func TestExecuteContextStopsOnSIGTERM(t *testing.T) {
+	if os.Getenv("CREATORHUB_SIGTERM_HELPER") == "1" {
+		command := &cobra.Command{Use: "shutdown-test", RunE: func(command *cobra.Command, _ []string) error {
+			_, _ = os.Stdout.WriteString("ready\n")
+			<-command.Context().Done()
+			_, _ = os.Stdout.WriteString("stopped\n")
+			return nil
+		}}
+		if err := execute(command); err != nil {
+			t.Fatal(err)
+		}
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	process := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestExecuteContextStopsOnSIGTERM$")
+	process.Env = append(os.Environ(), "CREATORHUB_SIGTERM_HELPER=1")
+	stdout, err := process.StdoutPipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := process.Start(); err != nil {
+		t.Fatal(err)
+	}
+	scanner := bufio.NewScanner(stdout)
+	if !scanner.Scan() || scanner.Text() != "ready" {
+		t.Fatalf("helper did not become ready: %q err=%v", scanner.Text(), scanner.Err())
+	}
+	if err := process.Process.Signal(syscall.SIGTERM); err != nil {
+		t.Fatal(err)
+	}
+	if !scanner.Scan() || scanner.Text() != "stopped" {
+		t.Fatalf("helper did not finish cleanup: %q err=%v", scanner.Text(), scanner.Err())
+	}
+	if err := process.Wait(); err != nil || ctx.Err() != nil {
+		t.Fatalf("process did not exit gracefully after SIGTERM: wait=%v context=%v", err, ctx.Err())
+	}
+}
+
+func TestStartupLogRedactsGatewayURL(t *testing.T) {
+	gateway, err := url.Parse("https://internal-user:internal-pass@docker-gateway:8443/v1?access_token=internal-token")
+	if err != nil {
+		t.Fatal(err)
+	}
+	logger := logrus.StandardLogger()
+	previousOutput, previousFormatter, previousLevel := logger.Out, logger.Formatter, logger.Level
+	t.Cleanup(func() {
+		logrus.SetOutput(previousOutput)
+		logrus.SetFormatter(previousFormatter)
+		logrus.SetLevel(previousLevel)
+	})
+	var output bytes.Buffer
+	logrus.SetOutput(&output)
+	logrus.SetFormatter(&logrus.JSONFormatter{})
+	logrus.SetLevel(logrus.InfoLevel)
+
+	logStartup(config{listenAddr: ":8080", gateway: gateway})
+	var entry map[string]any
+	if err := json.Unmarshal(output.Bytes(), &entry); err != nil {
+		t.Fatal(err)
+	}
+	if entry["docker_gateway"] != "https://docker-gateway:8443" {
+		t.Fatalf("unexpected redacted gateway: %#v", entry["docker_gateway"])
+	}
+	for _, sensitive := range []string{"internal-user", "internal-pass", "access_token", "internal-token"} {
+		if strings.Contains(output.String(), sensitive) {
+			t.Fatalf("startup log leaked %q: %s", sensitive, output.String())
+		}
+	}
+}
+
+func TestPhaseARequestRejectsUnknownFields(t *testing.T) {
+	app := fiber.New()
+	app.Post("/", func(c fiber.Ctx) error {
+		var input struct {
+			ID string `json:"id"`
+		}
+		if err := decodePhaseA(c, &input); err != nil {
+			return phaseAError(c, err)
+		}
+		return c.SendStatus(http.StatusNoContent)
+	})
+	response, err := app.Test(httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"id":"account-a","password":"must-not-be-accepted"}`)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusBadRequest {
+		t.Fatalf("expected unknown sensitive field to be rejected, got %d", response.StatusCode)
+	}
+}
+
+func TestPhaseAErrorRedactsInternalDetails(t *testing.T) {
+	app := fiber.New()
+	app.Get("/", func(c fiber.Ctx) error {
+		return phaseAError(c, errors.New("database exposed a secret value"))
+	})
+	response, err := app.Test(httptest.NewRequest(http.MethodGet, "/", nil))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer response.Body.Close()
+	body, _ := io.ReadAll(response.Body)
+	if response.StatusCode != http.StatusInternalServerError || strings.Contains(string(body), "secret") || !strings.Contains(string(body), "phase A operation failed") {
+		t.Fatalf("internal error was not redacted: status=%d body=%s", response.StatusCode, body)
+	}
+}
+
 func TestSPAServesFileAndIndexFallback(t *testing.T) {
 	directory := t.TempDir()
 	files := map[string]string{
@@ -0,0 +1,157 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"io"
+
+	"git.ipao.vip/rogee/creator-hub/internal/phasea"
+	"github.com/gofiber/fiber/v3"
+)
+
+type accountRequest struct {
+	ID                  string `json:"id"`
+	ProfileID           string `json:"profile_id"`
+	CredentialReference struct {
+		ID       string `json:"id"`
+		Provider string `json:"provider"`
+		Key      string `json:"key"`
+	} `json:"credential_reference"`
+}
+
+func registerPhaseA(app *fiber.App, store *phasea.Store) {
+	app.Post("/api/phase-a/accounts", func(c fiber.Ctx) error {
+		var input accountRequest
+		if err := decodePhaseA(c, &input); err != nil {
+			return phaseAError(c, err)
+		}
+		err := store.CreateAccount(c.Context(), phasea.Account{
+			ID: input.ID, ProfileID: input.ProfileID, CredentialReferenceID: input.CredentialReference.ID,
+			CredentialProvider: input.CredentialReference.Provider, CredentialKey: input.CredentialReference.Key,
+		})
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID, "profile_id": input.ProfileID})
+	})
+
+	app.Post("/api/phase-a/accounts/:id/pause", func(c fiber.Ctx) error {
+		if err := store.PauseAccount(c.Context(), c.Params("id")); err != nil {
+			return phaseAError(c, err)
+		}
+		return c.SendStatus(fiber.StatusNoContent)
+	})
+
+	app.Post("/api/phase-a/runtimes", func(c fiber.Ctx) error {
+		var input phasea.RuntimeBinding
+		if err := decodePhaseA(c, &input); err != nil {
+			return phaseAError(c, err)
+		}
+		if err := store.BindRuntime(c.Context(), input); err != nil {
+			return phaseAError(c, err)
+		}
+		return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID})
+	})
+
+	app.Delete("/api/phase-a/runtimes/:id", func(c fiber.Ctx) error {
+		if err := store.ReleaseRuntime(c.Context(), c.Params("id")); err != nil {
+			return phaseAError(c, err)
+		}
+		return c.SendStatus(fiber.StatusNoContent)
+	})
+
+	app.Post("/api/phase-a/drafts", func(c fiber.Ctx) error {
+		var input phasea.Draft
+		if err := decodePhaseA(c, &input); err != nil {
+			return phaseAError(c, err)
+		}
+		if err := store.CreateDraft(c.Context(), input); err != nil {
+			return phaseAError(c, err)
+		}
+		return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID})
+	})
+
+	app.Post("/api/phase-a/confirmations", func(c fiber.Ctx) error {
+		var input phasea.Confirmation
+		if err := decodePhaseA(c, &input); err != nil {
+			return phaseAError(c, err)
+		}
+		if err := store.Confirm(c.Context(), input); err != nil {
+			return phaseAError(c, err)
+		}
+		return c.Status(fiber.StatusCreated).JSON(map[string]string{"id": input.ID})
+	})
+
+	app.Post("/api/phase-a/tasks", func(c fiber.Ctx) error {
+		var input phasea.Task
+		if err := decodePhaseA(c, &input); err != nil {
+			return phaseAError(c, err)
+		}
+		task, inserted, err := store.Enqueue(c.Context(), input)
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		status := fiber.StatusOK
+		if inserted {
+			status = fiber.StatusCreated
+		}
+		return c.Status(status).JSON(task)
+	})
+
+	app.Post("/api/phase-a/tasks/:id/cancel", func(c fiber.Ctx) error {
+		if err := store.CancelTask(c.Context(), c.Params("id")); err != nil {
+			return phaseAError(c, err)
+		}
+		return c.SendStatus(fiber.StatusNoContent)
+	})
+
+	app.Post("/api/phase-a/mock/execute", func(c fiber.Ctx) error {
+		var input struct {
+			WorkerID string `json:"worker_id"`
+			Outcome  string `json:"outcome"`
+		}
+		if err := decodePhaseA(c, &input); err != nil {
+			return phaseAError(c, err)
+		}
+		execution, err := store.ExecuteMock(c.Context(), input.WorkerID, input.Outcome)
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.JSON(execution)
+	})
+
+	app.Get("/api/phase-a/audit", func(c fiber.Ctx) error {
+		events, err := store.Audit(c.Context())
+		if err != nil {
+			return phaseAError(c, err)
+		}
+		return c.JSON(events)
+	})
+}
+
+func decodePhaseA(c fiber.Ctx, destination any) error {
+	decoder := json.NewDecoder(bytes.NewReader(c.Body()))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(destination); err != nil {
+		return phasea.ErrInvalid
+	}
+	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
+		return phasea.ErrInvalid
+	}
+	return nil
+}
+
+func phaseAError(c fiber.Ctx, err error) error {
+	status := fiber.StatusInternalServerError
+	message := "phase A operation failed"
+	switch {
+	case errors.Is(err, phasea.ErrInvalid):
+		status, message = fiber.StatusBadRequest, phasea.ErrInvalid.Error()
+	case errors.Is(err, phasea.ErrConflict):
+		status, message = fiber.StatusConflict, phasea.ErrConflict.Error()
+	case errors.Is(err, phasea.ErrNotFound):
+		status, message = fiber.StatusNotFound, phasea.ErrNotFound.Error()
+	}
+	return c.Status(status).JSON(map[string]string{"error": message})
+}
@@ -3,6 +3,7 @@ services:
    build: .
    environment:
      DOCKER_GATEWAY_URL: http://docker-gateway:8081
+      DATABASE_URL: postgres://creatorhub@postgres/creatorhub?sslmode=disable
    ports:
      - "127.0.0.1:${CREATORHUB_PORT:-8080}:8080"
    read_only: true
@@ -10,7 +11,34 @@ services:
      - /tmp:size=16m,noexec,nosuid,nodev
    cap_drop: [ALL]
    security_opt: [no-new-privileges:true]
-    depends_on: [docker-gateway]
+    depends_on:
+      docker-gateway:
+        condition: service_started
+      postgres:
+        condition: service_healthy
+    networks: [control]
+    restart: unless-stopped
+
+  postgres:
+    image: postgres:17-alpine@sha256:18cfe3ef5e6815560c98237d6216d1e5119702fb0f3894c8785dd58b8bbe5d73
+    environment:
+      POSTGRES_DB: creatorhub
+      POSTGRES_USER: creatorhub
+      POSTGRES_HOST_AUTH_METHOD: trust
+    healthcheck:
+      test: [CMD-SHELL, pg_isready -h 127.0.0.1 -U creatorhub -d creatorhub]
+      interval: 2s
+      timeout: 2s
+      retries: 15
+    volumes:
+      - creatorhub_postgres:/var/lib/postgresql/data
+    read_only: true
+    tmpfs:
+      - /run/postgresql:size=1m,nosuid,nodev
+      - /tmp:size=16m,nosuid,nodev
+    cap_drop: [ALL]
+    cap_add: [CHOWN, FOWNER, DAC_OVERRIDE, SETUID, SETGID]
+    security_opt: [no-new-privileges:true]
    networks: [control]
    restart: unless-stopped

@@ -35,3 +63,6 @@ networks:
  control:
    name: creatorhub_control
    internal: true
+
+volumes:
+  creatorhub_postgres:
@@ -4,15 +4,16 @@

 - 前端：React + Vite。阶段 A 只有一个运行环境页面，不引入路由、状态库或组件库。
 - 后端：Go 模块化单体，Fiber v3 提供 HTTP 路由，Viper 读取并校验启动配置，Logrus 输出 JSON 结构化日志，Cobra 保持当前两个服务入口。控制面提供同源 API 和静态文件，受限网关单独封装 Docker Engine API。
- 数据：本任务不引入业务数据库；Docker 容器标签是运行态事实，Profile 使用命名卷持久化。账号、任务和审计实体仍按产品规划在后续阶段落入 Postgres。
+- 数据：Docker 容器标签仍是运行态事实，Profile 使用命名卷持久化；阶段 A 账号、凭据引用、确认、任务、尝试和审计实体由控制面使用原生 SQL 持久化到 Postgres。
 - 部署：Docker Compose 启动控制面和受限网关；浏览器容器由网关动态创建。

 ## 调用链与契约

 ```text
-React ── /api/browsers ──> control-plane ── /v1/browsers ──> docker-gateway ──> docker.sock
-                                                                    │
-                                                                    └─> browser container
+React ──> control-plane ── /api/browsers ──> docker-gateway ──> docker.sock
+              │                                  │
+              │                                  └─> browser container
+              └─ /api/phase-a ──> PostgreSQL
 ```

 `POST /api/browsers` 接受 `{name, seed}` 并创建后启动；`GET /api/browsers` 列出受管容器；`POST /api/browsers/{name}/start|stop` 改变状态；`DELETE /api/browsers/{name}` 仅回收容器，保留 Profile 数据卷。名称唯一约束由 Docker 容器名保证，重复创建返回冲突；启停接受 Docker 的幂等响应，不自动重试未知结果。
@@ -43,3 +44,11 @@ DOCKER_GID=$(stat -c %g /var/run/docker.sock) docker compose up --build
 打开 <http://127.0.0.1:8080>。浏览器镜像固定为 `git.ipao.vip/rogee/fingerprint-chromium@sha256:b9f23b8e3ac640174db0dfa49e9095fe7eb06f5db55a4e7550d979b35ff3a1b7`。阶段 A 网络默认失败关闭且不能访问外网；后续授权沙箱必须先设计受管出口，再调整 `browser` 网络。

 创建成功但启动失败时，网关会立即删除失败容器并保留命名 Profile 卷，允许同名请求安全重试。跨容器隔离回归可用 `CREATORHUB_DOCKER_INTEGRATION=1 go test -run TestDockerIntegrationBlocksCrossContainerTraffic ./cmd/docker-gateway` 运行。
+
+## 阶段 A 离线闭环
+
+`POST /api/phase-a/accounts` 只接受 OS Keyring/Secret Manager 的引用标识，不接受秘密值；账号的 `profile_id` 全局唯一。`POST /api/phase-a/runtimes` 通过部分唯一索引保证一个账号和一个运行时都只有一条活动绑定。
+
+草稿经 `POST /api/phase-a/confirmations` 显式确认后才可投递到 `/api/phase-a/tasks`。任务由幂等键去重；`POST /api/phase-a/mock/execute` 使用 `FOR UPDATE SKIP LOCKED` 领取一分钟租约，执行前统一核对账号、草稿和确认版本。缺少确认或版本不一致会进入 `needs_confirmation`，暂停账号或 Mock 策略结果会进入 `policy_hold`，不确定结果与过期租约进入 `needs_confirmation`；这些状态都不会自动重试。`GET /api/phase-a/audit` 只导出账号、确认版本、尝试和结果等非秘密证据。
+
+启动时控制面在事务和 advisory lock 下应用前向迁移 `internal/phasea/migrations/001_phase_a.sql`。本迁移只新建表、索引、约束和追加式审计触发器，不删除或改写现有数据；回滚需停服务后人工删除阶段 A 新表，本阶段不提供自动破坏性回滚。
@@ -4,6 +4,7 @@ go 1.26

 require (
 	github.com/gofiber/fiber/v3 v3.5.0
+	github.com/jackc/pgx/v5 v5.10.0
 	github.com/sirupsen/logrus v1.10.0
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/viper v1.21.0
@@ -17,6 +18,9 @@ require (
 	github.com/gofiber/utils/v2 v2.4.1 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/klauspost/compress v1.19.2 // indirect
 	github.com/mattn/go-colorable v0.1.15 // indirect
 	github.com/mattn/go-isatty v0.0.24 // indirect
@@ -34,6 +38,7 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/crypto v0.54.0 // indirect
 	golang.org/x/net v0.57.0 // indirect
+	golang.org/x/sync v0.22.0 // indirect
 	golang.org/x/sys v0.47.0 // indirect
 	golang.org/x/text v0.40.0 // indirect
 )
@@ -1,6 +1,7 @@
 github.com/andybalholm/brotli v1.2.2 h1:HzTuoo2ErYQqf5qvcJInB8uvqSVxRttzkFexPWtnceM=
 github.com/andybalholm/brotli v1.2.2/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
@@ -23,6 +24,14 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
+github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8=
 github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
@@ -61,6 +70,9 @@ github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk=
 github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.21.0 h1:x5S+0EU27Lbphp4UKm1C+1oQO+rKx36vfCoaVebLFSU=
 github.com/spf13/viper v1.21.0/go.mod h1:P0lhsswPGWD/1lZJ9ny3fYnVqxiegrlNrEmgLjbTCAY=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8=
@@ -81,12 +93,15 @@ golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
 golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
 golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
 golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
 golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
 golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
@@ -0,0 +1,107 @@
+CREATE TABLE credential_reference (
+    id text PRIMARY KEY,
+    provider text NOT NULL CHECK (provider IN ('os_keyring', 'secret_manager')),
+    reference_key text NOT NULL UNIQUE,
+    created_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE social_account (
+    id text PRIMARY KEY CHECK (id ~ '^[a-z0-9][a-z0-9-]{0,31}$'),
+    credential_reference_id text NOT NULL UNIQUE REFERENCES credential_reference(id),
+    profile_id text NOT NULL UNIQUE,
+    version bigint NOT NULL DEFAULT 1 CHECK (version > 0),
+    status text NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'paused')),
+    paused_at timestamptz,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE runtime_instance (
+    id text PRIMARY KEY,
+    account_id text NOT NULL REFERENCES social_account(id),
+    runtime_id text NOT NULL,
+    acquired_at timestamptz NOT NULL DEFAULT now(),
+    lease_until timestamptz NOT NULL,
+    released_at timestamptz
+);
+
+CREATE UNIQUE INDEX one_active_runtime_per_account
+    ON runtime_instance (account_id) WHERE released_at IS NULL;
+CREATE UNIQUE INDEX one_account_per_active_runtime
+    ON runtime_instance (runtime_id) WHERE released_at IS NULL;
+
+CREATE TABLE content_draft (
+    id text PRIMARY KEY,
+    account_id text NOT NULL REFERENCES social_account(id),
+    version bigint NOT NULL CHECK (version > 0),
+    content text NOT NULL,
+    created_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE confirmation (
+    id text PRIMARY KEY,
+    account_id text NOT NULL REFERENCES social_account(id),
+    account_version bigint NOT NULL CHECK (account_version > 0),
+    draft_id text NOT NULL REFERENCES content_draft(id),
+    draft_version bigint NOT NULL CHECK (draft_version > 0),
+    version bigint NOT NULL CHECK (version > 0),
+    confirmed_by text NOT NULL DEFAULT 'local-user' CHECK (confirmed_by = 'local-user'),
+    confirmed_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE TABLE operation_task (
+    id text PRIMARY KEY,
+    idempotency_key text NOT NULL UNIQUE,
+    account_id text NOT NULL REFERENCES social_account(id),
+    account_version bigint NOT NULL CHECK (account_version > 0),
+    draft_id text NOT NULL REFERENCES content_draft(id),
+    draft_version bigint NOT NULL CHECK (draft_version > 0),
+    confirmation_id text REFERENCES confirmation(id),
+    confirmation_version bigint CHECK (confirmation_version > 0),
+    state text NOT NULL DEFAULT 'queued' CHECK (state IN (
+        'queued', 'executing', 'succeeded', 'failed',
+        'needs_confirmation', 'policy_hold', 'cancelled'
+    )),
+    lease_owner text,
+    lease_until timestamptz,
+    current_attempt_id text,
+    created_at timestamptz NOT NULL DEFAULT now(),
+    updated_at timestamptz NOT NULL DEFAULT now(),
+    CHECK ((confirmation_id IS NULL) = (confirmation_version IS NULL))
+);
+
+CREATE TABLE execution_attempt (
+    id text PRIMARY KEY,
+    task_id text NOT NULL UNIQUE REFERENCES operation_task(id),
+    started_at timestamptz NOT NULL DEFAULT now(),
+    finished_at timestamptz,
+    outcome text CHECK (outcome IN ('succeeded', 'failed', 'uncertain', 'policy_hold')),
+    result jsonb NOT NULL DEFAULT '{}'::jsonb
+);
+
+ALTER TABLE operation_task
+    ADD CONSTRAINT operation_task_current_attempt_fk
+    FOREIGN KEY (current_attempt_id) REFERENCES execution_attempt(id);
+
+CREATE TABLE audit_event (
+    id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+    event_type text NOT NULL,
+    account_id text REFERENCES social_account(id),
+    confirmation_id text REFERENCES confirmation(id),
+    confirmation_version bigint,
+    attempt_id text REFERENCES execution_attempt(id),
+    task_id text REFERENCES operation_task(id),
+    details jsonb NOT NULL DEFAULT '{}'::jsonb,
+    created_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE FUNCTION reject_audit_event_mutation() RETURNS trigger
+LANGUAGE plpgsql AS $$
+BEGIN
+    RAISE EXCEPTION 'audit_event is append-only';
+END;
+$$;
+
+CREATE TRIGGER audit_event_append_only
+    BEFORE UPDATE OR DELETE ON audit_event
+    FOR EACH ROW EXECUTE FUNCTION reject_audit_event_mutation();
@@ -0,0 +1,658 @@
+package phasea
+
+import (
+	"context"
+	"crypto/rand"
+	"database/sql"
+	_ "embed"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"regexp"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+//go:embed migrations/001_phase_a.sql
+var migration001 string
+
+var (
+	ErrConflict          = errors.New("resource conflicts with existing state")
+	ErrInvalid           = errors.New("invalid phase A input")
+	ErrNotFound          = errors.New("resource not found")
+	idPattern            = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`)
+	refPattern           = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$`)
+	credentialKeyPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}/[A-Za-z0-9][A-Za-z0-9._/-]{0,126}$`)
+)
+
+type Store struct{ db *sql.DB }
+
+type Account struct {
+	ID                    string `json:"id"`
+	CredentialReferenceID string `json:"credential_reference_id"`
+	CredentialProvider    string `json:"credential_provider"`
+	CredentialKey         string `json:"-"`
+	ProfileID             string `json:"profile_id"`
+}
+
+type RuntimeBinding struct {
+	ID         string    `json:"id"`
+	AccountID  string    `json:"account_id"`
+	RuntimeID  string    `json:"runtime_id"`
+	LeaseUntil time.Time `json:"lease_until"`
+}
+
+type Draft struct {
+	ID        string `json:"id"`
+	AccountID string `json:"account_id"`
+	Version   int64  `json:"version"`
+	Content   string `json:"content"`
+}
+
+type Confirmation struct {
+	ID             string `json:"id"`
+	AccountID      string `json:"account_id"`
+	AccountVersion int64  `json:"account_version"`
+	DraftID        string `json:"draft_id"`
+	DraftVersion   int64  `json:"draft_version"`
+	Version        int64  `json:"version"`
+}
+
+type Task struct {
+	ID                  string `json:"id"`
+	IdempotencyKey      string `json:"idempotency_key"`
+	AccountID           string `json:"account_id"`
+	AccountVersion      int64  `json:"account_version"`
+	DraftID             string `json:"draft_id"`
+	DraftVersion        int64  `json:"draft_version"`
+	ConfirmationID      string `json:"confirmation_id"`
+	ConfirmationVersion int64  `json:"confirmation_version"`
+	State               string `json:"state"`
+}
+
+type Execution struct {
+	TaskID              string `json:"task_id"`
+	AttemptID           string `json:"attempt_id"`
+	AccountID           string `json:"account_id"`
+	ConfirmationID      string `json:"confirmation_id"`
+	ConfirmationVersion int64  `json:"confirmation_version"`
+	State               string `json:"state"`
+	WasClaimed          bool   `json:"was_claimed"`
+}
+
+type AuditEvent struct {
+	ID                  int64           `json:"id"`
+	EventType           string          `json:"event_type"`
+	AccountID           string          `json:"account_id,omitempty"`
+	ConfirmationID      string          `json:"confirmation_id,omitempty"`
+	ConfirmationVersion int64           `json:"confirmation_version,omitempty"`
+	AttemptID           string          `json:"attempt_id,omitempty"`
+	TaskID              string          `json:"task_id,omitempty"`
+	Details             json.RawMessage `json:"details"`
+	CreatedAt           time.Time       `json:"created_at"`
+}
+
+func Open(ctx context.Context, databaseURL string) (*Store, error) {
+	db, err := sql.Open("pgx", databaseURL)
+	if err != nil {
+		return nil, errors.New("open phase A database")
+	}
+	db.SetMaxOpenConns(10)
+	db.SetMaxIdleConns(2)
+	db.SetConnMaxIdleTime(5 * time.Minute)
+	if err := db.PingContext(ctx); err != nil {
+		db.Close()
+		return nil, errors.New("connect to phase A database")
+	}
+	store := &Store{db: db}
+	if err := store.migrate(ctx); err != nil {
+		db.Close()
+		return nil, err
+	}
+	return store, nil
+}
+
+func (s *Store) Close() error { return s.db.Close() }
+
+func (s *Store) migrate(ctx context.Context) error {
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.New("begin schema migration")
+	}
+	defer tx.Rollback()
+	if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(1542738011)`); err != nil {
+		return errors.New("lock schema migration")
+	}
+	if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migration (version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil {
+		return errors.New("create schema migration table")
+	}
+	var applied bool
+	if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = 1)`).Scan(&applied); err != nil {
+		return errors.New("read schema migration state")
+	}
+	if !applied {
+		if _, err := tx.ExecContext(ctx, migration001); err != nil {
+			return fmt.Errorf("apply schema migration 1: %w", err)
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migration (version) VALUES (1)`); err != nil {
+			return errors.New("record schema migration 1")
+		}
+	}
+	if err := tx.Commit(); err != nil {
+		return errors.New("commit schema migration")
+	}
+	return nil
+}
+
+func (s *Store) CreateAccount(ctx context.Context, account Account) error {
+	if !idPattern.MatchString(account.ID) || !refPattern.MatchString(account.CredentialReferenceID) ||
+		!credentialKeyPattern.MatchString(account.CredentialKey) || !refPattern.MatchString(account.ProfileID) ||
+		(account.CredentialProvider != "os_keyring" && account.CredentialProvider != "secret_manager") {
+		return ErrInvalid
+	}
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.New("begin account transaction")
+	}
+	defer tx.Rollback()
+	if _, err := tx.ExecContext(ctx, `INSERT INTO credential_reference (id, provider, reference_key) VALUES ($1, $2, $3)`,
+		account.CredentialReferenceID, account.CredentialProvider, account.CredentialKey); err != nil {
+		return publicDatabaseError(err)
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO social_account (id, credential_reference_id, profile_id) VALUES ($1, $2, $3)`,
+		account.ID, account.CredentialReferenceID, account.ProfileID); err != nil {
+		return publicDatabaseError(err)
+	}
+	if err := appendAudit(ctx, tx, "account_created", account.ID, "", 0, "", "", map[string]string{"profile_id": account.ProfileID}); err != nil {
+		return err
+	}
+	return commit(tx)
+}
+
+func (s *Store) BindRuntime(ctx context.Context, binding RuntimeBinding) error {
+	if !refPattern.MatchString(binding.ID) || !idPattern.MatchString(binding.AccountID) ||
+		!refPattern.MatchString(binding.RuntimeID) || !binding.LeaseUntil.After(time.Now()) {
+		return ErrInvalid
+	}
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.New("begin runtime transaction")
+	}
+	defer tx.Rollback()
+	if _, err := tx.ExecContext(ctx, `UPDATE runtime_instance SET released_at = now() WHERE released_at IS NULL AND lease_until < now()`); err != nil {
+		return errors.New("expire runtime bindings")
+	}
+	var accountID string
+	if err := tx.QueryRowContext(ctx, `
+		INSERT INTO runtime_instance (id, account_id, runtime_id, lease_until)
+		SELECT $1, id, $3, $4 FROM social_account WHERE id = $2 AND status = 'active'
+		RETURNING account_id`, binding.ID, binding.AccountID, binding.RuntimeID, binding.LeaseUntil).Scan(&accountID); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return ErrConflict
+		}
+		return rowError(err)
+	}
+	if err := appendAudit(ctx, tx, "runtime_bound", accountID, "", 0, "", "", map[string]string{"runtime_id": binding.RuntimeID}); err != nil {
+		return err
+	}
+	return commit(tx)
+}
+
+func (s *Store) ReleaseRuntime(ctx context.Context, id string) error {
+	if !refPattern.MatchString(id) {
+		return ErrInvalid
+	}
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.New("begin runtime release")
+	}
+	defer tx.Rollback()
+	var accountID string
+	if err := tx.QueryRowContext(ctx, `UPDATE runtime_instance SET released_at = now() WHERE id = $1 AND released_at IS NULL RETURNING account_id`, id).Scan(&accountID); err != nil {
+		return rowError(err)
+	}
+	if err := appendAudit(ctx, tx, "runtime_released", accountID, "", 0, "", "", map[string]string{"binding_id": id}); err != nil {
+		return err
+	}
+	return commit(tx)
+}
+
+func (s *Store) CreateDraft(ctx context.Context, draft Draft) error {
+	if !refPattern.MatchString(draft.ID) || !idPattern.MatchString(draft.AccountID) || draft.Version < 1 || strings.TrimSpace(draft.Content) == "" {
+		return ErrInvalid
+	}
+	_, err := s.db.ExecContext(ctx, `INSERT INTO content_draft (id, account_id, version, content) VALUES ($1, $2, $3, $4)`,
+		draft.ID, draft.AccountID, draft.Version, draft.Content)
+	return publicDatabaseError(err)
+}
+
+func (s *Store) Confirm(ctx context.Context, confirmation Confirmation) error {
+	if !refPattern.MatchString(confirmation.ID) || !idPattern.MatchString(confirmation.AccountID) ||
+		!refPattern.MatchString(confirmation.DraftID) || confirmation.AccountVersion < 1 || confirmation.DraftVersion < 1 || confirmation.Version < 1 {
+		return ErrInvalid
+	}
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.New("begin confirmation transaction")
+	}
+	defer tx.Rollback()
+	var valid bool
+	if err := tx.QueryRowContext(ctx, `
+		SELECT EXISTS (
+			SELECT 1 FROM social_account a JOIN content_draft d ON d.account_id = a.id
+			WHERE a.id = $1 AND a.version = $2 AND d.id = $3 AND d.version = $4
+		)`, confirmation.AccountID, confirmation.AccountVersion, confirmation.DraftID, confirmation.DraftVersion).Scan(&valid); err != nil {
+		return errors.New("validate confirmation versions")
+	}
+	if !valid {
+		return ErrConflict
+	}
+	if _, err := tx.ExecContext(ctx, `
+		INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
+		VALUES ($1, $2, $3, $4, $5, $6)`, confirmation.ID, confirmation.AccountID, confirmation.AccountVersion,
+		confirmation.DraftID, confirmation.DraftVersion, confirmation.Version); err != nil {
+		return publicDatabaseError(err)
+	}
+	if err := appendAudit(ctx, tx, "draft_confirmed", confirmation.AccountID, confirmation.ID, confirmation.Version, "", "", map[string]any{
+		"account_version": confirmation.AccountVersion, "draft_id": confirmation.DraftID, "draft_version": confirmation.DraftVersion,
+	}); err != nil {
+		return err
+	}
+	return commit(tx)
+}
+
+func (s *Store) Enqueue(ctx context.Context, task Task) (Task, bool, error) {
+	if !refPattern.MatchString(task.ID) || !refPattern.MatchString(task.IdempotencyKey) || !idPattern.MatchString(task.AccountID) ||
+		!refPattern.MatchString(task.DraftID) || task.AccountVersion < 1 || task.DraftVersion < 1 ||
+		((task.ConfirmationID == "") != (task.ConfirmationVersion == 0)) ||
+		(task.ConfirmationID != "" && (!refPattern.MatchString(task.ConfirmationID) || task.ConfirmationVersion < 1)) {
+		return Task{}, false, ErrInvalid
+	}
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return Task{}, false, errors.New("begin task transaction")
+	}
+	defer tx.Rollback()
+	var insertedID string
+	err = tx.QueryRowContext(ctx, `
+		INSERT INTO operation_task (id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version)
+		VALUES ($1, $2, $3, $4, $5, $6, NULLIF($7, ''), NULLIF($8, 0))
+		ON CONFLICT (idempotency_key) DO NOTHING RETURNING id`, task.ID, task.IdempotencyKey, task.AccountID, task.AccountVersion,
+		task.DraftID, task.DraftVersion, task.ConfirmationID, task.ConfirmationVersion).Scan(&insertedID)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return Task{}, false, publicDatabaseError(err)
+	}
+	if insertedID != "" {
+		task.State = "queued"
+		if err := appendAudit(ctx, tx, "task_queued", task.AccountID, task.ConfirmationID, task.ConfirmationVersion, "", task.ID, nil); err != nil {
+			return Task{}, false, err
+		}
+		if err := commit(tx); err != nil {
+			return Task{}, false, err
+		}
+		return task, true, nil
+	}
+
+	var existing Task
+	var confirmationID sql.NullString
+	var confirmationVersion sql.NullInt64
+	err = tx.QueryRowContext(ctx, `
+		SELECT id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version, state
+		FROM operation_task WHERE idempotency_key = $1`, task.IdempotencyKey).Scan(&existing.ID, &existing.IdempotencyKey,
+		&existing.AccountID, &existing.AccountVersion, &existing.DraftID, &existing.DraftVersion, &confirmationID, &confirmationVersion, &existing.State)
+	if err != nil {
+		return Task{}, false, rowError(err)
+	}
+	existing.ConfirmationID = confirmationID.String
+	existing.ConfirmationVersion = confirmationVersion.Int64
+	if existing.AccountID != task.AccountID || existing.AccountVersion != task.AccountVersion || existing.DraftID != task.DraftID ||
+		existing.DraftVersion != task.DraftVersion || existing.ConfirmationID != task.ConfirmationID || existing.ConfirmationVersion != task.ConfirmationVersion {
+		return Task{}, false, ErrConflict
+	}
+	if err := commit(tx); err != nil {
+		return Task{}, false, err
+	}
+	return existing, false, nil
+}
+
+func (s *Store) PauseAccount(ctx context.Context, accountID string) error {
+	if !idPattern.MatchString(accountID) {
+		return ErrInvalid
+	}
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.New("begin pause transaction")
+	}
+	defer tx.Rollback()
+	var version int64
+	var status string
+	if err := tx.QueryRowContext(ctx, `SELECT version, status FROM social_account WHERE id = $1 FOR UPDATE`, accountID).Scan(&version, &status); err != nil {
+		return rowError(err)
+	}
+	if status == "paused" {
+		return commit(tx)
+	}
+	if err := tx.QueryRowContext(ctx, `
+		UPDATE social_account SET status = 'paused', paused_at = now(), version = version + 1, updated_at = now()
+		WHERE id = $1 RETURNING version`, accountID).Scan(&version); err != nil {
+		return errors.New("pause account")
+	}
+	result, err := tx.ExecContext(ctx, `
+		UPDATE operation_task SET state = 'policy_hold', updated_at = now()
+		WHERE account_id = $1 AND state = 'queued'`, accountID)
+	if err != nil {
+		return errors.New("hold queued account tasks")
+	}
+	held, _ := result.RowsAffected()
+	if err := appendAudit(ctx, tx, "account_paused", accountID, "", 0, "", "", map[string]any{"account_version": version, "tasks_held": held}); err != nil {
+		return err
+	}
+	return commit(tx)
+}
+
+func (s *Store) CancelTask(ctx context.Context, taskID string) error {
+	if !refPattern.MatchString(taskID) {
+		return ErrInvalid
+	}
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return errors.New("begin cancel transaction")
+	}
+	defer tx.Rollback()
+	var accountID, state string
+	var attemptID sql.NullString
+	var confirmationID sql.NullString
+	var confirmationVersion sql.NullInt64
+	err = tx.QueryRowContext(ctx, `
+		UPDATE operation_task SET state = CASE WHEN state = 'executing' THEN 'needs_confirmation' ELSE 'cancelled' END,
+			lease_owner = NULL, lease_until = NULL, updated_at = now()
+		WHERE id = $1 AND state IN ('queued', 'executing', 'needs_confirmation', 'policy_hold')
+		RETURNING account_id, state, current_attempt_id, confirmation_id, confirmation_version`, taskID).Scan(
+		&accountID, &state, &attemptID, &confirmationID, &confirmationVersion)
+	if err != nil {
+		return rowError(err)
+	}
+	if attemptID.Valid {
+		if _, err := tx.ExecContext(ctx, `
+			UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain'
+			WHERE id = $1 AND finished_at IS NULL`, attemptID.String); err != nil {
+			return errors.New("mark cancelled attempt uncertain")
+		}
+	}
+	if err := appendAudit(ctx, tx, "task_cancel_requested", accountID, confirmationID.String, confirmationVersion.Int64, attemptID.String, taskID, map[string]string{"state": state}); err != nil {
+		return err
+	}
+	return commit(tx)
+}
+
+func (s *Store) ExecuteMock(ctx context.Context, workerID, outcome string) (Execution, error) {
+	if !refPattern.MatchString(workerID) || (outcome != "succeeded" && outcome != "failed" && outcome != "uncertain" && outcome != "policy_hold") {
+		return Execution{}, ErrInvalid
+	}
+	execution, err := s.claim(ctx, workerID)
+	if err != nil || !execution.WasClaimed {
+		return execution, err
+	}
+	return s.complete(ctx, workerID, execution, outcome)
+}
+
+func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return Execution{}, errors.New("begin claim transaction")
+	}
+	defer tx.Rollback()
+	if err := quarantineExpired(ctx, tx); err != nil {
+		return Execution{}, err
+	}
+	if err := quarantineInvalid(ctx, tx); err != nil {
+		return Execution{}, err
+	}
+	var execution Execution
+	err = tx.QueryRowContext(ctx, `
+		WITH candidate AS (
+			SELECT t.id
+			FROM operation_task t
+			JOIN social_account a ON a.id = t.account_id
+			JOIN content_draft d ON d.id = t.draft_id
+			JOIN confirmation c ON c.id = t.confirmation_id
+			WHERE t.state = 'queued' AND a.status = 'active'
+			  AND a.version = t.account_version
+			  AND d.account_id = t.account_id AND d.version = t.draft_version
+			  AND c.account_id = t.account_id AND c.account_version = t.account_version
+			  AND c.draft_id = t.draft_id AND c.draft_version = t.draft_version
+			  AND c.version = t.confirmation_version
+			ORDER BY t.created_at, t.id
+			FOR UPDATE OF t SKIP LOCKED LIMIT 1
+		)
+		UPDATE operation_task t SET state = 'executing', lease_owner = $1,
+			lease_until = now() + interval '1 minute', updated_at = now()
+		FROM candidate WHERE t.id = candidate.id
+		RETURNING t.id, t.account_id, t.confirmation_id, t.confirmation_version`, workerID).Scan(
+		&execution.TaskID, &execution.AccountID, &execution.ConfirmationID, &execution.ConfirmationVersion)
+	if errors.Is(err, sql.ErrNoRows) {
+		if err := commit(tx); err != nil {
+			return Execution{}, err
+		}
+		return Execution{}, nil
+	}
+	if err != nil {
+		return Execution{}, errors.New("claim task")
+	}
+	execution.AttemptID = newID()
+	execution.State = "executing"
+	execution.WasClaimed = true
+	if _, err := tx.ExecContext(ctx, `INSERT INTO execution_attempt (id, task_id) VALUES ($1, $2)`, execution.AttemptID, execution.TaskID); err != nil {
+		return Execution{}, publicDatabaseError(err)
+	}
+	if _, err := tx.ExecContext(ctx, `UPDATE operation_task SET current_attempt_id = $1 WHERE id = $2`, execution.AttemptID, execution.TaskID); err != nil {
+		return Execution{}, errors.New("attach execution attempt")
+	}
+	if err := appendAudit(ctx, tx, "task_claimed", execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"worker_id": workerID}); err != nil {
+		return Execution{}, err
+	}
+	if err := commit(tx); err != nil {
+		return Execution{}, err
+	}
+	return execution, nil
+}
+
+func (s *Store) complete(ctx context.Context, workerID string, execution Execution, outcome string) (Execution, error) {
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return Execution{}, errors.New("begin completion transaction")
+	}
+	defer tx.Rollback()
+	var leaseUntil time.Time
+	err = tx.QueryRowContext(ctx, `
+		SELECT lease_until FROM operation_task
+		WHERE id = $1 AND state = 'executing' AND lease_owner = $2 AND current_attempt_id = $3
+		FOR UPDATE`, execution.TaskID, workerID, execution.AttemptID).Scan(&leaseUntil)
+	if err != nil {
+		return Execution{}, rowError(err)
+	}
+	state := map[string]string{"succeeded": "succeeded", "failed": "failed", "uncertain": "needs_confirmation", "policy_hold": "policy_hold"}[outcome]
+	if leaseUntil.Before(time.Now()) {
+		outcome, state = "uncertain", "needs_confirmation"
+	}
+	result, _ := json.Marshal(map[string]string{"mock_outcome": outcome})
+	if _, err := tx.ExecContext(ctx, `UPDATE execution_attempt SET finished_at = now(), outcome = $1, result = $2 WHERE id = $3`, outcome, result, execution.AttemptID); err != nil {
+		return Execution{}, errors.New("finish execution attempt")
+	}
+	if _, err := tx.ExecContext(ctx, `
+		UPDATE operation_task SET state = $1, lease_owner = NULL, lease_until = NULL, updated_at = now() WHERE id = $2`, state, execution.TaskID); err != nil {
+		return Execution{}, errors.New("finish task")
+	}
+	if err := appendAudit(ctx, tx, "task_finished", execution.AccountID, execution.ConfirmationID, execution.ConfirmationVersion, execution.AttemptID, execution.TaskID, map[string]string{"state": state}); err != nil {
+		return Execution{}, err
+	}
+	if err := commit(tx); err != nil {
+		return Execution{}, err
+	}
+	execution.State = state
+	return execution, nil
+}
+
+func quarantineExpired(ctx context.Context, tx *sql.Tx) error {
+	rows, err := tx.QueryContext(ctx, `
+		UPDATE operation_task SET state = 'needs_confirmation', lease_owner = NULL, lease_until = NULL, updated_at = now()
+		WHERE state = 'executing' AND lease_until < now()
+		RETURNING id, account_id, current_attempt_id, confirmation_id, confirmation_version`)
+	if err != nil {
+		return errors.New("quarantine expired tasks")
+	}
+	type expiredTask struct {
+		taskID, accountID, attemptID, confirmationID string
+		confirmationVersion                          int64
+	}
+	var expired []expiredTask
+	for rows.Next() {
+		var taskID, accountID string
+		var attemptID sql.NullString
+		var confirmationID sql.NullString
+		var confirmationVersion sql.NullInt64
+		if err := rows.Scan(&taskID, &accountID, &attemptID, &confirmationID, &confirmationVersion); err != nil {
+			rows.Close()
+			return errors.New("read expired task")
+		}
+		expired = append(expired, expiredTask{taskID, accountID, attemptID.String, confirmationID.String, confirmationVersion.Int64})
+	}
+	if err := rows.Close(); err != nil {
+		return errors.New("close expired task rows")
+	}
+	for _, task := range expired {
+		if task.attemptID != "" {
+			if _, err := tx.ExecContext(ctx, `UPDATE execution_attempt SET finished_at = now(), outcome = 'uncertain' WHERE id = $1 AND finished_at IS NULL`, task.attemptID); err != nil {
+				return errors.New("mark expired attempt uncertain")
+			}
+		}
+		if err := appendAudit(ctx, tx, "task_lease_expired", task.accountID, task.confirmationID, task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func quarantineInvalid(ctx context.Context, tx *sql.Tx) error {
+	rows, err := tx.QueryContext(ctx, `
+		UPDATE operation_task t SET
+			state = CASE WHEN EXISTS (
+				SELECT 1 FROM social_account paused WHERE paused.id = t.account_id AND paused.status = 'paused'
+			) THEN 'policy_hold' ELSE 'needs_confirmation' END,
+			updated_at = now()
+		WHERE t.state = 'queued' AND NOT EXISTS (
+			SELECT 1
+			FROM social_account a
+			JOIN content_draft d ON d.id = t.draft_id
+			JOIN confirmation c ON c.id = t.confirmation_id
+			WHERE a.id = t.account_id AND a.status = 'active' AND a.version = t.account_version
+			  AND d.account_id = t.account_id AND d.version = t.draft_version
+			  AND c.account_id = t.account_id AND c.account_version = t.account_version
+			  AND c.draft_id = t.draft_id AND c.draft_version = t.draft_version
+			  AND c.version = t.confirmation_version
+		)
+		RETURNING t.id, t.account_id, t.confirmation_id, t.confirmation_version, t.state`)
+	if err != nil {
+		return errors.New("quarantine invalid tasks")
+	}
+	type invalidTask struct {
+		taskID, accountID, confirmationID, state string
+		confirmationVersion                      int64
+	}
+	var invalid []invalidTask
+	for rows.Next() {
+		var taskID, accountID, state string
+		var confirmationID sql.NullString
+		var confirmationVersion sql.NullInt64
+		if err := rows.Scan(&taskID, &accountID, &confirmationID, &confirmationVersion, &state); err != nil {
+			rows.Close()
+			return errors.New("read quarantined task")
+		}
+		invalid = append(invalid, invalidTask{taskID, accountID, confirmationID.String, state, confirmationVersion.Int64})
+	}
+	if err := rows.Close(); err != nil {
+		return errors.New("close quarantined task rows")
+	}
+	for _, task := range invalid {
+		if err := appendAudit(ctx, tx, "task_quarantined", task.accountID, task.confirmationID, task.confirmationVersion, "", task.taskID, map[string]string{"state": task.state}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT id, event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id, details, created_at
+		FROM audit_event ORDER BY id`)
+	if err != nil {
+		return nil, errors.New("read audit events")
+	}
+	defer rows.Close()
+	var events []AuditEvent
+	for rows.Next() {
+		var event AuditEvent
+		var accountID, confirmationID, attemptID, taskID sql.NullString
+		var confirmationVersion sql.NullInt64
+		if err := rows.Scan(&event.ID, &event.EventType, &accountID, &confirmationID, &confirmationVersion, &attemptID, &taskID, &event.Details, &event.CreatedAt); err != nil {
+			return nil, errors.New("decode audit event")
+		}
+		event.AccountID, event.ConfirmationID, event.ConfirmationVersion = accountID.String, confirmationID.String, confirmationVersion.Int64
+		event.AttemptID, event.TaskID = attemptID.String, taskID.String
+		events = append(events, event)
+	}
+	return events, rows.Err()
+}
+
+func appendAudit(ctx context.Context, tx *sql.Tx, eventType, accountID, confirmationID string, confirmationVersion int64, attemptID, taskID string, details any) error {
+	if details == nil {
+		details = map[string]any{}
+	}
+	encoded, err := json.Marshal(details)
+	if err != nil {
+		return errors.New("encode audit details")
+	}
+	_, err = tx.ExecContext(ctx, `
+		INSERT INTO audit_event (event_type, account_id, confirmation_id, confirmation_version, attempt_id, task_id, details)
+		VALUES ($1, NULLIF($2, ''), NULLIF($3, ''), NULLIF($4, 0), NULLIF($5, ''), NULLIF($6, ''), $7)`,
+		eventType, accountID, confirmationID, confirmationVersion, attemptID, taskID, encoded)
+	if err != nil {
+		return errors.New("append audit event")
+	}
+	return nil
+}
+
+func newID() string {
+	var value [16]byte
+	_, _ = rand.Read(value[:])
+	return hex.EncodeToString(value[:])
+}
+
+func commit(tx *sql.Tx) error {
+	if err := tx.Commit(); err != nil {
+		return errors.New("commit transaction")
+	}
+	return nil
+}
+
+func rowError(err error) error {
+	if errors.Is(err, sql.ErrNoRows) {
+		return ErrNotFound
+	}
+	return publicDatabaseError(err)
+}
+
+func publicDatabaseError(err error) error {
+	if err == nil {
+		return nil
+	}
+	var postgresError *pgconn.PgError
+	if errors.As(err, &postgresError) && (postgresError.Code == "23505" || postgresError.Code == "23503" || postgresError.Code == "23514") {
+		return ErrConflict
+	}
+	return errors.New("phase A persistence operation failed")
+}
@@ -0,0 +1,263 @@
+package phasea
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestValidationRejectsInvalidInputsBeforePersistence(t *testing.T) {
+	store := &Store{}
+	if err := store.CreateAccount(context.Background(), Account{ID: "INVALID"}); !errors.Is(err, ErrInvalid) {
+		t.Fatalf("expected invalid account, got %v", err)
+	}
+	if err := store.CreateAccount(context.Background(), Account{
+		ID: "account-a", CredentialReferenceID: "credential-a", CredentialProvider: "os_keyring",
+		CredentialKey: "raw-value", ProfileID: "profile-a",
+	}); !errors.Is(err, ErrInvalid) {
+		t.Fatalf("expected non-reference credential value to be rejected, got %v", err)
+	}
+	if _, _, err := store.Enqueue(context.Background(), Task{ID: "task-a"}); !errors.Is(err, ErrInvalid) {
+		t.Fatalf("expected invalid task, got %v", err)
+	}
+	if _, err := store.ExecuteMock(context.Background(), "worker-a", "retry"); !errors.Is(err, ErrInvalid) {
+		t.Fatalf("expected unsupported outcome to be rejected, got %v", err)
+	}
+}
+
+func TestPhaseAOfflineWorkflow(t *testing.T) {
+	databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL")
+	if databaseURL == "" {
+		t.Skip("set CREATORHUB_POSTGRES_TEST_URL to run PostgreSQL integration coverage")
+	}
+	ctx := context.Background()
+	store, err := Open(ctx, databaseURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = store.Close() })
+	if _, err := store.db.ExecContext(ctx, `
+		TRUNCATE audit_event, execution_attempt, operation_task, confirmation, content_draft,
+			runtime_instance, social_account, credential_reference RESTART IDENTITY CASCADE`); err != nil {
+		t.Fatal(err)
+	}
+
+	accounts := []Account{
+		{ID: "account-a", CredentialReferenceID: "credential-a", CredentialProvider: "os_keyring", CredentialKey: "creatorhub/account-a", ProfileID: "profile-a"},
+		{ID: "account-b", CredentialReferenceID: "credential-b", CredentialProvider: "os_keyring", CredentialKey: "creatorhub/account-b", ProfileID: "profile-b"},
+	}
+	for _, account := range accounts {
+		if err := store.CreateAccount(ctx, account); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := store.CreateAccount(ctx, Account{
+		ID: "account-c", CredentialReferenceID: "credential-c", CredentialProvider: "os_keyring",
+		CredentialKey: "creatorhub/account-c", ProfileID: "profile-a",
+	}); !errors.Is(err, ErrConflict) {
+		t.Fatalf("expected profiles to be unique, got %v", err)
+	}
+
+	lease := time.Now().Add(time.Minute)
+	if err := store.BindRuntime(ctx, RuntimeBinding{ID: "binding-a", AccountID: "account-a", RuntimeID: "runtime-a", LeaseUntil: lease}); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.BindRuntime(ctx, RuntimeBinding{ID: "binding-a-2", AccountID: "account-a", RuntimeID: "runtime-b", LeaseUntil: lease}); !errors.Is(err, ErrConflict) {
+		t.Fatalf("expected one active runtime per account, got %v", err)
+	}
+	if err := store.BindRuntime(ctx, RuntimeBinding{ID: "binding-b", AccountID: "account-b", RuntimeID: "runtime-a", LeaseUntil: lease}); !errors.Is(err, ErrConflict) {
+		t.Fatalf("expected one account per runtime, got %v", err)
+	}
+	if err := store.ReleaseRuntime(ctx, "binding-a"); err != nil {
+		t.Fatal(err)
+	}
+
+	createApprovedDraft(t, store, "account-a", "draft-a", "confirmation-a")
+	for index := range 20 {
+		task := approvedTask(index, "account-a", "draft-a", "confirmation-a")
+		if _, inserted, err := store.Enqueue(ctx, task); err != nil || !inserted {
+			t.Fatalf("enqueue task %d: inserted=%v err=%v", index, inserted, err)
+		}
+		task.ID = fmt.Sprintf("duplicate-%02d", index)
+		if existing, inserted, err := store.Enqueue(ctx, task); err != nil || inserted || existing.ID == task.ID {
+			t.Fatalf("redeliver task %d: existing=%q inserted=%v err=%v", index, existing.ID, inserted, err)
+		}
+	}
+
+	var executed atomic.Int64
+	var workers sync.WaitGroup
+	errorsFromWorkers := make(chan error, 4)
+	for worker := range 4 {
+		workers.Add(1)
+		go func() {
+			defer workers.Done()
+			for {
+				execution, err := store.ExecuteMock(ctx, fmt.Sprintf("worker-%d", worker), "succeeded")
+				if err != nil {
+					errorsFromWorkers <- err
+					return
+				}
+				if !execution.WasClaimed {
+					return
+				}
+				executed.Add(1)
+			}
+		}()
+	}
+	workers.Wait()
+	close(errorsFromWorkers)
+	for err := range errorsFromWorkers {
+		t.Fatal(err)
+	}
+	if executed.Load() != 20 {
+		t.Fatalf("expected exactly 20 executions, got %d", executed.Load())
+	}
+	assertCount(t, store, `SELECT count(*) FROM operation_task WHERE state = 'succeeded'`, 20)
+	assertCount(t, store, `SELECT count(*) FROM execution_attempt`, 20)
+
+	createApprovedDraft(t, store, "account-b", "draft-b", "confirmation-b")
+	unconfirmed := approvedTask(30, "account-b", "draft-b", "")
+	unconfirmed.ConfirmationVersion = 0
+	if _, _, err := store.Enqueue(ctx, unconfirmed); err != nil {
+		t.Fatal(err)
+	}
+	mismatch := approvedTask(31, "account-b", "draft-b", "confirmation-b")
+	mismatch.AccountVersion = 2
+	if _, _, err := store.Enqueue(ctx, mismatch); err != nil {
+		t.Fatal(err)
+	}
+	if execution, err := store.ExecuteMock(ctx, "worker-validation", "succeeded"); err != nil || execution.WasClaimed {
+		t.Fatalf("invalid tasks must not execute: execution=%+v err=%v", execution, err)
+	}
+	assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id IN ('task-30', 'task-31') AND state = 'needs_confirmation'`, 2)
+	assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id IN ('task-30', 'task-31')`, 0)
+
+	uncertain := approvedTask(32, "account-b", "draft-b", "confirmation-b")
+	if _, _, err := store.Enqueue(ctx, uncertain); err != nil {
+		t.Fatal(err)
+	}
+	if execution, err := store.ExecuteMock(ctx, "worker-uncertain", "uncertain"); err != nil || execution.State != "needs_confirmation" {
+		t.Fatalf("uncertain outcome was not held: execution=%+v err=%v", execution, err)
+	}
+	if execution, err := store.ExecuteMock(ctx, "worker-uncertain", "succeeded"); err != nil || execution.WasClaimed {
+		t.Fatalf("uncertain task was retried: execution=%+v err=%v", execution, err)
+	}
+	assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id = 'task-32'`, 1)
+
+	policy := approvedTask(33, "account-b", "draft-b", "confirmation-b")
+	if _, _, err := store.Enqueue(ctx, policy); err != nil {
+		t.Fatal(err)
+	}
+	if execution, err := store.ExecuteMock(ctx, "worker-policy", "policy_hold"); err != nil || execution.State != "policy_hold" {
+		t.Fatalf("policy outcome was not held: execution=%+v err=%v", execution, err)
+	}
+	if execution, err := store.ExecuteMock(ctx, "worker-policy", "succeeded"); err != nil || execution.WasClaimed {
+		t.Fatalf("policy task was retried: execution=%+v err=%v", execution, err)
+	}
+
+	paused := approvedTask(34, "account-b", "draft-b", "confirmation-b")
+	if _, _, err := store.Enqueue(ctx, paused); err != nil {
+		t.Fatal(err)
+	}
+	pauseStarted := time.Now()
+	if err := store.PauseAccount(ctx, "account-b"); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.PauseAccount(ctx, "account-b"); err != nil {
+		t.Fatalf("pause should be idempotent: %v", err)
+	}
+	if execution, err := store.ExecuteMock(ctx, "worker-paused", "succeeded"); err != nil || execution.WasClaimed {
+		t.Fatalf("paused account task was claimed: execution=%+v err=%v", execution, err)
+	}
+	assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-34' AND state = 'policy_hold'`, 1)
+	var pausedAt time.Time
+	if err := store.db.QueryRowContext(ctx, `SELECT paused_at FROM social_account WHERE id = 'account-b'`).Scan(&pausedAt); err != nil || pausedAt.Sub(pauseStarted) >= time.Minute {
+		t.Fatalf("account was not paused within one minute: paused_at=%v err=%v", pausedAt, err)
+	}
+
+	cancelled := approvedTask(35, "account-a", "draft-a", "confirmation-a")
+	if _, _, err := store.Enqueue(ctx, cancelled); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.CancelTask(ctx, cancelled.ID); err != nil {
+		t.Fatal(err)
+	}
+	assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-35' AND state = 'cancelled'`, 1)
+
+	expired := approvedTask(36, "account-a", "draft-a", "confirmation-a")
+	if _, _, err := store.Enqueue(ctx, expired); err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := store.claim(ctx, "worker-crashed")
+	if err != nil || !claimed.WasClaimed {
+		t.Fatalf("claim task for lease test: execution=%+v err=%v", claimed, err)
+	}
+	if _, err := store.db.ExecContext(ctx, `UPDATE operation_task SET lease_until = now() - interval '1 second' WHERE id = $1`, claimed.TaskID); err != nil {
+		t.Fatal(err)
+	}
+	if execution, err := store.ExecuteMock(ctx, "worker-after-crash", "succeeded"); err != nil || execution.WasClaimed {
+		t.Fatalf("expired unknown result was retried: execution=%+v err=%v", execution, err)
+	}
+	assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-36' AND state = 'needs_confirmation'`, 1)
+	assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id = 'task-36' AND outcome = 'uncertain'`, 1)
+
+	events, err := store.Audit(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	traceable := false
+	for _, event := range events {
+		if event.EventType == "task_finished" && event.ConfirmationID != "" && event.ConfirmationVersion == 1 && event.AttemptID != "" && event.TaskID != "" {
+			traceable = true
+			break
+		}
+	}
+	if !traceable {
+		t.Fatal("audit does not trace confirmation version, task, and attempt")
+	}
+	exported, _ := json.Marshal(events)
+	for _, forbidden := range []string{"password", "cookie", "token", "credential-a", "creatorhub/account-a"} {
+		if strings.Contains(strings.ToLower(string(exported)), forbidden) {
+			t.Fatalf("audit export contains sensitive field or credential reference %q", forbidden)
+		}
+	}
+	if _, err := store.db.ExecContext(ctx, `UPDATE audit_event SET event_type = 'rewritten' WHERE id = 1`); err == nil {
+		t.Fatal("audit events must be append-only")
+	}
+}
+
+func createApprovedDraft(t *testing.T, store *Store, accountID, draftID, confirmationID string) {
+	t.Helper()
+	ctx := context.Background()
+	if err := store.CreateDraft(ctx, Draft{ID: draftID, AccountID: accountID, Version: 1, Content: "offline mock content"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.Confirm(ctx, Confirmation{
+		ID: confirmationID, AccountID: accountID, AccountVersion: 1, DraftID: draftID, DraftVersion: 1, Version: 1,
+	}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func approvedTask(index int, accountID, draftID, confirmationID string) Task {
+	return Task{
+		ID: fmt.Sprintf("task-%02d", index), IdempotencyKey: fmt.Sprintf("idempotency-%02d", index),
+		AccountID: accountID, AccountVersion: 1, DraftID: draftID, DraftVersion: 1,
+		ConfirmationID: confirmationID, ConfirmationVersion: 1,
+	}
+}
+
+func assertCount(t *testing.T, store *Store, query string, expected int) {
+	t.Helper()
+	var actual int
+	if err := store.db.QueryRowContext(context.Background(), query).Scan(&actual); err != nil || actual != expected {
+		t.Fatalf("count mismatch: expected=%d actual=%d err=%v query=%s", expected, actual, err, query)
+	}
+}