HH-867: 控制面单用户认证与停机通知 (#30)

.env.example

@@ -1,3 +1,5 @@
 DOCKER_GID=989
 CREATORHUB_PORT=8082
 GATEWAY_TOKEN=dev-creatorhub-gateway-token
+CONTROL_PLANE_USERNAME=
+CONTROL_PLANE_PASSWORD=

README.md

@@ -14,10 +14,14 @@
 需要 Docker Compose：
 
 ```bash
+export CONTROL_PLANE_USERNAME=creatorhub
+export CONTROL_PLANE_PASSWORD="$(openssl rand -hex 24)"
 DOCKER_GID=$(stat -c %g /var/run/docker.sock) docker compose up --build
 ```
 
-打开 <http://127.0.0.1:8080>；局域网内用宿主机 IP 访问同一端口。首次使用：在「网关管理」用 Compose 里的 `GATEWAY_TOKEN` 注册 `http://docker-gateway:8081`，在「镜像版本」添加可用的指纹浏览器镜像引用，即可创建环境；网关会在镜像缺失时自动拉取。架构、API 契约、失败语义和 `docker.sock` 风险边界见
+两个控制面凭据变量均为必填；Compose 不提供默认凭据。开发环境可复制 `.env.example` 后显式填写本地值。
+
+打开 <http://127.0.0.1:8080>，使用 `CONTROL_PLANE_USERNAME` / `CONTROL_PLANE_PASSWORD` 登录；局域网内用宿主机 IP 访问同一端口。首次使用：在「网关管理」用 Compose 里的 `GATEWAY_TOKEN` 注册 `http://docker-gateway:8081`，在「镜像版本」添加可用的指纹浏览器镜像引用，即可创建环境；网关会在镜像缺失时自动拉取。架构、API 契约、失败语义和 `docker.sock` 风险边界见
 [《浏览器容器控制面》](docs/architecture/container-control.md)。
 
 最小验证：

cmd/control-plane/hub_test.go

@@ -782,13 +782,17 @@ func newTestAppWithNetwork(t *testing.T, store *memoryStore, gateway *fakeGatewa
 	return app
 }
 
-func do(app *fiber.App, method, path, body string) *httptest.ResponseRecorder {
+func do(app *fiber.App, method, path, body string, credentials ...string) *httptest.ResponseRecorder {
 	response := httptest.NewRecorder()
 	var reader io.Reader
 	if body != "" {
 		reader = strings.NewReader(body)
 	}
-	adaptor.FiberApp(app).ServeHTTP(response, httptest.NewRequest(method, path, reader))
+	request := httptest.NewRequest(method, path, reader)
+	if len(credentials) == 2 {
+		request.SetBasicAuth(credentials[0], credentials[1])
+	}
+	adaptor.FiberApp(app).ServeHTTP(response, request)
 	return response
 }
 

cmd/control-plane/main.go

@@ -2,7 +2,11 @@ package main
 
 import (
 	"context"
+	"crypto/sha256"
+	"crypto/subtle"
+	"encoding/base64"
 	"errors"
+	"io"
 	"net"
 	"net/url"
 	"os"
@@ -15,6 +19,7 @@ import (
 
 	"git.ipao.vip/rogee/creator-hub/internal/hub"
 	"git.ipao.vip/rogee/creator-hub/internal/phasea"
+	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
 	"github.com/gofiber/fiber/v3"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -22,10 +27,9 @@ import (
 )
 
 type config struct {
-	listenAddr  string
-	webDir      string
-	databaseURL string
-	logLevel    logrus.Level
+	listenAddr, webDir, databaseURL string
+	username, password              string
+	logLevel                        logrus.Level
 }
 
 func main() {
@@ -65,6 +69,9 @@ func newCommand() *cobra.Command {
 				return err
 			}
 			defer hubStore.Close()
+			notify := newAttentionNotifier(os.Stderr)
+			phaseAStore.SetTaskNotifier(notify)
+			hubStore.SetTaskNotifier(notify)
 			logStartup(cfg)
 			heartbeatContext, stopHeartbeat := context.WithCancel(command.Context())
 			heartbeatDone := make(chan struct{})
@@ -72,7 +79,7 @@ func newCommand() *cobra.Command {
 				defer close(heartbeatDone)
 				runtimeLeaseHeartbeat(heartbeatContext, hubStore)
 			}()
-			listenErr := newHandlerWithStores(cfg.webDir, phaseAStore, hubStore).Listen(cfg.listenAddr, fiber.ListenConfig{
+			listenErr := newHandlerWithStores(cfg.webDir, cfg.username, cfg.password, phaseAStore, hubStore).Listen(cfg.listenAddr, fiber.ListenConfig{
 				GracefulContext:       command.Context(),
 				DisableStartupMessage: true,
 			})
@@ -84,6 +91,21 @@ func newCommand() *cobra.Command {
 	return command
 }
 
+func newAttentionNotifier(output io.Writer) taskstate.Notifier {
+	logger := logrus.New()
+	logger.SetFormatter(&logrus.JSONFormatter{})
+	logger.SetOutput(output)
+	return func(transition taskstate.Transition) {
+		if transition.State != "policy_hold" && transition.State != "needs_confirmation" {
+			return
+		}
+		logger.WithFields(logrus.Fields{
+			"service": "control-plane", "notification_channel": "structured_log", "event_type": transition.State,
+			"reason_code": transition.ReasonCode, "account_id": transition.AccountID, "task_id": transition.TaskID,
+		}).Warn("operator attention required")
+	}
+}
+
 func runtimeLeaseHeartbeat(ctx context.Context, store hubStore) {
 	ticker := time.NewTicker(20 * time.Second)
 	defer ticker.Stop()
@@ -116,6 +138,8 @@ func loadConfig() (config, error) {
 	_ = v.BindEnv("web_dir", "WEB_DIR")
 	_ = v.BindEnv("database_url", "DATABASE_URL")
 	_ = v.BindEnv("log_level", "LOG_LEVEL")
+	_ = v.BindEnv("username", "CONTROL_PLANE_USERNAME")
+	_ = v.BindEnv("password", "CONTROL_PLANE_PASSWORD")
 
 	level, err := logrus.ParseLevel(v.GetString("log_level"))
 	if err != nil {
@@ -125,6 +149,8 @@ func loadConfig() (config, error) {
 		listenAddr:  strings.TrimSpace(v.GetString("listen_addr")),
 		webDir:      strings.TrimSpace(v.GetString("web_dir")),
 		databaseURL: strings.TrimSpace(v.GetString("database_url")),
+		username:    strings.TrimSpace(v.GetString("username")),
+		password:    v.GetString("password"),
 		logLevel:    level,
 	}
 	if cfg.listenAddr == "" {
@@ -136,6 +162,12 @@ func loadConfig() (config, error) {
 	if cfg.webDir == "" {
 		return config{}, errors.New("WEB_DIR must not be empty")
 	}
+	if cfg.username == "" || strings.Contains(cfg.username, ":") {
+		return config{}, errors.New("CONTROL_PLANE_USERNAME must be non-empty and contain no colon")
+	}
+	if len(cfg.password) < 16 {
+		return config{}, errors.New("CONTROL_PLANE_PASSWORD must contain at least 16 bytes")
+	}
 	databaseURL, err := url.Parse(cfg.databaseURL)
 	if err != nil || databaseURL.Host == "" || databaseURL.Path == "" || databaseURL.Path == "/" ||
 		(databaseURL.Scheme != "postgres" && databaseURL.Scheme != "postgresql") {
@@ -156,11 +188,11 @@ func validateListenAddr(addr string) error {
 	return nil
 }
 
-func newHandler(webDirectory string) *fiber.App {
-	return newHandlerWithStores(webDirectory, nil, nil)
+func newHandler(webDirectory, username, password string) *fiber.App {
+	return newHandlerWithStores(webDirectory, username, password, nil, nil)
 }
 
-func newHandlerWithStores(webDirectory string, phaseAStore *phasea.Store, hubStore *hub.Store) *fiber.App {
+func newHandlerWithStores(webDirectory, username, password string, phaseAStore *phasea.Store, hubStore *hub.Store) *fiber.App {
 	app := fiber.New(fiber.Config{
 		AppName:   "CreatorHub control plane",
 		BodyLimit: 1 << 20,
@@ -172,6 +204,7 @@ func newHandlerWithStores(webDirectory string, phaseAStore *phasea.Store, hubSto
 		c.Status(fiber.StatusNoContent)
 		return nil
 	})
+	app.Use(authenticate(username, password))
 	if hubStore != nil {
 		registerHub(app, hubStore)
 	}
@@ -182,6 +215,23 @@ func newHandlerWithStores(webDirectory string, phaseAStore *phasea.Store, hubSto
 	return app
 }
 
+func authenticate(username, password string) fiber.Handler {
+	wantUser, wantPassword := sha256.Sum256([]byte(username)), sha256.Sum256([]byte(password))
+	return func(c fiber.Ctx) error {
+		encoded, ok := strings.CutPrefix(c.Get(fiber.HeaderAuthorization), "Basic ")
+		decoded, err := base64.StdEncoding.DecodeString(encoded)
+		user, suppliedPassword, found := strings.Cut(string(decoded), ":")
+		gotUser, gotPassword := sha256.Sum256([]byte(user)), sha256.Sum256([]byte(suppliedPassword))
+		userOK := subtle.ConstantTimeCompare(gotUser[:], wantUser[:])
+		passwordOK := subtle.ConstantTimeCompare(gotPassword[:], wantPassword[:])
+		if !ok || err != nil || !found || userOK != 1 || passwordOK != 1 {
+			c.Set(fiber.HeaderWWWAuthenticate, `Basic realm="CreatorHub", charset="UTF-8"`)
+			return c.Status(fiber.StatusUnauthorized).JSON(map[string]string{"error": "authentication required"})
+		}
+		return c.Next()
+	}
+}
+
 func spaHandler(directory string) fiber.Handler {
 	return func(c fiber.Ctx) error {
 		path, err := url.PathUnescape(strings.TrimPrefix(c.Path(), "/"))

cmd/control-plane/main_test.go

@@ -20,6 +20,7 @@ import (
 
 	"git.ipao.vip/rogee/creator-hub/internal/hub"
 	"git.ipao.vip/rogee/creator-hub/internal/phasea"
+	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
 	"github.com/gofiber/fiber/v3"
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -67,12 +68,116 @@ func TestExecuteContextStopsOnSIGTERM(t *testing.T) {
 }
 
 func TestLoadConfigRejectsInvalidDatabase(t *testing.T) {
+	t.Setenv("CONTROL_PLANE_USERNAME", "operator")
+	t.Setenv("CONTROL_PLANE_PASSWORD", "unit-test-password")
 	t.Setenv("DATABASE_URL", "file:///tmp/creatorhub.db")
 	if _, err := loadConfig(); err == nil {
 		t.Fatal("expected non-Postgres database URL to be rejected")
 	}
 }
 
+func TestLoadConfigRequiresControlPlaneCredentials(t *testing.T) {
+	t.Setenv("CONTROL_PLANE_USERNAME", "")
+	t.Setenv("CONTROL_PLANE_PASSWORD", "")
+	if _, err := loadConfig(); err == nil {
+		t.Fatal("expected missing control-plane credentials to be rejected")
+	}
+}
+
+func TestControlPlaneAuthentication(t *testing.T) {
+	logger := logrus.StandardLogger()
+	previousOutput := logger.Out
+	var logs bytes.Buffer
+	logrus.SetOutput(&logs)
+	t.Cleanup(func() { logrus.SetOutput(previousOutput) })
+
+	directory := t.TempDir()
+	if err := os.WriteFile(filepath.Join(directory, "index.html"), []byte("index"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	app := newHandler(directory, "operator", "unit-test-password")
+
+	health, err := app.Test(httptest.NewRequest(http.MethodGet, "/healthz", nil))
+	if err != nil || health.StatusCode != http.StatusNoContent {
+		t.Fatalf("health check must remain public: status=%d err=%v", health.StatusCode, err)
+	}
+	health.Body.Close()
+
+	for _, path := range []string{
+		"/", "/api/phase-a/accounts", "/api/browsers", "/api/network-exits", "/api/phase-a/tasks", "/api/phase-a/audit",
+	} {
+		request := httptest.NewRequest(http.MethodGet, path, nil)
+		response, err := app.Test(request)
+		if err != nil {
+			t.Fatal(err)
+		}
+		response.Body.Close()
+		if response.StatusCode != http.StatusUnauthorized || response.Header.Get("WWW-Authenticate") == "" {
+			t.Fatalf("GET %s was not protected: status=%d", path, response.StatusCode)
+		}
+	}
+
+	for name, test := range map[string]struct {
+		user, password string
+		want           int
+	}{
+		"valid":          {"operator", "unit-test-password", http.StatusOK},
+		"wrong user":     {"other", "unit-test-password", http.StatusUnauthorized},
+		"wrong password": {"operator", "credential-must-not-be-logged", http.StatusUnauthorized},
+	} {
+		t.Run(name, func(t *testing.T) {
+			request := httptest.NewRequest(http.MethodGet, "/", nil)
+			request.SetBasicAuth(test.user, test.password)
+			response, err := app.Test(request)
+			if err != nil {
+				t.Fatal(err)
+			}
+			response.Body.Close()
+			if response.StatusCode != test.want {
+				t.Fatalf("status=%d want=%d", response.StatusCode, test.want)
+			}
+		})
+	}
+	if strings.Contains(logs.String(), "credential-must-not-be-logged") {
+		t.Fatalf("authentication credential reached logs: %s", logs.String())
+	}
+}
+
+func TestOperatorNotificationFiltersAndRedacts(t *testing.T) {
+	previousLevel := logrus.GetLevel()
+	t.Cleanup(func() { logrus.SetLevel(previousLevel) })
+	var output bytes.Buffer
+	notify := newAttentionNotifier(&output)
+
+	notify(taskstate.Transition{State: "succeeded", ReasonCode: "task_succeeded", AccountID: "account-a", TaskID: "task-a"})
+	if output.Len() != 0 {
+		t.Fatalf("successful task emitted a notification: %s", output.String())
+	}
+
+	for _, level := range []logrus.Level{logrus.ErrorLevel, logrus.FatalLevel, logrus.PanicLevel} {
+		for _, test := range []struct{ state, reason string }{
+			{"policy_hold", "account_paused"}, {"needs_confirmation", "exit_unhealthy"},
+		} {
+			output.Reset()
+			logrus.SetLevel(level)
+			notify(taskstate.Transition{State: test.state, ReasonCode: test.reason, AccountID: "account-a", TaskID: "task-a"})
+			var entry map[string]any
+			if err := json.Unmarshal(output.Bytes(), &entry); err != nil {
+				t.Fatal(err)
+			}
+			if entry["event_type"] != test.state || entry["reason_code"] != test.reason ||
+				entry["notification_channel"] != "structured_log" || entry["account_id"] != "account-a" || entry["task_id"] != "task-a" {
+				t.Fatalf("unexpected notification at LOG_LEVEL=%s: %#v", level, entry)
+			}
+			for _, forbidden := range []string{"password", "authorization", "credential", "token", "secret"} {
+				if strings.Contains(strings.ToLower(output.String()), forbidden) {
+					t.Fatalf("notification contained sensitive field %q: %s", forbidden, output.String())
+				}
+			}
+		}
+	}
+}
+
 func TestPhaseAAccountRequestRejectsSecretsAndUnknownFields(t *testing.T) {
 	for name, body := range map[string]string{
 		"raw password":  `{"id":"account-a","platform":"mock","platform_account_key":"a","authorization_kind":"owned","credential_reference":{"id":"credential-a","provider":"os_keyring","key":"creatorhub/account-a"},"password":"secret"}`,
@@ -116,17 +221,38 @@ func TestPhaseAAccountHTTPWorkflowRedactsSecrets(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if err := hubStore.Close(); err != nil {
+	t.Cleanup(func() { _ = hubStore.Close() })
+
+	webDirectory := t.TempDir()
+	if err := os.WriteFile(filepath.Join(webDirectory, "index.html"), []byte("index"), 0o600); err != nil {
 		t.Fatal(err)
 	}
-
+	protectedApp := newHandlerWithStores(webDirectory, "operator", "unit-test-password", store, hubStore)
+	for _, route := range []struct{ method, path string }{
+		{http.MethodGet, "/api/phase-a/accounts"}, {http.MethodPost, "/api/phase-a/accounts"},
+		{http.MethodPost, "/api/phase-a/accounts/account-http/pause"}, {http.MethodGet, "/api/phase-a/tasks"},
+		{http.MethodPost, "/api/phase-a/mock/execute"}, {http.MethodGet, "/api/phase-a/audit"},
+		{http.MethodGet, "/api/browsers"}, {http.MethodPost, "/api/browsers/account-http/start"},
+		{http.MethodDelete, "/api/browsers/account-http"}, {http.MethodPut, "/api/browser-images/1"},
+		{http.MethodPost, "/api/network-exits/exit-http/check"}, {http.MethodPost, "/api/network-exits/exit-http/disable"},
+	} {
+		if response := do(protectedApp, route.method, route.path, ""); response.Code != http.StatusUnauthorized {
+			t.Fatalf("unauthenticated %s %s returned %d", route.method, route.path, response.Code)
+		}
+	}
+	for _, path := range []string{"/api/phase-a/accounts", "/api/browsers", "/api/phase-a/audit"} {
+		if response := do(protectedApp, http.MethodGet, path, "", "operator", "unit-test-password"); response.Code != http.StatusOK {
+			t.Fatalf("authenticated GET %s returned %d: %s", path, response.Code, response.Body.String())
+		}
+	}
 	app := fiber.New()
+	app.Use(authenticate("operator", "unit-test-password"))
 	registerPhaseA(app, store, nil)
 	credentialKey := "creatorhub/phase-http-secret"
 	accountBody := `{"id":"account-http","platform":"mock","platform_account_key":"phase-http","authorization_kind":"owned","credential_reference":{"id":"credential-http","provider":"os_keyring","key":"` + credentialKey + `"}}`
 	request := func(method, path, body string, wantStatus int) *httptest.ResponseRecorder {
 		t.Helper()
-		response := do(app, method, path, body)
+		response := do(app, method, path, body, "operator", "unit-test-password")
 		if response.Code != wantStatus {
 			t.Fatalf("%s %s returned %d, want %d: %s", method, path, response.Code, wantStatus, response.Body.String())
 		}
@@ -204,7 +330,7 @@ func TestSPAServesFileAndIndexFallback(t *testing.T) {
 	if err := os.WriteFile(filepath.Join(directory, "..", "outside-secret.txt"), []byte("secret"), 0o600); err != nil {
 		t.Fatal(err)
 	}
-	app := newHandler(directory)
+	app := newHandler(directory, "operator", "unit-test-password")
 	for path, want := range map[string]string{
 		"/app.js":                      "asset",
 		"/hello%20world.js":            "space",
@@ -212,7 +338,9 @@ func TestSPAServesFileAndIndexFallback(t *testing.T) {
 		"/%2e%2e%2foutside-secret.txt": "index",
 		"/dashboard":                   "index",
 	} {
-		response, err := app.Test(httptest.NewRequest(http.MethodGet, path, nil))
+		request := httptest.NewRequest(http.MethodGet, path, nil)
+		request.SetBasicAuth("operator", "unit-test-password")
+		response, err := app.Test(request)
 		if err != nil {
 			t.Fatal(err)
 		}

compose.yaml

@@ -3,6 +3,8 @@ services:
     build: .
     environment:
       DATABASE_URL: postgres://creatorhub@postgres/creatorhub?sslmode=disable
+      CONTROL_PLANE_USERNAME: ${CONTROL_PLANE_USERNAME:?required}
+      CONTROL_PLANE_PASSWORD: ${CONTROL_PLANE_PASSWORD:?required}
     ports:
       - "${CREATORHUB_PORT:-8080}:8080"
     read_only: true

docs/deployment.md

@@ -1,6 +1,6 @@
 # CreatorHub 部署
 
-本文档适用于当前阶段 A：在一台 Linux 主机上通过 Docker Compose 部署。开发阶段不设访问限制，安全由部署者自行把控。
+本文档适用于当前阶段 A：在一台 Linux 主机上通过 Docker Compose 部署。控制面使用单用户 HTTP Basic Auth；当前不提供 RBAC 或多租户隔离。
 
 ## 部署内容
 
@@ -29,6 +29,8 @@
 test -S /var/run/docker.sock
 docker info >/dev/null
 docker compose version
+: "${CONTROL_PLANE_USERNAME:?set CONTROL_PLANE_USERNAME or fill .env}"
+: "${CONTROL_PLANE_PASSWORD:?set CONTROL_PLANE_PASSWORD or fill .env}"
 docker compose config --quiet
 ```
 
@@ -40,6 +42,8 @@ docker compose config --quiet
 export DOCKER_GID="$(stat -c '%g' /var/run/docker.sock)"
 export CREATORHUB_PORT=8080
 export GATEWAY_TOKEN="$(openssl rand -hex 24)"   # 亦可在 .env 中设置
+export CONTROL_PLANE_USERNAME=creatorhub
+export CONTROL_PLANE_PASSWORD="$(openssl rand -hex 24)"
 
 docker compose config --quiet
 docker compose up --detach --build
@@ -64,6 +68,7 @@ curl --fail --silent --show-error \
 
 curl --fail --silent --show-error \
   --retry 30 --retry-delay 2 --retry-connrefused \
+  --user "${CONTROL_PLANE_USERNAME}:${CONTROL_PLANE_PASSWORD}" \
   "http://127.0.0.1:${CREATORHUB_PORT}/api/browsers" >/dev/null
 
 docker compose exec -T postgres \
@@ -91,8 +96,10 @@ Compose 部署时通常只需设置以下宿主机变量：
 | `CREATORHUB_PORT` | `8080` | 控制面宿主机端口，局域网可访问 |
 | `DOCKER_GID` | `999` | Docker socket 的宿主机组 ID；必须按实际值设置 |
 | `GATEWAY_TOKEN` | `dev-creatorhub-gateway-token` | 网关与控制面共享的 Bearer 令牌；生产须改为随机值，并同步填入网关注册表单 |
+| `CONTROL_PLANE_USERNAME` | 无（必填） | 控制面唯一用户；不能包含冒号 |
+| `CONTROL_PLANE_PASSWORD` | 无（必填） | 控制面密码，至少 16 字节；使用随机值 |
 
-服务本身支持并校验以下环境变量；`compose.yaml` 已提供当前部署所需的值：
+服务本身支持并校验以下环境变量；`compose.yaml` 会在控制面凭据缺失或为空时拒绝渲染：
 
 | 服务 | 变量 | 当前 Compose 值 |
 | --- | --- | --- |
@@ -100,6 +107,8 @@ Compose 部署时通常只需设置以下宿主机变量：
 | `creator-hub` | `WEB_DIR` | 镜像内固定为 `/app/web` |
 | `creator-hub` | `DATABASE_URL` | `postgres://creatorhub@postgres/creatorhub?sslmode=disable` |
 | `creator-hub` | `LOG_LEVEL` | 默认 `info` |
+| `creator-hub` | `CONTROL_PLANE_USERNAME` | 必填；HTTP Basic Auth 用户名 |
+| `creator-hub` | `CONTROL_PLANE_PASSWORD` | 必填且至少 16 字节；不会写入日志或响应 |
 | `docker-gateway` | `LISTEN_ADDR` | 默认 `:8081` |
 | `docker-gateway` | `DOCKER_SOCKET` | 默认值和 Compose 挂载均固定为 `/var/run/docker.sock`；不能只覆盖环境变量 |
 | `docker-gateway` | `BROWSER_NETWORK` | `creatorhub_browser` |
@@ -108,6 +117,16 @@ Compose 部署时通常只需设置以下宿主机变量：
 
 不要把凭据写入仓库或 Compose 文件。
 
+### P0-lite 停机通知
+
+当前只使用 `creator-hub` 的专用 Logrus JSON logger 作为通知渠道；它固定输出警告，不继承业务 `LOG_LEVEL`。选择它是因为 Compose 已可靠收集服务日志，不需要新增外部账号、凭据、网络重试或通知依赖。仅 `policy_hold` 和 `needs_confirmation` 会产生 `operator attention required`，字段限定为 `event_type`、`reason_code`、账号/任务 ID；不会包含请求头、Secret 引用或凭据值。运维可用下列命令接入现有日志采集或人工查看：
+
+```bash
+docker compose logs creator-hub | grep 'operator attention required'
+```
+
+该渠道是单实例 P0-lite 能力，不保证外部送达、升级或确认回执；只有出现明确的多渠道/送达需求时才增加 webhook 或消息平台。
+
 ## 更新与回滚
 
 更新前记录当前版本并备份数据库：

internal/hub/environment.go

@@ -11,6 +11,8 @@ import (
 	"regexp"
 	"strings"
 	"time"
+
+	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
 )
 
 var exitIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]{0,127}$`)
@@ -260,12 +262,14 @@ func (s *Store) RecordNetworkExitCheck(ctx context.Context, id string, observati
 		WHERE id = $1`, id, observation.PublicIP, observation.Region, status, reason, version); err != nil {
 		return NetworkExit{}, "persistence_failed", errors.New("record network exit check")
 	}
+	var transitions []taskstate.Transition
 	if changed {
-		if err := invalidateAccountsForExit(ctx, tx, id); err != nil {
+		transitions, err = invalidateAccountsForExit(ctx, tx, id)
+		if err != nil {
 			return NetworkExit{}, "persistence_failed", err
 		}
 	}
-	if err := commitHub(tx); err != nil {
+	if err := s.commitTransitions(tx, transitions); err != nil {
 		return NetworkExit{}, "persistence_failed", err
 	}
 	exit, err := s.GetNetworkExit(ctx, id)
@@ -301,6 +305,7 @@ func (s *Store) DisableNetworkExit(ctx context.Context, id string) (NetworkExit,
 	if err := tx.QueryRowContext(ctx, `SELECT health_status FROM network_exit WHERE id = $1 FOR UPDATE`, id).Scan(&oldStatus); err != nil {
 		return NetworkExit{}, rowError(err)
 	}
+	var transitions []taskstate.Transition
 	if oldStatus != "disabled" {
 		if _, err := tx.ExecContext(ctx, `
 			UPDATE network_exit SET health_status = 'disabled', last_check_reason = 'exit_disabled',
@@ -308,18 +313,19 @@ func (s *Store) DisableNetworkExit(ctx context.Context, id string) (NetworkExit,
 			WHERE id = $1`, id); err != nil {
 			return NetworkExit{}, errors.New("disable network exit")
 		}
-		if err := invalidateAccountsForExit(ctx, tx, id); err != nil {
+		transitions, err = invalidateAccountsForExit(ctx, tx, id)
+		if err != nil {
 			return NetworkExit{}, err
 		}
 	}
-	if err := commitHub(tx); err != nil {
+	if err := s.commitTransitions(tx, transitions); err != nil {
 		return NetworkExit{}, err
 	}
 	return s.GetNetworkExit(ctx, id)
 }
 
-func invalidateAccountsForExit(ctx context.Context, tx *sql.Tx, exitID string) error {
-	if _, err := tx.ExecContext(ctx, `
+func invalidateAccountsForExit(ctx context.Context, tx *sql.Tx, exitID string) ([]taskstate.Transition, error) {
+	rows, err := tx.QueryContext(ctx, `
 		WITH changed AS (
 			UPDATE social_account account SET status = 'paused', paused_at = COALESCE(paused_at, now()),
 				version = account.version + 1, updated_at = now()
@@ -332,14 +338,18 @@ func invalidateAccountsForExit(ctx context.Context, tx *sql.Tx, exitID string) e
 				hold_reason = 'exit_unhealthy', verification_result = NULL, verified_at = NULL, verified_by = NULL,
 				lease_owner = NULL, lease_until = NULL, updated_at = now()
 			FROM changed WHERE task.account_id = changed.id AND task.state IN ('queued', 'executing')
-			RETURNING task.current_attempt_id, task.state
+			RETURNING task.current_attempt_id, task.state, task.hold_reason, task.account_id, task.id
+		), finished AS (
+			UPDATE execution_attempt attempt SET finished_at = now(), outcome = 'uncertain'
+			FROM held WHERE held.state = 'needs_confirmation' AND attempt.id = held.current_attempt_id
+				AND attempt.finished_at IS NULL
+			RETURNING attempt.id
 		)
-		UPDATE execution_attempt attempt SET finished_at = now(), outcome = 'uncertain'
-		FROM held WHERE held.state = 'needs_confirmation' AND attempt.id = held.current_attempt_id
-			AND attempt.finished_at IS NULL`, exitID); err != nil {
-		return errors.New("invalidate network exit accounts")
+		SELECT state, hold_reason, account_id, id FROM held`, exitID)
+	if err != nil {
+		return nil, errors.New("invalidate network exit accounts")
 	}
-	return nil
+	return scanTransitions(rows, "read invalidated network exit tasks")
 }
 
 func (s *Store) CreateBoundEnv(ctx context.Context, env Env, accountID, exitID string) (EnvironmentContext, bool, error) {

internal/hub/migration_test.go

@@ -7,10 +7,12 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"slices"
 	"testing"
 	"time"
 
 	"git.ipao.vip/rogee/creator-hub/internal/phasea"
+	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
 )
 
 func TestUnifiedAccountMigration(t *testing.T) {
@@ -294,6 +296,8 @@ func TestUnifiedAccountMigration(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
+		var notifications []taskstate.Transition
+		store.SetTaskNotifier(func(transition taskstate.Transition) { notifications = append(notifications, transition) })
 		if err := store.UpgradeEnv(ctx, "mapped", "2"); err != nil {
 			t.Fatal(err)
 		}
@@ -304,6 +308,14 @@ func TestUnifiedAccountMigration(t *testing.T) {
 		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-task' AND state = 'policy_hold' AND hold_reason = 'binding_version_changed'`, 1)
 		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-executing' AND state = 'needs_confirmation' AND hold_reason = 'task_result_uncertain' AND lease_owner IS NULL`, 1)
 		assertDatabaseCount(t, db, `SELECT count(*) FROM execution_attempt WHERE id = 'upgrade-attempt' AND outcome = 'uncertain' AND finished_at IS NOT NULL`, 1)
+		for _, want := range []taskstate.Transition{
+			{State: "policy_hold", ReasonCode: "binding_version_changed", AccountID: "mapped", TaskID: "upgrade-task"},
+			{State: "needs_confirmation", ReasonCode: "task_result_uncertain", AccountID: "mapped", TaskID: "upgrade-executing"},
+		} {
+			if !slices.Contains(notifications, want) {
+				t.Fatalf("missing upgrade transition %+v in %+v", want, notifications)
+			}
+		}
 		phaseAStore, err = phasea.Open(ctx, testURL)
 		if err != nil {
 			t.Fatal(err)

internal/hub/store.go

@@ -15,6 +15,7 @@ import (
 	"time"
 	"unicode/utf8"
 
+	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
 	"github.com/jackc/pgx/v5/pgconn"
 	_ "github.com/jackc/pgx/v5/stdlib"
 )
@@ -69,7 +70,10 @@ var (
 	imageRefPattern     = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:/@-]{0,300}$`)
 )
 
-type Store struct{ db *sql.DB }
+type Store struct {
+	db     *sql.DB
+	notify taskstate.Notifier
+}
 
 // Gateway 是平台注册的 docker-gateway 实例；Token 由平台生成，明文存储供页面复制（开发阶段约定）。
 type Gateway struct {
@@ -122,6 +126,17 @@ func Open(ctx context.Context, databaseURL string) (*Store, error) {
 
 func (s *Store) Close() error { return s.db.Close() }
 
+func (s *Store) SetTaskNotifier(notify taskstate.Notifier) { s.notify = notify }
+
+func (s *Store) notifyTransitions(transitions []taskstate.Transition) {
+	if s.notify == nil {
+		return
+	}
+	for _, transition := range transitions {
+		s.notify(transition)
+	}
+}
+
 func (s *Store) migrate(ctx context.Context) error {
 	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
@@ -393,7 +408,7 @@ func (s *Store) UpgradeEnv(ctx context.Context, alias, version string) error {
 		WHERE browser_env_alias = $1`, alias); err != nil {
 		return errors.New("version upgraded environment binding")
 	}
-	if _, err := tx.ExecContext(ctx, `
+	rows, err := tx.QueryContext(ctx, `
 		WITH changed AS (
 			UPDATE social_account account
 			SET status = 'paused', paused_at = now(), version = account.version + 1, updated_at = now()
@@ -407,14 +422,22 @@ func (s *Store) UpgradeEnv(ctx context.Context, alias, version string) error {
 				verification_result = NULL, verified_at = NULL, verified_by = NULL,
 				lease_owner = NULL, lease_until = NULL, updated_at = now()
 			FROM changed WHERE task.account_id = changed.id AND task.state IN ('queued', 'executing')
-			RETURNING task.current_attempt_id, task.state
+			RETURNING task.current_attempt_id, task.state, task.hold_reason, task.account_id, task.id
+		), finished AS (
+			UPDATE execution_attempt attempt SET finished_at = now(), outcome = 'uncertain'
+			FROM held WHERE held.state = 'needs_confirmation' AND attempt.id = held.current_attempt_id
+				AND attempt.finished_at IS NULL
+			RETURNING attempt.id
 		)
-		UPDATE execution_attempt attempt SET finished_at = now(), outcome = 'uncertain'
-		FROM held WHERE held.state = 'needs_confirmation' AND attempt.id = held.current_attempt_id
-			AND attempt.finished_at IS NULL`, alias); err != nil {
+		SELECT state, hold_reason, account_id, id FROM held`, alias)
+	if err != nil {
 		return errors.New("invalidate upgraded environment account")
 	}
-	return commitHub(tx)
+	transitions, err := scanTransitions(rows, "read upgraded environment tasks")
+	if err != nil {
+		return err
+	}
+	return s.commitTransitions(tx, transitions)
 }
 
 func (s *Store) DeleteEnv(ctx context.Context, alias string) error {
@@ -474,6 +497,30 @@ func commitHub(tx *sql.Tx) error {
 	return nil
 }
 
+func (s *Store) commitTransitions(tx *sql.Tx, transitions []taskstate.Transition) error {
+	if err := commitHub(tx); err != nil {
+		return err
+	}
+	s.notifyTransitions(transitions)
+	return nil
+}
+
+func scanTransitions(rows *sql.Rows, failure string) ([]taskstate.Transition, error) {
+	defer rows.Close()
+	var transitions []taskstate.Transition
+	for rows.Next() {
+		var transition taskstate.Transition
+		if err := rows.Scan(&transition.State, &transition.ReasonCode, &transition.AccountID, &transition.TaskID); err != nil {
+			return nil, errors.New(failure)
+		}
+		transitions = append(transitions, transition)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, errors.New(failure)
+	}
+	return transitions, nil
+}
+
 func rowError(err error) error {
 	if errors.Is(err, sql.ErrNoRows) {
 		return ErrNotFound

internal/hub/store_test.go

@@ -6,9 +6,12 @@ import (
 	"errors"
 	"os"
 	"reflect"
+	"slices"
 	"strings"
 	"testing"
 	"time"
+
+	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
 )
 
 func TestFingerprintArgsFollowUpstreamCommandLineContract(t *testing.T) {
@@ -250,6 +253,8 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) {
 	ctx := context.Background()
 	store := openFullyMigratedHub(t, ctx, isolatedDatabaseURL(t, databaseURL))
 	t.Cleanup(func() { _ = store.Close() })
+	var notifications []taskstate.Transition
+	store.SetTaskNotifier(func(transition taskstate.Transition) { notifications = append(notifications, transition) })
 	if _, err := store.db.ExecContext(ctx, `TRUNCATE audit_event, runtime_instance, environment_binding, network_exit,
 		social_account, credential_reference, browser_env, browser_image, gateway CASCADE`); err != nil {
 		t.Fatal(err)
@@ -532,6 +537,42 @@ func TestNetworkExitBindingRuntimeAndAuditWorkflow(t *testing.T) {
 	assertDatabaseCount(t, store.db, `SELECT count(*) FROM operation_task WHERE id = 'exit-hold-queued' AND state = 'policy_hold' AND hold_reason = 'exit_unhealthy'`, 1)
 	assertDatabaseCount(t, store.db, `SELECT count(*) FROM operation_task WHERE id = 'exit-hold-executing' AND state = 'needs_confirmation' AND hold_reason = 'exit_unhealthy' AND lease_owner IS NULL`, 1)
 	assertDatabaseCount(t, store.db, `SELECT count(*) FROM execution_attempt WHERE id = 'exit-hold-attempt' AND outcome = 'uncertain' AND finished_at IS NOT NULL`, 1)
+	for _, want := range []taskstate.Transition{
+		{State: "policy_hold", ReasonCode: "exit_unhealthy", AccountID: "account-a", TaskID: "exit-hold-queued"},
+		{State: "needs_confirmation", ReasonCode: "exit_unhealthy", AccountID: "account-a", TaskID: "exit-hold-executing"},
+	} {
+		if !slices.Contains(notifications, want) {
+			t.Fatalf("missing network exit transition %+v in %+v", want, notifications)
+		}
+	}
+	if err := store.db.QueryRowContext(ctx, `SELECT version FROM social_account WHERE id = 'account-a'`).Scan(&accountVersion); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.db.ExecContext(ctx, `INSERT INTO content_draft (id, account_id, version, content) VALUES ('exit-disable-draft', 'account-a', 1, 'test')`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.db.ExecContext(ctx, `INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
+		VALUES ('exit-disable-confirmation', 'account-a', $1, 'exit-disable-draft', 1, 1)`, accountVersion); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.db.ExecContext(ctx, `INSERT INTO operation_task
+			(id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version, state, lease_owner, lease_until)
+		VALUES
+			('exit-disable-queued', 'exit-disable-queued-key', 'account-a', $1, 'exit-disable-draft', 1, 'exit-disable-confirmation', 1, 'queued', NULL, NULL),
+			('exit-disable-executing', 'exit-disable-executing-key', 'account-a', $1, 'exit-disable-draft', 1, 'exit-disable-confirmation', 1, 'executing', 'worker-disabled', now() + interval '1 minute')`, accountVersion); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.DisableNetworkExit(ctx, newGeneration.Exit.ID); err != nil {
+		t.Fatal(err)
+	}
+	for _, want := range []taskstate.Transition{
+		{State: "policy_hold", ReasonCode: "exit_unhealthy", AccountID: "account-a", TaskID: "exit-disable-queued"},
+		{State: "needs_confirmation", ReasonCode: "exit_unhealthy", AccountID: "account-a", TaskID: "exit-disable-executing"},
+	} {
+		if !slices.Contains(notifications, want) {
+			t.Fatalf("missing disabled exit transition %+v in %+v", want, notifications)
+		}
+	}
 	var auditText string
 	if err := store.db.QueryRowContext(ctx, `SELECT string_agg(row_to_json(event)::text, '') FROM audit_event event`).Scan(&auditText); err != nil {
 		t.Fatal(err)

internal/phasea/store.go

@@ -14,6 +14,7 @@ import (
 	"strings"
 	"time"
 
+	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
 	"github.com/jackc/pgx/v5/pgconn"
 	_ "github.com/jackc/pgx/v5/stdlib"
 )
@@ -32,7 +33,10 @@ var (
 	eventPattern         = regexp.MustCompile(`^[a-z0-9_]{1,64}$`)
 )
 
-type Store struct{ db *sql.DB }
+type Store struct {
+	db     *sql.DB
+	notify taskstate.Notifier
+}
 
 type Account struct {
 	ID                  string              `json:"id"`
@@ -207,6 +211,17 @@ func Open(ctx context.Context, databaseURL string) (*Store, error) {
 
 func (s *Store) Close() error { return s.db.Close() }
 
+func (s *Store) SetTaskNotifier(notify taskstate.Notifier) { s.notify = notify }
+
+func (s *Store) notifyTransitions(transitions []taskstate.Transition) {
+	if s.notify == nil {
+		return
+	}
+	for _, transition := range transitions {
+		s.notify(transition)
+	}
+}
+
 func (s *Store) migrate(ctx context.Context) error {
 	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
@@ -985,19 +1000,18 @@ func (s *Store) disableAccount(ctx context.Context, accountID string, revoke boo
 	if err != nil {
 		return err
 	}
-	if unchanged {
-		return commit(tx)
+	if !unchanged {
+		eventType := "account_paused"
+		if revoke {
+			eventType = "account_revoked"
+		}
+		if err := appendAudit(ctx, tx, eventType, reason, accountID, "", 0, "", "", map[string]any{
+			"account_version": version, "tasks_held": len(held), "attempts_interrupted": len(interrupted),
+		}); err != nil {
+			return err
+		}
 	}
-	eventType := "account_paused"
-	if revoke {
-		eventType = "account_revoked"
-	}
-	if err := appendAudit(ctx, tx, eventType, reason, accountID, "", 0, "", "", map[string]any{
-		"account_version": version, "tasks_held": held, "attempts_interrupted": interrupted,
-	}); err != nil {
-		return err
-	}
-	return commit(tx)
+	return s.commitTransitions(tx, append(held, interrupted...))
 }
 
 func (s *Store) ResumeAccount(ctx context.Context, accountID string) error {
@@ -1049,19 +1063,19 @@ func (s *Store) ResumeAccount(ctx context.Context, accountID string) error {
 	return commit(tx)
 }
 
-func holdQueuedTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) (int64, error) {
-	result, err := tx.ExecContext(ctx, `
+func holdQueuedTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) ([]taskstate.Transition, error) {
+	rows, err := tx.QueryContext(ctx, `
 		UPDATE operation_task SET state = 'policy_hold', hold_reason = $2,
 			verification_result = NULL, verified_at = NULL, verified_by = NULL, updated_at = now()
-		WHERE account_id = $1 AND state = 'queued'`, accountID, reason)
+		WHERE account_id = $1 AND state = 'queued'
+		RETURNING state, hold_reason, account_id, id`, accountID, reason)
 	if err != nil {
-		return 0, errors.New("hold queued account tasks")
+		return nil, errors.New("hold queued account tasks")
 	}
-	held, _ := result.RowsAffected()
-	return held, nil
+	return scanTransitions(rows, "read held account tasks")
 }
 
-func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) (int64, error) {
+func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason string) ([]taskstate.Transition, error) {
 	rows, err := tx.QueryContext(ctx, `
 		UPDATE operation_task SET state = 'needs_confirmation', hold_reason = $2,
 			verification_result = NULL, verified_at = NULL, verified_by = NULL,
@@ -1069,7 +1083,7 @@ func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason
 		WHERE account_id = $1 AND state = 'executing'
 		RETURNING id, current_attempt_id, confirmation_id, confirmation_version`, accountID, reason)
 	if err != nil {
-		return 0, errors.New("interrupt executing account tasks")
+		return nil, errors.New("interrupt executing account tasks")
 	}
 	type interruptedTask struct {
 		taskID, attemptID, confirmationID string
@@ -1082,28 +1096,32 @@ func interruptExecutingTasks(ctx context.Context, tx *sql.Tx, accountID, reason
 		var confirmationVersion sql.NullInt64
 		if err := rows.Scan(&task.taskID, &attemptID, &confirmationID, &confirmationVersion); err != nil {
 			rows.Close()
-			return 0, errors.New("read interrupted account task")
+			return nil, errors.New("read interrupted account task")
 		}
 		task.attemptID, task.confirmationID, task.confirmationVersion = attemptID.String, confirmationID.String, confirmationVersion.Int64
 		tasks = append(tasks, task)
 	}
 	if err := rows.Close(); err != nil {
-		return 0, errors.New("close interrupted account task rows")
+		return nil, errors.New("close interrupted account task rows")
 	}
+	transitions := make([]taskstate.Transition, 0, len(tasks))
 	for _, task := range tasks {
 		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 0, errors.New("mark interrupted attempt uncertain")
+				return nil, errors.New("mark interrupted attempt uncertain")
 			}
 		}
 		if err := appendAudit(ctx, tx, "task_interrupted", reason, accountID, task.confirmationID,
 			task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil {
-			return 0, err
+			return nil, err
 		}
+		transitions = append(transitions, taskstate.Transition{
+			State: "needs_confirmation", ReasonCode: reason, AccountID: accountID, TaskID: task.taskID,
+		})
 	}
-	return int64(len(tasks)), nil
+	return transitions, nil
 }
 
 func (s *Store) VerifyTask(ctx context.Context, taskID, result string) error {
@@ -1300,7 +1318,7 @@ func (s *Store) CancelTask(ctx context.Context, taskID string) error {
 		return errors.New("begin cancel transaction")
 	}
 	defer tx.Rollback()
-	var accountID, state string
+	var accountID, state, holdReason string
 	var attemptID sql.NullString
 	var confirmationID sql.NullString
 	var confirmationVersion sql.NullInt64
@@ -1310,8 +1328,8 @@ func (s *Store) CancelTask(ctx context.Context, taskID string) error {
 			verification_result = NULL, verified_at = NULL, verified_by = NULL,
 			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)
+		RETURNING account_id, state, COALESCE(hold_reason, ''), current_attempt_id, confirmation_id, confirmation_version`, taskID).Scan(
+		&accountID, &state, &holdReason, &attemptID, &confirmationID, &confirmationVersion)
 	if err != nil {
 		return rowError(err)
 	}
@@ -1325,7 +1343,7 @@ func (s *Store) CancelTask(ctx context.Context, taskID string) error {
 	if err := appendAudit(ctx, tx, "task_cancel_requested", "task_cancel_requested", accountID, confirmationID.String, confirmationVersion.Int64, attemptID.String, taskID, map[string]string{"state": state}); err != nil {
 		return err
 	}
-	return commit(tx)
+	return s.commitTransitions(tx, []taskstate.Transition{{State: state, ReasonCode: holdReason, AccountID: accountID, TaskID: taskID}})
 }
 
 func (s *Store) ExecuteMock(ctx context.Context, workerID, outcome string) (Execution, error) {
@@ -1345,12 +1363,15 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
 		return Execution{}, errors.New("begin claim transaction")
 	}
 	defer tx.Rollback()
-	if err := quarantineExpired(ctx, tx); err != nil {
+	expired, err := quarantineExpired(ctx, tx)
+	if err != nil {
 		return Execution{}, err
 	}
-	if err := quarantineInvalid(ctx, tx); err != nil {
+	invalid, err := quarantineInvalid(ctx, tx)
+	if err != nil {
 		return Execution{}, err
 	}
+	transitions := append(expired, invalid...)
 	var execution Execution
 	err = tx.QueryRowContext(ctx, `
 		WITH candidate AS (
@@ -1381,7 +1402,7 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
 		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 {
+		if err := s.commitTransitions(tx, transitions); err != nil {
 			return Execution{}, err
 		}
 		return Execution{}, nil
@@ -1401,7 +1422,7 @@ func (s *Store) claim(ctx context.Context, workerID string) (Execution, error) {
 	if err := appendAudit(ctx, tx, "task_claimed", "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 {
+	if err := s.commitTransitions(tx, transitions); err != nil {
 		return Execution{}, err
 	}
 	return execution, nil
@@ -1498,14 +1519,16 @@ func (s *Store) complete(ctx context.Context, workerID string, execution Executi
 	if err := appendAudit(ctx, tx, "task_finished", reason, 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 {
+	if err := s.commitTransitions(tx, []taskstate.Transition{{
+		State: state, ReasonCode: holdReason, AccountID: execution.AccountID, TaskID: execution.TaskID,
+	}}); err != nil {
 		return Execution{}, err
 	}
 	execution.State = state
 	return execution, nil
 }
 
-func quarantineExpired(ctx context.Context, tx *sql.Tx) error {
+func quarantineExpired(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition, error) {
 	rows, err := tx.QueryContext(ctx, `
 		UPDATE operation_task SET state = 'needs_confirmation', hold_reason = 'execution_lease_expired',
 			verification_result = NULL, verified_at = NULL, verified_by = NULL,
@@ -1513,7 +1536,7 @@ func quarantineExpired(ctx context.Context, tx *sql.Tx) error {
 		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")
+		return nil, errors.New("quarantine expired tasks")
 	}
 	type expiredTask struct {
 		taskID, accountID, attemptID, confirmationID string
@@ -1527,27 +1550,31 @@ func quarantineExpired(ctx context.Context, tx *sql.Tx) error {
 		var confirmationVersion sql.NullInt64
 		if err := rows.Scan(&taskID, &accountID, &attemptID, &confirmationID, &confirmationVersion); err != nil {
 			rows.Close()
-			return errors.New("read expired task")
+			return nil, 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")
+		return nil, errors.New("close expired task rows")
 	}
+	transitions := make([]taskstate.Transition, 0, len(expired))
 	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")
+				return nil, errors.New("mark expired attempt uncertain")
 			}
 		}
 		if err := appendAudit(ctx, tx, "task_lease_expired", "execution_lease_expired", task.accountID, task.confirmationID, task.confirmationVersion, task.attemptID, task.taskID, nil); err != nil {
-			return err
+			return nil, err
 		}
+		transitions = append(transitions, taskstate.Transition{
+			State: "needs_confirmation", ReasonCode: "execution_lease_expired", AccountID: task.accountID, TaskID: task.taskID,
+		})
 	}
-	return nil
+	return transitions, nil
 }
 
-func quarantineInvalid(ctx context.Context, tx *sql.Tx) error {
+func quarantineInvalid(ctx context.Context, tx *sql.Tx) ([]taskstate.Transition, error) {
 	rows, err := tx.QueryContext(ctx, `
 		WITH invalid AS (
 			SELECT t.id,
@@ -1606,7 +1633,7 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) error {
 		FROM invalid WHERE task.id = invalid.id
 		RETURNING task.id, task.account_id, task.confirmation_id, task.confirmation_version, task.state, invalid.reason_code`)
 	if err != nil {
-		return errors.New("quarantine invalid tasks")
+		return nil, errors.New("quarantine invalid tasks")
 	}
 	type invalidTask struct {
 		taskID, accountID, confirmationID, state, reasonCode string
@@ -1619,19 +1646,39 @@ func quarantineInvalid(ctx context.Context, tx *sql.Tx) error {
 		var confirmationVersion sql.NullInt64
 		if err := rows.Scan(&taskID, &accountID, &confirmationID, &confirmationVersion, &state, &reasonCode); err != nil {
 			rows.Close()
-			return errors.New("read quarantined task")
+			return nil, errors.New("read quarantined task")
 		}
 		invalid = append(invalid, invalidTask{taskID, accountID, confirmationID.String, state, reasonCode, confirmationVersion.Int64})
 	}
 	if err := rows.Close(); err != nil {
-		return errors.New("close quarantined task rows")
+		return nil, errors.New("close quarantined task rows")
 	}
+	transitions := make([]taskstate.Transition, 0, len(invalid))
 	for _, task := range invalid {
 		if err := appendAudit(ctx, tx, "task_quarantined", task.reasonCode, task.accountID, task.confirmationID, task.confirmationVersion, "", task.taskID, map[string]string{"state": task.state}); err != nil {
-			return err
+			return nil, err
 		}
+		transitions = append(transitions, taskstate.Transition{
+			State: task.state, ReasonCode: task.reasonCode, AccountID: task.accountID, TaskID: task.taskID,
+		})
 	}
-	return nil
+	return transitions, nil
+}
+
+func scanTransitions(rows *sql.Rows, failure string) ([]taskstate.Transition, error) {
+	defer rows.Close()
+	var transitions []taskstate.Transition
+	for rows.Next() {
+		var transition taskstate.Transition
+		if err := rows.Scan(&transition.State, &transition.ReasonCode, &transition.AccountID, &transition.TaskID); err != nil {
+			return nil, errors.New(failure)
+		}
+		transitions = append(transitions, transition)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, errors.New(failure)
+	}
+	return transitions, nil
 }
 
 func (s *Store) Audit(ctx context.Context) ([]AuditEvent, error) {
@@ -1785,6 +1832,14 @@ func commit(tx *sql.Tx) error {
 	return nil
 }
 
+func (s *Store) commitTransitions(tx *sql.Tx, transitions []taskstate.Transition) error {
+	if err := commit(tx); err != nil {
+		return err
+	}
+	s.notifyTransitions(transitions)
+	return nil
+}
+
 func rowError(err error) error {
 	if errors.Is(err, sql.ErrNoRows) {
 		return ErrNotFound

internal/phasea/store_test.go

@@ -12,6 +12,8 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
 )
 
 func TestNewAccountIDIsStoreValidAndUnique(t *testing.T) {
@@ -95,6 +97,25 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() { _ = store.Close() })
+	var notificationMu sync.Mutex
+	var notifications []taskstate.Transition
+	store.SetTaskNotifier(func(transition taskstate.Transition) {
+		notificationMu.Lock()
+		defer notificationMu.Unlock()
+		notifications = append(notifications, transition)
+	})
+	assertNotified := func(state, reason, accountID, taskID string) {
+		t.Helper()
+		notificationMu.Lock()
+		defer notificationMu.Unlock()
+		want := taskstate.Transition{State: state, ReasonCode: reason, AccountID: accountID, TaskID: taskID}
+		for _, notification := range notifications {
+			if notification == want {
+				return
+			}
+		}
+		t.Fatalf("missing task notification %+v in %+v", want, notifications)
+	}
 	applyHubMigrationsForPhaseATest(t, store)
 	if _, err := store.db.ExecContext(ctx, `
 		TRUNCATE audit_event, execution_attempt, operation_task, confirmation, content_draft,
@@ -299,6 +320,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 				t.Fatalf("invalid task was claimed: execution=%+v err=%v", execution, err)
 			}
 			assertTaskGate(t, store, taskID, test.wantState, test.wantReason)
+			assertNotified(test.wantState, test.wantReason, "gate-"+test.name, taskID)
 		})
 	}
 	if err := store.VerifyTask(ctx, "task-gate-exit-unhealthy", "not_executed"); err != nil {
@@ -345,6 +367,8 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 		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)
+	assertNotified("needs_confirmation", "confirmation_missing", "account-b", "task-30")
+	assertNotified("needs_confirmation", "account_version_changed", "account-b", "task-31")
 	assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id IN ('task-30', 'task-31')`, 0)
 	nullConfirmation, err := store.GetTaskDetail(ctx, "task-30")
 	if err != nil || nullConfirmation.Confirmation.ID != "" || nullConfirmation.AllowedAction != "" {
@@ -358,6 +382,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 	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)
 	}
+	assertNotified("needs_confirmation", "task_result_uncertain", "account-b", "task-32")
 	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)
 	}
@@ -370,6 +395,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 	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)
 	}
+	assertNotified("policy_hold", "task_policy_hold", "account-b", "task-33")
 	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)
 	}
@@ -378,6 +404,14 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 	if _, _, err := store.Enqueue(ctx, paused); err != nil {
 		t.Fatal(err)
 	}
+	pausedQueued := approvedTask(340, "account-b", accountB.Version, "draft-b", "confirmation-b")
+	pausedQueued.ID, pausedQueued.IdempotencyKey = "task-34-queued", "idempotency-34-queued"
+	if _, _, err := store.Enqueue(ctx, pausedQueued); err != nil {
+		t.Fatal(err)
+	}
+	if claimed, err := store.claim(ctx, "worker-account-pause"); err != nil || claimed.TaskID != paused.ID {
+		t.Fatalf("claim executing pause fixture: execution=%+v err=%v", claimed, err)
+	}
 	pauseStarted := time.Now()
 	if err := store.PauseAccount(ctx, "account-b"); err != nil {
 		t.Fatal(err)
@@ -388,7 +422,10 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 	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)
+	assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-34' AND state = 'needs_confirmation'`, 1)
+	assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-34-queued' AND state = 'policy_hold'`, 1)
+	assertNotified("needs_confirmation", "account_paused", "account-b", "task-34")
+	assertNotified("policy_hold", "account_paused", "account-b", "task-34-queued")
 	pausedAccount, err := store.GetAccount(ctx, "account-b")
 	if err != nil || pausedAccount.Version != accountB.Version+1 {
 		t.Fatalf("idempotent pause must increment once: %#v %v", pausedAccount, err)
@@ -398,6 +435,18 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 		t.Fatalf("account was not paused within one minute: paused_at=%v err=%v", pausedAt, err)
 	}
 
+	revokeQueued := approvedTask(341, "account-b", pausedAccount.Version, "draft-b", "confirmation-b")
+	revokeQueued.ID, revokeQueued.IdempotencyKey = "task-revoke-queued", "idempotency-revoke-queued"
+	revokeExecuting := approvedTask(342, "account-b", pausedAccount.Version, "draft-b", "confirmation-b")
+	revokeExecuting.ID, revokeExecuting.IdempotencyKey = "task-revoke-executing", "idempotency-revoke-executing"
+	for _, task := range []Task{revokeQueued, revokeExecuting} {
+		if _, _, err := store.Enqueue(ctx, task); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := store.db.ExecContext(ctx, `UPDATE operation_task SET state = 'executing', lease_owner = 'worker-revoke', lease_until = now() + interval '1 minute' WHERE id = $1`, revokeExecuting.ID); err != nil {
+		t.Fatal(err)
+	}
 	if err := store.RevokeAccount(ctx, "account-b"); err != nil {
 		t.Fatal(err)
 	}
@@ -408,6 +457,8 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 	if err != nil || revokedAccount.AuthorizationStatus != "revoked" || revokedAccount.Version != accountB.Version+2 {
 		t.Fatalf("unexpected revoked account: %#v %v", revokedAccount, err)
 	}
+	assertNotified("policy_hold", "account_revoked", "account-b", revokeQueued.ID)
+	assertNotified("needs_confirmation", "account_revoked", "account-b", revokeExecuting.ID)
 	if err := store.ResumeAccount(ctx, "account-b"); !errors.Is(err, ErrConflict) {
 		t.Fatalf("revoked account must not resume: %v", err)
 	}
@@ -436,6 +487,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 		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)
+	assertNotified("needs_confirmation", "execution_lease_expired", "account-a", "task-36")
 	assertCount(t, store, `SELECT count(*) FROM execution_attempt WHERE task_id = 'task-36' AND outcome = 'uncertain'`, 1)
 	if err := store.ResumeTask(ctx, "task-36"); !errors.Is(err, ErrConflict) {
 		t.Fatalf("unknown result resumed without verification: %v", err)
@@ -482,6 +534,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 		t.Fatalf("stale worker persisted a successful result: execution=%+v err=%v", completed, err)
 	}
 	assertCount(t, store, `SELECT count(*) FROM operation_task WHERE id = 'task-37' AND state = 'needs_confirmation' AND hold_reason = 'exit_unhealthy'`, 1)
+	assertNotified("needs_confirmation", "exit_unhealthy", "account-a", "task-37")
 	if _, err := store.db.ExecContext(ctx, `UPDATE network_exit SET health_status = 'healthy' WHERE id = 'exit-shared'`); err != nil {
 		t.Fatal(err)
 	}
@@ -532,6 +585,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 	}
 	assertCount(t, store, `SELECT count(*) FROM operation_task
 		WHERE id = 'task-38' AND state = 'needs_confirmation' AND hold_reason = 'runtime_missing'`, 1)
+	assertNotified("needs_confirmation", "runtime_missing", "account-a", "task-38")
 	concurrentDetail, err := store.GetTaskDetail(ctx, concurrent.ID)
 	if err != nil || concurrentDetail.RuntimeInstanceID != "runtime-instance-a" {
 		t.Fatalf("task detail lost its immutable claim runtime: detail=%+v err=%v", concurrentDetail, err)
@@ -562,6 +616,7 @@ func TestPhaseAOfflineWorkflow(t *testing.T) {
 	}
 	assertCount(t, store, `SELECT count(*) FROM operation_task
 		WHERE id = 'task-39' AND state = 'needs_confirmation' AND hold_reason = 'task_result_uncertain'`, 1)
+	assertNotified("needs_confirmation", "task_result_uncertain", "account-a", "task-39")
 	replacedDetail, err := store.GetTaskDetail(ctx, replaced.ID)
 	if err != nil || replacedDetail.AllowedAction != "verify" || replacedDetail.BrowserEnvAlias != "account-a" ||
 		replacedDetail.NetworkExitID != "exit-shared" || replacedDetail.RuntimeInstanceID != "runtime-instance-a2" || replacedDetail.BindingVersion != 1 {

internal/taskstate/transition.go

@@ -0,0 +1,7 @@
+package taskstate
+
+type Transition struct {
+	State, ReasonCode, AccountID, TaskID string
+}
+
+type Notifier func(Transition)